A2A Protocol

libs/agent_framework/pyproject.toml

@@ -27,6 +27,9 @@ dependencies = [
   "PyJWT[crypto]>=2.9.0"
 ]
 
+[project.optional-dependencies]
+a2a = ["a2a-sdk[http-server]==1.2.2"]
+
 [tool.setuptools.packages.find]
 where = ["src"]
 

libs/agent_framework/src/agent_framework/a2a/__init__.py

@@ -0,0 +1 @@
+"""Optional A2A interoperability; imported only when enabled."""

libs/agent_framework/src/agent_framework/a2a/client.py

@@ -0,0 +1,108 @@
+"""A2A JSON-RPC adapter. URLs and credential references are operator configured."""
+import asyncio
+import json
+import os
+from uuid import uuid4, uuid5, NAMESPACE_URL
+
+import httpx
+from google.protobuf.json_format import ParseDict, MessageToDict
+from a2a.types import SendMessageResponse, Task
+
+from .progress import report_progress
+
+
+def _normalize_legacy(result):
+    # Preserve the original wire response separately; normalize only presentation.
+    if result.get('kind') == 'message':
+        return {'message': result}
+    if result.get('kind') == 'task':
+        return {'task': result}
+    raise ValueError('Expected an A2A Message or Task')
+
+
+async def call_a2a(backend, request, timeout_seconds):
+    payload = request.get('payload') or {}
+    session = payload.get('session_id') or str(uuid4())
+    tenant = request.get('tenant_id') or 'default'
+    context_id = str(uuid5(NAMESPACE_URL, json.dumps([tenant, backend.backend_id, session])))
+    legacy = backend.a2a_version == '0.3'
+    text = str(payload.get('message') or payload.get('text') or payload.get('transcript') or payload.get('content') or '')
+    parts = [{'text': text}, {'data': payload}]
+    message = {'role': 'ROLE_USER', 'messageId': str(uuid4()),
+               'contextId': payload.get('a2a_context_id') or context_id, 'parts': parts,
+               'metadata': {'tenant_id': tenant, 'agent_id': request.get('agent_id') or backend.default_agent_id}}
+    if payload.get('a2a_task_id'):
+        message['taskId'] = payload['a2a_task_id']
+    if legacy:
+        message.update(kind='message', role='user')
+        for part in parts:
+            part['kind'] = 'text' if 'text' in part else 'data'
+    headers = {'A2A-Version': backend.a2a_version}
+    for header, env_name in backend.a2a_headers_env.items():
+        value = os.environ.get(env_name)
+        if not value:
+            raise ValueError(f'Missing credential environment variable: {env_name}')
+        headers[header] = value
+    endpoint = backend.base_url + backend.a2a_rpc_path
+
+    async with asyncio.timeout(timeout_seconds):
+        async with httpx.AsyncClient(timeout=timeout_seconds, headers=headers) as client:
+            async def rpc(method, params):
+                request_id = str(uuid4())
+                response = await client.post(endpoint, json={
+                    'jsonrpc': '2.0', 'id': request_id, 'method': method, 'params': params,
+                })
+                response.raise_for_status()
+                data = response.json()
+                if data.get('jsonrpc') != '2.0' or data.get('id') != request_id:
+                    raise ValueError('Invalid A2A JSON-RPC response')
+                if 'error' in data:
+                    raise RuntimeError(f'A2A error {data["error"].get("code")}')
+                result = data.get('result')
+                if not isinstance(result, dict):
+                    raise ValueError('Missing A2A result')
+                return result
+
+            raw = await rpc('message/send' if legacy else 'SendMessage', {
+                'message': message,
+                'configuration': {'blocking': False} if legacy else {'returnImmediately': True},
+            })
+            result = _normalize_legacy(raw) if legacy else raw
+            if not legacy:
+                result = MessageToDict(ParseDict(result, SendMessageResponse()))
+            while 'task' in result:
+                task = result['task']
+                state = task.get('status', {}).get('state', '')
+                await report_progress('a2a.backend.status', {
+                    'backend_id': backend.backend_id, 'task_id': task.get('id'), 'state': state,
+                })
+                if state not in {'submitted', 'working', 'TASK_STATE_SUBMITTED', 'TASK_STATE_WORKING'}:
+                    break
+                await asyncio.sleep(backend.a2a_poll_interval_seconds)
+                raw = await rpc('tasks/get' if legacy else 'GetTask', {'id': task['id']})
+                result = {'task': raw if legacy else MessageToDict(ParseDict(raw, Task()))}
+
+    task = result.get('task')
+    if task:
+        state, task_id = task['status']['state'], task['id']
+        if state in {'failed', 'canceled', 'rejected', 'unknown', 'TASK_STATE_FAILED', 'TASK_STATE_CANCELED', 'TASK_STATE_REJECTED', 'TASK_STATE_UNSPECIFIED'}:
+            raise RuntimeError(f'A2A task ended with state: {state}')
+        output_parts = [p for a in task.get('artifacts', []) for p in a.get('parts', [])]
+        if not output_parts and task['status'].get('message'):
+            output_parts += task['status']['message'].get('parts', [])
+        if not output_parts:
+            messages = [m for m in task.get('history', []) if m.get('role') in {'agent', 'ROLE_AGENT'}]
+            output_parts = messages[-1].get('parts', []) if messages else []
+        remote_context = task.get('contextId')
+    elif result.get('message'):
+        reply = result['message']
+        output_parts = reply.get('parts', [])
+        state, task_id, remote_context = None, None, reply.get('contextId')
+    else:
+        raise ValueError('Expected an A2A Message or Task')
+    return {
+        'channel': request.get('channel', 'web'), 'session_id': session,
+        'text': '\n'.join(p['text'] for p in output_parts if 'text' in p),
+        'metadata': {'a2a_context_id': remote_context, 'a2a_task_id': task_id,
+                     'a2a_state': state, 'a2a_result': raw},
+    }

libs/agent_framework/src/agent_framework/a2a/progress.py

@@ -0,0 +1,10 @@
+"""Task-local progress hook; safe to import without the optional A2A SDK."""
+from contextvars import ContextVar
+
+_progress_callback = ContextVar('framework_a2a_progress', default=None)
+
+
+async def report_progress(event: str, payload: dict | None = None):
+    callback = _progress_callback.get()
+    if callback is not None:
+        await callback(event, payload or {})

libs/agent_framework/src/agent_framework/a2a/server.py

@@ -0,0 +1,181 @@
+"""A2A 1.0 Tasks and SSE with SDK-managed 0.3 compatibility."""
+import asyncio
+import logging
+from uuid import uuid4
+
+from google.protobuf.json_format import MessageToDict, ParseDict
+from a2a.auth.user import User
+from a2a.server.agent_execution import AgentExecutor
+from a2a.server.routes import (
+    DefaultServerCallContextBuilder, add_a2a_routes_to_fastapi,
+    create_agent_card_routes, create_jsonrpc_routes,
+)
+from a2a.server.request_handlers import DefaultRequestHandler
+from a2a.server.tasks import InMemoryTaskStore, TaskUpdater
+from a2a.types import AgentCard, Message, Part, Role, Task, TaskState, TaskStatus
+from a2a.utils.errors import InvalidParamsError, TaskNotCancelableError, UnsupportedOperationError
+
+from .progress import _progress_callback
+
+logger = logging.getLogger(__name__)
+
+
+class _PrincipalUser(User):
+    def __init__(self, principal):
+        self.principal = principal
+
+    @property
+    def is_authenticated(self):
+        return True
+
+    @property
+    def user_name(self):
+        # Use middleware identity, never client-supplied message metadata.
+        return f'{self.principal.scheme}:{self.principal.subject}'
+
+
+class FrameworkCallContextBuilder(DefaultServerCallContextBuilder):
+    def build_user(self, request):
+        principal = getattr(request.state, 'auth_principal', None)
+        return _PrincipalUser(principal) if principal else super().build_user(request)
+
+
+def _response_message(response, context_id, task_id=None):
+    data = {
+        'messageId': str(uuid4()), 'role': 'ROLE_AGENT', 'contextId': context_id,
+        'parts': [{'text': str(response.get('text') or response.get('speak') or '')},
+                  {'data': response}],
+    }
+    if task_id:
+        data['taskId'] = task_id
+    return ParseDict(data, Message())
+
+
+class GatewayExecutor(AgentExecutor):
+    """Wrap any native async handler as an A2A task, without detached worker hacks."""
+
+    def __init__(self, handler, task_mode=True, timeout_seconds=120.0):
+        self.handler = handler
+        self.task_mode = task_mode
+        self.timeout_seconds = timeout_seconds
+
+    async def execute(self, context, event_queue):
+        message = context.message
+        if message is None or message.role != Role.ROLE_USER:
+            raise InvalidParamsError('Expected a user message')
+        payload = {}
+        for part in MessageToDict(message).get('parts', []):
+            if 'data' in part:
+                if not isinstance(part['data'], dict):
+                    raise InvalidParamsError('Structured data must be an object')
+                payload.update(part['data'])
+            elif 'text' not in part:
+                raise UnsupportedOperationError('Only text and structured data are supported')
+        metadata = MessageToDict(message.metadata)
+        session_id = context.context_id or str(uuid4())
+        payload.update(message=context.get_user_input(), session_id=session_id)
+        request = {
+            'channel': 'web', 'payload': payload, 'session_id': session_id,
+            'tenant_id': metadata.get('tenant_id'), 'agent_id': metadata.get('agent_id'),
+            'metadata': metadata, 'backend_id': metadata.get('backend_id'),
+        }
+        if not self.task_mode:
+            async with asyncio.timeout(self.timeout_seconds):
+                response = await self.handler(request)
+            await event_queue.enqueue_event(_response_message(response, session_id))
+            return
+
+        updater = TaskUpdater(event_queue, context.task_id, session_id)
+        if context.current_task is None:
+            await event_queue.enqueue_event(Task(
+                id=context.task_id, context_id=session_id,
+                status=TaskStatus(state=TaskState.TASK_STATE_SUBMITTED),
+                history=[message],
+            ))
+        await updater.update_status(TaskState.TASK_STATE_WORKING)
+
+        async def progress(event, data):
+            await updater.update_status(TaskState.TASK_STATE_WORKING, metadata={
+                'framework_event': event, 'framework_payload': data,
+            })
+
+        token = _progress_callback.set(progress)
+        try:
+            async with asyncio.timeout(self.timeout_seconds):
+                response = await self.handler(request)
+            reply = _response_message(response, session_id, context.task_id)
+            response_metadata = response.get('metadata') or {}
+            requested_state = response_metadata.get('a2a_state') or response_metadata.get('a2a_task_state')
+            state = {
+                'input-required': TaskState.TASK_STATE_INPUT_REQUIRED,
+                'auth-required': TaskState.TASK_STATE_AUTH_REQUIRED,
+                'TASK_STATE_INPUT_REQUIRED': TaskState.TASK_STATE_INPUT_REQUIRED,
+                'TASK_STATE_AUTH_REQUIRED': TaskState.TASK_STATE_AUTH_REQUIRED,
+            }.get(requested_state, TaskState.TASK_STATE_COMPLETED)
+            await updater.add_artifact(list(reply.parts), artifact_id='response', name='Framework response', last_chunk=True)
+            await updater.update_status(state, message=reply)
+        except asyncio.CancelledError:
+            raise
+        except Exception as exc:
+            logger.exception('A2A task %s failed', context.task_id)
+            text = 'Processing timed out' if isinstance(exc, TimeoutError) else 'Agent processing failed'
+            await updater.update_status(TaskState.TASK_STATE_FAILED, message=_response_message({'text': text}, session_id, context.task_id))
+        finally:
+            _progress_callback.reset(token)
+
+    async def cancel(self, context, event_queue):
+        if not self.task_mode:
+            raise TaskNotCancelableError()
+        # The SDK cancels the running execute coroutine after this callback.
+        await TaskUpdater(event_queue, context.task_id, context.context_id).update_status(TaskState.TASK_STATE_CANCELED)
+
+
+def install_a2a(app, handler, registry, public_url, security_schemes=None, security=None,
+                *, task_mode=True, timeout_seconds=120.0, task_store=None,
+                name='Agent Framework OCI Gateway', skills=None):
+    """Reuse parent middleware. A persistent SDK TaskStore may be injected."""
+    if timeout_seconds <= 0:
+        raise ValueError('A2A timeout_seconds must be positive')
+    if skills is None:
+        skills = [{'id': b.backend_id, 'name': b.name or b.backend_id,
+                   'description': b.description or b.backend_id,
+                   'tags': b.domains or [b.backend_id], 'examples': b.examples}
+                  for b in registry.list()]
+    card = ParseDict({
+        'name': name, 'version': '1.1.0',
+        'description': 'Native framework agents and A2A interoperability',
+        'supportedInterfaces': [{'url': public_url.rstrip('/') + '/a2a',
+                                 'protocolBinding': 'JSONRPC', 'protocolVersion': '1.0'}],
+        'capabilities': {'streaming': task_mode, 'pushNotifications': False},
+        'defaultInputModes': ['text/plain', 'application/json'],
+        'defaultOutputModes': ['text/plain', 'application/json'],
+        'securitySchemes': security_schemes or {},
+        'securityRequirements': [{'schemes': {k: {'list': v} for k, v in item.items()}}
+                                 for item in (security or [])],
+        'skills': skills,
+    }, AgentCard())
+    request_handler = DefaultRequestHandler(
+        GatewayExecutor(handler, task_mode=task_mode, timeout_seconds=timeout_seconds),
+        task_store if task_store is not None else InMemoryTaskStore(), card,
+    )
+    add_a2a_routes_to_fastapi(
+        app, agent_card_routes=create_agent_card_routes(card),
+        jsonrpc_routes=create_jsonrpc_routes(
+            request_handler, rpc_url='/a2a', enable_v0_3_compat=True,
+            context_builder=FrameworkCallContextBuilder(),
+        ),
+    )
+    return request_handler
+
+
+def install_backend_a2a(app, handler, profiles, settings, *, task_store=None):
+    """Expose a framework backend directly, with skills built from agent profiles."""
+    return install_a2a(
+        app, handler, None, settings.A2A_PUBLIC_URL,
+        settings.A2A_SECURITY_SCHEMES or None, settings.A2A_SECURITY or None,
+        task_mode=settings.A2A_TASK_MODE, timeout_seconds=settings.A2A_TASK_TIMEOUT_SECONDS,
+        task_store=task_store, name=settings.APP_NAME,
+        skills=[{'id': p.agent_id, 'name': p.name or p.agent_id,
+                 'description': p.description or p.agent_id, 'tags': [p.agent_id]}
+                for p in profiles.list_profiles()],
+    )

libs/agent_framework/src/agent_framework/config/settings.py

@@ -14,6 +14,13 @@ load_dotenv(override=False)
 class Settings(BaseSettings):
     model_config = SettingsConfigDict(env_file='.env', env_file_encoding='utf-8', extra='ignore')
 
+    ENABLE_A2A: bool = False
+    A2A_PUBLIC_URL: str = 'http://localhost:8000'
+    A2A_TASK_MODE: bool = True
+    A2A_TASK_TIMEOUT_SECONDS: float = Field(default=120.0, gt=0)
+    A2A_SECURITY_SCHEMES: dict = Field(default_factory=dict)
+    A2A_SECURITY: list[dict[str, list[str]]] = Field(default_factory=list)
+
     APP_NAME: str = 'ai-agent-template'
     APP_ENV: str = 'local'
     LOG_LEVEL: str = 'INFO'

libs/agent_framework/src/agent_framework/global_supervisor/client.py

@@ -1,6 +1,8 @@
 from __future__ import annotations
 
 import time
+import os
+from collections import OrderedDict
 from typing import Any
 
 import httpx
@@ -11,6 +13,8 @@ from .models import BackendCallResult, BackendDefinition, GlobalRouteDecision
 class BackendClient:
     def __init__(self, timeout_seconds: float = 120.0):
         self.timeout_seconds = timeout_seconds
+        # Bounded, process-local continuation cache; native calls do not use it.
+        self._a2a_sessions = OrderedDict()
 
     async def call_message(
         self,
@@ -19,6 +23,27 @@ class BackendClient:
         route_decision: GlobalRouteDecision,
         use_sse: bool = False,
     ) -> BackendCallResult:
+        if backend.protocol == "a2a":
+            from agent_framework.a2a.client import call_a2a
+            started = time.monotonic()
+            payload = dict(request_payload.get("payload") or {})
+            key = (request_payload.get("tenant_id") or "default", backend.backend_id, payload.get("session_id"))
+            previous = self._a2a_sessions.get(key) if key[2] else None
+            if previous and time.monotonic() - previous[0] < 3600:
+                payload.setdefault("a2a_context_id", previous[1].get("a2a_context_id"))
+                if previous[1].get("a2a_state") in {"input-required", "auth-required", "TASK_STATE_INPUT_REQUIRED", "TASK_STATE_AUTH_REQUIRED"}:
+                    payload.setdefault("a2a_task_id", previous[1].get("a2a_task_id"))
+            data = await call_a2a(backend, {**request_payload, "payload": payload}, self.timeout_seconds)
+            if key[2]:
+                self._a2a_sessions[key] = (time.monotonic(), data.get("metadata", {}))
+                self._a2a_sessions.move_to_end(key)
+                while len(self._a2a_sessions) > 10000:
+                    self._a2a_sessions.popitem(last=False)
+            return BackendCallResult(
+                backend_id=backend.backend_id, backend_url=backend.base_url,
+                status_code=200, response=data, route_decision=route_decision,
+                elapsed_ms=int((time.monotonic() - started) * 1000),
+            )
         path = backend.sse_message_path if use_sse else backend.message_path
         url = f"{backend.base_url}{path}"
         payload = dict(request_payload)
@@ -45,8 +70,16 @@ class BackendClient:
         )
 
     async def health(self, backend: BackendDefinition) -> dict[str, Any]:
-        url = f"{backend.base_url}{backend.health_path}"
-        async with httpx.AsyncClient(timeout=10.0) as client:
+        path = backend.a2a_agent_card_path if backend.protocol == "a2a" else backend.health_path
+        url = f"{backend.base_url}{path}"
+        headers = {}
+        if backend.protocol == "a2a":
+            for header, env_name in backend.a2a_headers_env.items():
+                value = os.environ.get(env_name)
+                if not value:
+                    return {"backend_id": backend.backend_id, "ok": False, "error": f"Missing credential environment variable: {env_name}"}
+                headers[header] = value
+        async with httpx.AsyncClient(timeout=10.0, headers=headers) as client:
             try:
                 resp = await client.get(url)
                 return {"backend_id": backend.backend_id, "status_code": resp.status_code, "ok": resp.is_success, "body": self._safe_json(resp)}

libs/agent_framework/src/agent_framework/global_supervisor/models.py

@@ -21,6 +21,12 @@ class BackendDefinition(BaseModel):
     examples: list[str] = Field(default_factory=list)
     priority: int = 100
     enabled: bool = True
+    protocol: Literal["native", "a2a"] = "native"
+    a2a_version: Literal["1.0", "0.3"] = "1.0"
+    a2a_agent_card_path: str = "/.well-known/agent-card.json"
+    a2a_rpc_path: str = "/a2a"
+    a2a_headers_env: dict[str, str] = Field(default_factory=dict)
+    a2a_poll_interval_seconds: float = Field(default=0.5, gt=0)
     health_path: str = "/health"
     message_path: str = "/gateway/message"
     sse_message_path: str = "/gateway/message/sse"