From 34922720cdcc4356463e4a2fdd38bc83af77faf0 Mon Sep 17 00:00:00 2001 From: Luke Galea Date: Sun, 23 Aug 2026 16:12:41 +0000 Subject: [PATCH 1/3] feat(agent-server): add A2A server-mode support (agent card + JSON-RPC 2.0 endpoint) Expose the agent-server as an A2A agent (Linux Foundation a2a-spec, rev ~0.3, JSON-RPC transport), addressing OpenHands/software-agent-sdk#1060. - /.well-known/agent-card.json: AgentCard v0.3 discovery document (no auth) - POST /api/a2a: JSON-RPC 2.0 methods message/send, message/stream (SSE), tasks/get, tasks/cancel; taskId maps to conversationId - Auth accepts Authorization: Bearer or X-Session-API-Key - Minimal local pydantic models; no a2a-sdk dependency, no new deps - Tests, example script, additive api.py wiring only --- .../17_a2a_agent_card.py | 82 ++ .../openhands/agent_server/a2a_router.py | 753 ++++++++++++++++++ .../openhands/agent_server/api.py | 12 + tests/agent_server/test_a2a_router.py | 540 +++++++++++++ 4 files changed, 1387 insertions(+) create mode 100644 examples/02_remote_agent_server/17_a2a_agent_card.py create mode 100644 openhands-agent-server/openhands/agent_server/a2a_router.py create mode 100644 tests/agent_server/test_a2a_router.py diff --git a/examples/02_remote_agent_server/17_a2a_agent_card.py b/examples/02_remote_agent_server/17_a2a_agent_card.py new file mode 100644 index 0000000000..c4f7c85924 --- /dev/null +++ b/examples/02_remote_agent_server/17_a2a_agent_card.py @@ -0,0 +1,82 @@ +"""A2A example: talk to the agent-server as an A2A agent. + +The agent-server exposes an A2A (Agent2Agent) JSON-RPC 2.0 endpoint at +``/api/a2a`` and a discovery document at ``/.well-known/agent-card.json``. +This example uses plain httpx (no a2a-sdk dependency): + +1. Fetch the agent card from the well-known URI. +2. Send a user message with the ``message/send`` JSON-RPC method and print + the resulting Task (status + artifact). + +Usage: + python 17_a2a_agent_card.py [base_url] [session_api_key] + + # start a server first, e.g.: + # uv run python -m openhands.agent_server.base --port 9000 \ + # --session-api-key my-secret + python 17_a2a_agent_card.py http://localhost:9000 my-secret + +Requires an agent profile to be configured (the default profile is used +automatically). +""" + +import json +import sys + +import httpx + + +DEFAULT_BASE_URL = "http://localhost:9000" + + +def main() -> None: + base_url = sys.argv[1] if len(sys.argv) > 1 else DEFAULT_BASE_URL + api_key = sys.argv[2] if len(sys.argv) > 2 else None + + headers = {"Content-Type": "application/json"} + if api_key: + # A2A convention is Authorization: Bearer; the server also accepts + # the native X-Session-API-Key header. + headers["Authorization"] = f"Bearer {api_key}" + + with httpx.Client(timeout=120) as client: + # 1. Agent card discovery (no auth required). + card = client.get(f"{base_url}/.well-known/agent-card.json").json() + print("=== Agent Card ===") + print(json.dumps(card, indent=2)) + + # 2. message/send over JSON-RPC 2.0. + payload = { + "jsonrpc": "2.0", + "id": "a2a-example-1", + "method": "message/send", + "params": { + "message": { + "role": "user", + "parts": [{"kind": "text", "text": "Say hello from A2A!"}], + } + }, + } + print("\n=== message/send ===") + response = client.post(f"{base_url}/api/a2a", headers=headers, json=payload) + print(json.dumps(response.json(), indent=2)) + + # 3. Poll the task afterwards (taskId == conversationId). + task_id = response.json().get("result", {}).get("id") + if task_id: + print("\n=== tasks/get ===") + poll = client.post( + f"{base_url}/api/a2a", + headers=headers, + json={ + "jsonrpc": "2.0", + "id": "a2a-example-2", + "method": "tasks/get", + "params": {"id": task_id}, + }, + ) + print(json.dumps(poll.json(), indent=2)) + + +if __name__ == "__main__": + main() diff --git a/openhands-agent-server/openhands/agent_server/a2a_router.py b/openhands-agent-server/openhands/agent_server/a2a_router.py new file mode 100644 index 0000000000..c4e3397ee3 --- /dev/null +++ b/openhands-agent-server/openhands/agent_server/a2a_router.py @@ -0,0 +1,753 @@ +"""A2A (Agent2Agent) protocol server-mode router. + +Exposes the agent-server as an A2A agent (Google's Agent2Agent protocol, now +the Linux Foundation ``a2a-spec``) over the JSON-RPC 2.0 transport, following +the same pattern the SDK already uses for ACP. + +This module targets A2A spec rev ~0.3 (JSON-RPC transport, AgentCard v0.3). +It intentionally implements minimal local pydantic object models instead of +depending on the ``a2a-sdk`` package, so it stays dependency-free. Only the +subset of the spec needed for a server-mode agent is modelled: + +- ``GET /.well-known/agent-card.json`` — AgentCard v0.3 discovery document + (mounted at the app root: well-known URIs must live outside any API prefix). +- ``POST /api/a2a`` — JSON-RPC 2.0 endpoint with methods: + ``message/send``, ``message/stream`` (SSE), ``tasks/get``, ``tasks/cancel``. + +Mapping to agent-server concepts: + +- A2A task == OpenHands conversation; ``taskId`` is the conversationId. +- New task → ``conversation_service.start_conversation`` (resolved from the + active agent profile, or an explicit ``agentProfileId`` param) + + ``event_service.send_message(Message(role="user", ...), run=True)``. +- ``tasks/get`` → conversation execution status + ``get_agent_final_response`` + as a text artifact. +- ``tasks/cancel`` → ``conversation_service.interrupt_conversation``. + +Auth accepts either the agent-server's usual ``X-Session-API-Key`` header or +the A2A-conventional ``Authorization: Bearer `` header, both validated +against ``config.session_api_keys``. +""" + +from __future__ import annotations + +import asyncio +import json +import logging +import uuid +from collections.abc import AsyncIterator +from importlib.metadata import PackageNotFoundError, version +from typing import Any, Literal + +from fastapi import APIRouter, Depends, Header, HTTPException, Request, status +from fastapi.responses import StreamingResponse +from pydantic import BaseModel, Field, ValidationError + +from openhands.agent_server.config import Config +from openhands.agent_server.conversation_service import ConversationService +from openhands.agent_server.dependencies import get_conversation_service +from openhands.agent_server.event_service import EventService +from openhands.agent_server.init_router import require_initialized +from openhands.agent_server.models import StartConversationRequest +from openhands.agent_server.pub_sub import Subscriber +from openhands.agent_server.utils import utc_now +from openhands.sdk.event.conversation_state import ConversationStateUpdateEvent +from openhands.sdk.llm.message import Message, TextContent +from openhands.sdk.workspace import LocalWorkspace + + +logger = logging.getLogger(__name__) + +A2A_PROTOCOL_VERSION = "0.3.0" +A2A_MEDIA_TYPE = "text/event-stream" + +# JSON-RPC 2.0 error codes (plus the A2A-style task-not-found extension). +JSONRPC_PARSE_ERROR = -32700 +JSONRPC_INVALID_REQUEST = -32600 +JSONRPC_METHOD_NOT_FOUND = -32601 +JSONRPC_INVALID_PARAMS = -32602 +JSONRPC_INTERNAL_ERROR = -32603 +JSONRPC_TASK_NOT_FOUND = -32001 + +# ConversationExecutionStatus values that end an A2A task. IDLE means the +# conversation finished its run and is ready for new input; FINISHED is the +# explicit terminal state. ERROR / STUCK / DELETING are terminal failures. +_TERMINAL_EXECUTION_STATUSES = frozenset( + {"idle", "finished", "error", "stuck", "deleting"} +) + +_TASK_STATE_LITERAL = Literal[ + "submitted", + "working", + "input-required", + "completed", + "canceled", + "failed", + "rejected", + "unknown", +] + +_EXECUTION_STATUS_TO_TASK_STATE: dict[str, str] = { + "idle": "completed", + "finished": "completed", + "running": "working", + "paused": "input-required", + "waiting_for_confirmation": "input-required", + "error": "failed", + "stuck": "failed", + "deleting": "failed", +} + + +# --------------------------------------------------------------------------- +# Minimal local A2A object models (no a2a-sdk dependency). +# --------------------------------------------------------------------------- + + +class AgentProvider(BaseModel): + """A2A AgentCard ``provider`` block.""" + + organization: str = "OpenHands" + url: str = "https://github.com/OpenHands/software-agent-sdk" + + +class AgentCapabilities(BaseModel): + """A2A AgentCard ``capabilities`` block.""" + + streaming: bool = True + pushNotifications: bool = False # noqa: N815 - A2A field name + + +class AgentSkill(BaseModel): + """A2A AgentCard skill entry (one per agent profile).""" + + id: str + name: str + description: str + tags: list[str] = Field(default_factory=list) + + +class AgentCard(BaseModel): + """A2A AgentCard v0.3 discovery document.""" + + name: str + description: str + url: str + version: str + protocolVersion: str = A2A_PROTOCOL_VERSION # noqa: N815 - A2A field name + preferredTransport: str = "JSONRPC" # noqa: N815 - A2A field name + capabilities: AgentCapabilities = Field(default_factory=AgentCapabilities) + defaultInputModes: list[str] = Field(default_factory=lambda: ["text/plain"]) # noqa: N815 + defaultOutputModes: list[str] = Field(default_factory=lambda: ["text/plain"]) # noqa: N815 + skills: list[AgentSkill] = Field(default_factory=list) + provider: AgentProvider = Field(default_factory=AgentProvider) + + +class TextPart(BaseModel): + """A2A text content part.""" + + kind: Literal["text"] = "text" + text: str + + +class A2AMessage(BaseModel): + """A2A message (subset: text parts only).""" + + role: Literal["user", "agent"] + parts: list[TextPart] = Field(default_factory=list) + messageId: str | None = None # noqa: N815 - A2A field name + taskId: str | None = None # noqa: N815 - A2A field name + contextId: str | None = None # noqa: N815 - A2A field name + + +class MessageSendParams(BaseModel): + """Params for ``message/send`` / ``message/stream``. + + ``agentProfileId`` is an OpenHands extension: selects the agent profile to + launch the conversation from. When omitted, the server's active agent + profile (or the first stored profile) is used. + """ + + message: A2AMessage + agentProfileId: str | None = None # noqa: N815 - extension field + + +class TaskGetParams(BaseModel): + """Params for ``tasks/get``.""" + + id: str + + +class TaskCancelParams(BaseModel): + """Params for ``tasks/cancel``.""" + + id: str + + +TaskState = _TASK_STATE_LITERAL + + +class TaskStatus(BaseModel): + """A2A task status.""" + + state: TaskState + message: A2AMessage | None = None + timestamp: str = Field(default_factory=lambda: utc_now().isoformat()) + + +class Artifact(BaseModel): + """A2A task artifact (the agent's final response text).""" + + artifactId: str # noqa: N815 - A2A field name + name: str = "response" + parts: list[TextPart] + + +class Task(BaseModel): + """A2A task — maps 1:1 onto an OpenHands conversation.""" + + id: str + contextId: str # noqa: N815 - A2A field name + status: TaskStatus + artifacts: list[Artifact] = Field(default_factory=list) + kind: Literal["task"] = "task" + + +class TaskStatusUpdateEvent(BaseModel): + """A2A streaming status-update event.""" + + taskId: str # noqa: N815 - A2A field name + contextId: str # noqa: N815 - A2A field name + status: TaskStatus + final: bool = False + kind: Literal["status-update"] = "status-update" + + +class TaskArtifactUpdateEvent(BaseModel): + """A2A streaming artifact-update event.""" + + taskId: str # noqa: N815 - A2A field name + contextId: str # noqa: N815 - A2A field name + artifact: Artifact + lastChunk: bool = True # noqa: N815 - A2A field name + kind: Literal["artifact-update"] = "artifact-update" + + +class JSONRPCErrorObject(BaseModel): + """JSON-RPC 2.0 error object.""" + + code: int + message: str + data: str | None = None + + +class JSONRPCResponse(BaseModel): + """JSON-RPC 2.0 response envelope for the A2A endpoint.""" + + jsonrpc: Literal["2.0"] = "2.0" + id: str | int | None = None + result: Task | TaskStatusUpdateEvent | TaskArtifactUpdateEvent | None = None + error: JSONRPCErrorObject | None = None + + +# --------------------------------------------------------------------------- +# Auth +# --------------------------------------------------------------------------- + + +async def check_a2a_session_api_key( + request: Request, + authorization: str | None = Header(default=None), +) -> None: + """A2A auth: accept either ``X-Session-API-Key`` or ``Authorization: Bearer``. + + A2A clients conventionally send ``Authorization: Bearer ``; the + agent-server's own clients send ``X-Session-API-Key``. Both are validated + against ``config.session_api_keys`` (empty list disables auth, same as the + built-in dependency). + """ + config: Config | None = getattr(request.app.state, "config", None) + if config is None: + raise HTTPException( + status_code=status.HTTP_503_SERVICE_UNAVAILABLE, + detail="Server not fully initialized", + ) + if not config.session_api_keys: + return + header_key = request.headers.get("X-Session-API-Key") + bearer_key = None + if authorization and authorization.startswith("Bearer "): + bearer_key = authorization[len("Bearer ") :].strip() + for candidate in (header_key, bearer_key): + if candidate and candidate in config.session_api_keys: + return + raise HTTPException(status.HTTP_401_UNAUTHORIZED) + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _server_version() -> str: + try: + return version("openhands-agent-server") + except PackageNotFoundError: + return "dev" + + +def _agent_card_skills() -> list[AgentSkill]: + """List stored agent profiles as A2A skills (best-effort, never raises).""" + try: + from openhands.agent_server.persistence import get_agent_profile_store + + store = get_agent_profile_store() + skills = [] + for summary in store.list_summaries(): + name = str(summary.get("name", "profile")) + skill_id = str(summary.get("id") or name) + skills.append( + AgentSkill( + id=skill_id, + name=name, + description=f"OpenHands agent profile '{name}'", + tags=[str(summary.get("agent_kind", "openhands"))], + ) + ) + return skills + except Exception: + logger.debug("Could not list agent profiles for the A2A card", exc_info=True) + return [] + + +def _resolve_agent_profile_id() -> str | None: + """Resolve the agent profile a new A2A conversation should launch from. + + Prefers the active profile pointer from persisted settings, then falls + back to the first stored profile. Best-effort: returns None when no + profile is available. + """ + try: + from openhands.agent_server.persistence import ( + get_agent_profile_store, + get_settings_store, + ) + + settings = get_settings_store().load() + if settings is not None and settings.active_agent_profile_id is not None: + return str(settings.active_agent_profile_id) + for summary in get_agent_profile_store().list_summaries(): + sid = summary.get("id") + if sid is not None: + return str(sid) + except Exception: + logger.debug("Could not resolve an agent profile for A2A", exc_info=True) + return None + + +def _parse_task_id(raw: str) -> uuid.UUID: + try: + return uuid.UUID(raw) + except ValueError: + raise ValueError(f"Invalid taskId (not a conversationId): {raw}") from None + + +def _task_state_for_execution_status(execution_status: Any) -> TaskState: + raw = str(getattr(execution_status, "value", execution_status) or "") + return _EXECUTION_STATUS_TO_TASK_STATE.get(raw, "unknown") # type: ignore[return-value] + + +async def _get_event_service_for( + conversation_service: ConversationService, task_id: uuid.UUID +) -> EventService | None: + return await conversation_service.get_event_service(task_id) + + +async def _task_from_event_service( + task_id: uuid.UUID, event_service: EventService +) -> Task: + """Build an A2A Task snapshot from a conversation's event service.""" + execution_status = None + final_response = "" + try: + state = await event_service.get_state() + execution_status = getattr(state, "execution_status", None) + except Exception: + logger.debug("A2A: could not read conversation state", exc_info=True) + try: + final_response = await event_service.get_agent_final_response() + except Exception: + logger.debug("A2A: could not read agent final response", exc_info=True) + + artifacts: list[Artifact] = [] + if final_response: + artifacts.append( + Artifact( + artifactId=str(uuid.uuid4()), + parts=[TextPart(text=final_response)], + ) + ) + return Task( + id=str(task_id), + contextId=str(task_id), + status=TaskStatus(state=_task_state_for_execution_status(execution_status)), + artifacts=artifacts, + ) + + +def _jsonrpc(result: Any, rpc_id: str | int | None) -> JSONRPCResponse: + return JSONRPCResponse(jsonrpc="2.0", id=rpc_id, result=result) + + +def _jsonrpc_error( + code: int, message: str, rpc_id: str | int | None = None, data: str | None = None +) -> JSONRPCResponse: + return JSONRPCResponse( + jsonrpc="2.0", + id=rpc_id, + error=JSONRPCErrorObject(code=code, message=message, data=data), + ) + + +def _user_text(message: A2AMessage) -> str: + return "".join(part.text for part in message.parts if part.kind == "text") + + +class _QueueSubscriber(Subscriber): + """Bridges agent-server events into an asyncio queue for the SSE stream.""" + + def __init__(self, queue: asyncio.Queue): + self._queue = queue + + async def __call__(self, event: Any): + await self._queue.put(event) + + +# --------------------------------------------------------------------------- +# Routers +# --------------------------------------------------------------------------- + +# Mounted at the app ROOT (outside /api): well-known URIs must be discoverable +# at /.well-known/ regardless of API prefixes. +a2a_agent_card_router = APIRouter(tags=["A2A"]) + +# Mounted under api_router (prefix /api) → served at /api/a2a. Uses its own +# auth dependency (Bearer OR X-Session-API-Key) plus the standard dormant +# gate; this deliberately does not go through the shared api_router +# dependency group, which only honors the X-Session-API-Key header. +a2a_router = APIRouter( + prefix="/a2a", + tags=["A2A"], + dependencies=[ + Depends(check_a2a_session_api_key), + Depends(require_initialized), + ], +) + + +@a2a_agent_card_router.get( + "/.well-known/agent-card.json", + response_model=AgentCard, + response_model_exclude_none=True, +) +async def get_agent_card(request: Request) -> AgentCard: + """Return the A2A AgentCard v0.3 discovery document.""" + base_url = str(request.base_url).rstrip("/") + return AgentCard( + name="OpenHands Agent Server", + description=( + "OpenHands software-agent SDK agent server, exposed as an A2A " + "agent. Each A2A task maps to one OpenHands conversation running " + "the server's configured agent profile." + ), + url=f"{base_url}/api/a2a", + version=_server_version(), + capabilities=AgentCapabilities(streaming=True, pushNotifications=False), + skills=_agent_card_skills(), + ) + + +@a2a_router.post("", response_model=JSONRPCResponse, response_model_exclude_none=True) +async def a2a_jsonrpc_endpoint( + request: Request, + conversation_service: ConversationService = Depends(get_conversation_service), +) -> Any: + """JSON-RPC 2.0 endpoint for A2A methods. + + Supported methods: ``message/send``, ``message/stream`` (returns an SSE + stream rather than a JSON envelope), ``tasks/get``, ``tasks/cancel``. + JSON-RPC-level errors are returned as 200 responses carrying a JSON-RPC + error object (``-32601`` method not found, ``-32700`` parse error, + ``-32602`` invalid params). + """ + raw = await request.body() + try: + payload = json.loads(raw) + except (json.JSONDecodeError, UnicodeDecodeError): + return _jsonrpc_error(JSONRPC_PARSE_ERROR, "Parse error") + + rpc_id = payload.get("id") if isinstance(payload, dict) else None + if not isinstance(payload, dict) or payload.get("jsonrpc") != "2.0": + return _jsonrpc_error(JSONRPC_INVALID_REQUEST, "Invalid Request", rpc_id) + method = payload.get("method") + params = payload.get("params") or {} + config: Config = request.app.state.config + + if not isinstance(method, str): + return _jsonrpc_error( + JSONRPC_INVALID_REQUEST, "Invalid Request: missing method", rpc_id + ) + + if method == "message/send": + try: + send_params = MessageSendParams.model_validate(params) + except ValidationError as exc: + return _jsonrpc_error( + JSONRPC_INVALID_PARAMS, "Invalid params", rpc_id, data=str(exc) + ) + handle_result = await _handle_message_send( + send_params, conversation_service, config + ) + if isinstance(handle_result, JSONRPCResponse): + return handle_result + return _jsonrpc(handle_result, rpc_id) + if method == "message/stream": + try: + send_params = MessageSendParams.model_validate(params) + except ValidationError as exc: + return _jsonrpc_error( + JSONRPC_INVALID_PARAMS, "Invalid params", rpc_id, data=str(exc) + ) + return await _handle_message_stream( + send_params, conversation_service, rpc_id, config + ) + + if method == "tasks/get": + try: + get_params = TaskGetParams.model_validate(params) + task_id = _parse_task_id(get_params.id) + except (ValidationError, ValueError) as exc: + return _jsonrpc_error( + JSONRPC_INVALID_PARAMS, "Invalid params", rpc_id, data=str(exc) + ) + event_service = await _get_event_service_for(conversation_service, task_id) + if event_service is None: + return _jsonrpc_error( + JSONRPC_TASK_NOT_FOUND, f"Task not found: {get_params.id}", rpc_id + ) + return _jsonrpc(await _task_from_event_service(task_id, event_service), rpc_id) + + if method == "tasks/cancel": + try: + cancel_params = TaskCancelParams.model_validate(params) + task_id = _parse_task_id(cancel_params.id) + except (ValidationError, ValueError) as exc: + return _jsonrpc_error( + JSONRPC_INVALID_PARAMS, "Invalid params", rpc_id, data=str(exc) + ) + try: + cancelled = await conversation_service.interrupt_conversation(task_id) + except ValueError: + cancelled = False + if not cancelled: + return _jsonrpc_error( + JSONRPC_TASK_NOT_FOUND, + f"Task not found: {cancel_params.id}", + rpc_id, + ) + return _jsonrpc( + Task( + id=cancel_params.id, + contextId=cancel_params.id, + status=TaskStatus(state="canceled"), + ), + rpc_id, + ) + + return _jsonrpc_error( + JSONRPC_METHOD_NOT_FOUND, f"Method not found: {method}", rpc_id + ) + + +async def _start_or_get_conversation( + send_params: MessageSendParams, + conversation_service: ConversationService, + config: Config, +) -> tuple[uuid.UUID, EventService, bool] | JSONRPCResponse: + """Return ``(task_id, event_service, created)`` or a JSON-RPC error. + + Reuses the conversation named by ``message.taskId`` when present, + otherwise starts a new one from the resolved agent profile. + """ + if send_params.message.taskId: + try: + task_id = _parse_task_id(send_params.message.taskId) + except ValueError as exc: + return _jsonrpc_error(JSONRPC_INVALID_PARAMS, str(exc)) + event_service = await _get_event_service_for(conversation_service, task_id) + if event_service is None: + return _jsonrpc_error( + JSONRPC_TASK_NOT_FOUND, + f"Task not found: {send_params.message.taskId}", + ) + return task_id, event_service, False + + profile_id = send_params.agentProfileId or _resolve_agent_profile_id() + if profile_id is None: + return _jsonrpc_error( + JSONRPC_INTERNAL_ERROR, + "No agent profile configured; create one via /api/agent-profiles " + "or pass agentProfileId in the params", + ) + try: + start_request = StartConversationRequest( + agent_profile_id=uuid.UUID(profile_id), + workspace=LocalWorkspace(working_dir=str(config.workspace_path)), + ) + info, _created = await conversation_service.start_conversation(start_request) + except ValueError as exc: + return _jsonrpc_error( + JSONRPC_INVALID_PARAMS, f"Could not start conversation: {exc}" + ) + event_service = await _get_event_service_for(conversation_service, info.id) + if event_service is None: + return _jsonrpc_error( + JSONRPC_INTERNAL_ERROR, f"Conversation {info.id} is not available" + ) + return info.id, event_service, True + + +async def _send_user_message( + send_params: MessageSendParams, event_service: EventService +) -> None: + text = _user_text(send_params.message) + message = Message(role="user", content=[TextContent(text=text)]) + await event_service.send_message(message, run=True) + + +async def _handle_message_send( + send_params: MessageSendParams, + conversation_service: ConversationService, + config: Config, +) -> Task | JSONRPCResponse: + started = await _start_or_get_conversation( + send_params, conversation_service, config + ) + if isinstance(started, JSONRPCResponse): + return started + task_id, event_service, _created = started + await _send_user_message(send_params, event_service) + return await _task_from_event_service(task_id, event_service) + + +def _sse_chunk(payload: JSONRPCResponse) -> str: + return ( + "data: " + + json.dumps( + payload.model_dump(exclude_none=True, mode="json"), separators=(",", ":") + ) + + "\r\n\r\n" + ) + + +def _state_update_to_status_update( + task_id: uuid.UUID, event: ConversationStateUpdateEvent, final: bool +) -> TaskStatusUpdateEvent | None: + if getattr(event, "key", None) != "execution_status": + return None + return TaskStatusUpdateEvent( + taskId=str(task_id), + contextId=str(task_id), + status=TaskStatus(state=_task_state_for_execution_status(event.value)), + final=final, + ) + + +async def _handle_message_stream( + send_params: MessageSendParams, + conversation_service: ConversationService, + rpc_id: str | int | None, + config: Config | None = None, +) -> Any: + """``message/stream``: run the task and stream A2A updates over SSE.""" + started = await _start_or_get_conversation( + send_params, conversation_service, config or Config() + ) + if isinstance(started, JSONRPCResponse): + return started + task_id, event_service, _created = started + + async def event_stream() -> AsyncIterator[str]: + queue: asyncio.Queue = asyncio.Queue() + subscriber = _QueueSubscriber(queue) + subscriber_id = await event_service.subscribe_to_events(subscriber) + try: + # Initial task snapshot; subsequent state updates arrive via the + # subscription as TaskStatusUpdateEvents. + yield _sse_chunk( + _jsonrpc( + Task( + id=str(task_id), + contextId=str(task_id), + status=TaskStatus(state="submitted"), + ), + rpc_id, + ) + ) + await _send_user_message(send_params, event_service) + + while True: + event = await queue.get() + if isinstance(event, ConversationStateUpdateEvent): + status_value = str(getattr(event, "value", "") or "") + is_terminal = status_value in _TERMINAL_EXECUTION_STATUSES + update = _state_update_to_status_update( + task_id, event, final=is_terminal + ) + if update is not None: + yield _sse_chunk(_jsonrpc(update, rpc_id)) + if is_terminal: + task = await _task_from_event_service(task_id, event_service) + for artifact in task.artifacts: + yield _sse_chunk( + _jsonrpc( + TaskArtifactUpdateEvent( + taskId=str(task_id), + contextId=str(task_id), + artifact=artifact, + lastChunk=True, + ), + rpc_id, + ) + ) + yield _sse_chunk(_jsonrpc(task, rpc_id)) + return + else: + # Surface agent message events as incremental artifacts. + llm_message = getattr(event, "llm_message", None) + if llm_message is not None and getattr( + event, "source", None + ) == "agent": + text = "".join( + getattr(part, "text", "") + for part in getattr(llm_message, "content", []) + ) + if text: + yield _sse_chunk( + _jsonrpc( + TaskArtifactUpdateEvent( + taskId=str(task_id), + contextId=str(task_id), + artifact=Artifact( + artifactId=str(uuid.uuid4()), + parts=[TextPart(text=text)], + ), + lastChunk=False, + ), + rpc_id, + ) + ) + finally: + await event_service.unsubscribe_from_events(subscriber_id) + + return StreamingResponse(event_stream(), media_type=A2A_MEDIA_TYPE) diff --git a/openhands-agent-server/openhands/agent_server/api.py b/openhands-agent-server/openhands/agent_server/api.py index 03871848db..7088cb2004 100644 --- a/openhands-agent-server/openhands/agent_server/api.py +++ b/openhands-agent-server/openhands/agent_server/api.py @@ -17,6 +17,10 @@ from fastapi.staticfiles import StaticFiles from starlette.requests import Request +from openhands.agent_server.a2a_router import ( + a2a_agent_card_router, + a2a_router, +) from openhands.agent_server.agent_profiles_router import agent_profiles_router from openhands.agent_server.auth_router import auth_router from openhands.agent_server.bash_router import bash_router @@ -452,8 +456,16 @@ def _add_api_routes(app: FastAPI) -> None: # /api/auth/* mints workspace cookies and requires the header to bootstrap, # so it lives under the header-only auth group. api_router.include_router(auth_router) + # A2A JSON-RPC endpoint. Mounted with its own dependencies (not the + # header-only group above) because A2A clients authenticate with the + # standard Authorization header rather than ``X-Session-API-Key``; the + # A2A auth dependency accepts either. + api_router.include_router(a2a_router) app.include_router(api_router) + # A2A discovery: well-known URIs must live at the app root, outside /api. + app.include_router(a2a_agent_card_router) + app.include_router(openai_router, dependencies=[Depends(check_openai_api_key)]) # Workspace static-file routes get their own auth group that accepts diff --git a/tests/agent_server/test_a2a_router.py b/tests/agent_server/test_a2a_router.py new file mode 100644 index 0000000000..be495a2730 --- /dev/null +++ b/tests/agent_server/test_a2a_router.py @@ -0,0 +1,540 @@ +"""Tests for a2a_router.py (A2A server-mode support). + +Builds the A2A routers the same way test_conversation_router.py builds the +conversation router: a bare FastAPI app with a mocked ConversationService / +EventService injected via dependency_overrides. +""" + +import json +from unittest.mock import AsyncMock, MagicMock +from uuid import uuid4 + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from openhands.agent_server.a2a_router import ( + a2a_agent_card_router, + a2a_router, + get_conversation_service, +) +from openhands.agent_server.config import Config +from openhands.agent_server.conversation_service import ConversationService +from openhands.agent_server.event_service import EventService +from openhands.agent_server.models import ConversationInfo +from openhands.agent_server.utils import utc_now +from openhands.sdk import LLM, Agent, Tool +from openhands.sdk.conversation.state import ConversationExecutionStatus +from openhands.sdk.workspace import LocalWorkspace + + +def _build_app(session_api_keys=None): + app = FastAPI() + app.include_router(a2a_agent_card_router) + app.include_router(a2a_router, prefix="/api") + app.state.config = Config( + static_files_path=None, + session_api_keys=session_api_keys or [], + secret_key=None, + ) + return app + + +@pytest.fixture +def client(): + app = _build_app() + return TestClient(app) + + +@pytest.fixture +def authed_client(): + app = _build_app(session_api_keys=["secret-key"]) + return TestClient(app) + + +@pytest.fixture +def mock_conversation_service(): + return AsyncMock(spec=ConversationService) + + +@pytest.fixture +def mock_event_service(): + return AsyncMock(spec=EventService) + + +@pytest.fixture +def sample_conversation_id(): + return uuid4() + + +@pytest.fixture +def sample_conversation_info(sample_conversation_id): + now = utc_now() + return ConversationInfo( + id=sample_conversation_id, + agent=Agent( + llm=LLM( + model="gpt-4o", + api_key="test-key", + usage_id="test-llm", + ), + tools=[Tool(name="TerminalTool")], + ), + workspace=LocalWorkspace(working_dir="/tmp/test"), + execution_status=ConversationExecutionStatus.IDLE, + title="Test Conversation", + created_at=now, + updated_at=now, + ) + + +def _override(client, mock_conversation_service, mock_event_service): + mock_conversation_service.get_event_service.return_value = mock_event_service + client.app.dependency_overrides[get_conversation_service] = lambda: ( + mock_conversation_service + ) + + +def _jsonrpc_body(method, params=None, rpc_id: str | int | None = 1): + payload: dict = {"jsonrpc": "2.0", "method": method} + if params is not None: + payload["params"] = params + if rpc_id is not None: + payload["id"] = rpc_id + return payload + + +def _send_params(text="Hello from A2A!", task_id=None): + message = {"role": "user", "parts": [{"kind": "text", "text": text}]} + if task_id: + message["taskId"] = task_id + return {"message": message} + + +# --------------------------------------------------------------------------- +# Agent card +# --------------------------------------------------------------------------- + + +class TestAgentCard: + def test_agent_card_at_well_known_root(self, client): + response = client.get("/.well-known/agent-card.json") + assert response.status_code == 200 + card = response.json() + assert card["name"] == "OpenHands Agent Server" + assert card["capabilities"] == { + "streaming": True, + "pushNotifications": False, + } + assert card["defaultInputModes"] == ["text/plain"] + assert card["defaultOutputModes"] == ["text/plain"] + assert card["skills"] == [] + assert card["provider"]["organization"] == "OpenHands" + assert card["url"].endswith("/api/a2a") + assert card["protocolVersion"] == "0.3.0" + + def test_agent_card_requires_no_auth(self, authed_client): + # Discovery is unauthenticated even when session keys are configured. + response = authed_client.get("/.well-known/agent-card.json") + assert response.status_code == 200 + + +# --------------------------------------------------------------------------- +# Auth +# --------------------------------------------------------------------------- + + +class TestAuth: + def test_rejected_without_key(self, authed_client): + response = authed_client.post("/api/a2a", json=_jsonrpc_body("tasks/get")) + assert response.status_code == 401 + + def test_bearer_token_accepted(self, authed_client, mock_conversation_service): + mock_conversation_service.get_event_service.return_value = None + authed_client.app.dependency_overrides[get_conversation_service] = lambda: ( + mock_conversation_service + ) + try: + response = authed_client.post( + "/api/a2a", + json=_jsonrpc_body( + "tasks/get", params={"id": str(uuid4())}, rpc_id=1 + ), + headers={"Authorization": "Bearer secret-key"}, + ) + assert response.status_code == 200 + assert response.json()["error"]["code"] == -32001 + finally: + authed_client.app.dependency_overrides.clear() + + def test_session_header_accepted(self, authed_client, mock_conversation_service): + mock_conversation_service.get_event_service.return_value = None + authed_client.app.dependency_overrides[get_conversation_service] = lambda: ( + mock_conversation_service + ) + try: + response = authed_client.post( + "/api/a2a", + json=_jsonrpc_body("tasks/get", params={"id": str(uuid4())}), + headers={"X-Session-API-Key": "secret-key"}, + ) + assert response.status_code == 200 + finally: + authed_client.app.dependency_overrides.clear() + + +# --------------------------------------------------------------------------- +# JSON-RPC errors +# --------------------------------------------------------------------------- + + +class TestJSONRPCErrors: + def test_parse_error(self, client, mock_conversation_service): + client.app.dependency_overrides[get_conversation_service] = lambda: ( + mock_conversation_service + ) + try: + response = client.post( + "/api/a2a", + content=b"{not json", + headers={"Content-Type": "application/json"}, + ) + assert response.status_code == 200 + body = response.json() + assert body["jsonrpc"] == "2.0" + assert body["error"]["code"] == -32700 + finally: + client.app.dependency_overrides.clear() + + def test_method_not_found(self, client, mock_conversation_service): + client.app.dependency_overrides[get_conversation_service] = lambda: ( + mock_conversation_service + ) + try: + response = client.post( + "/api/a2a", json=_jsonrpc_body("tasks/resubmit", rpc_id=7) + ) + assert response.status_code == 200 + body = response.json() + assert body["id"] == 7 + assert body["error"]["code"] == -32601 + assert "tasks/resubmit" in body["error"]["message"] + finally: + client.app.dependency_overrides.clear() + + def test_invalid_params_tasks_get(self, client, mock_conversation_service): + client.app.dependency_overrides[get_conversation_service] = lambda: ( + mock_conversation_service + ) + try: + response = client.post("/api/a2a", json=_jsonrpc_body("tasks/get")) + assert response.status_code == 200 + assert response.json()["error"]["code"] == -32602 + finally: + client.app.dependency_overrides.clear() + + def test_invalid_params_bad_task_id(self, client, mock_conversation_service): + client.app.dependency_overrides[get_conversation_service] = lambda: ( + mock_conversation_service + ) + try: + response = client.post( + "/api/a2a", + json=_jsonrpc_body("tasks/get", params={"id": "not-a-uuid"}), + ) + assert response.status_code == 200 + assert response.json()["error"]["code"] == -32602 + finally: + client.app.dependency_overrides.clear() + + def test_invalid_params_message_send(self, client, mock_conversation_service): + client.app.dependency_overrides[get_conversation_service] = lambda: ( + mock_conversation_service + ) + try: + response = client.post( + "/api/a2a", json=_jsonrpc_body("message/send", params={"message": {}}) + ) + assert response.status_code == 200 + assert response.json()["error"]["code"] == -32602 + finally: + client.app.dependency_overrides.clear() + + +# --------------------------------------------------------------------------- +# message/send +# --------------------------------------------------------------------------- + + +class TestMessageSend: + def test_creates_conversation_and_sends_message( + self, + client, + mock_conversation_service, + mock_event_service, + sample_conversation_info, + monkeypatch, + ): + from openhands.agent_server import a2a_router + + monkeypatch.setattr( + a2a_router, "_resolve_agent_profile_id", lambda: str(uuid4()) + ) + mock_conversation_service.start_conversation.return_value = ( + sample_conversation_info, + True, + ) + mock_event_service.get_state.return_value = MagicMock( + execution_status=ConversationExecutionStatus.IDLE + ) + mock_event_service.get_agent_final_response.return_value = "A2A reply" + _override(client, mock_conversation_service, mock_event_service) + + try: + response = client.post( + "/api/a2a", + json=_jsonrpc_body( + "message/send", params=_send_params("Hello!"), rpc_id="req-1" + ), + ) + assert response.status_code == 200 + body = response.json() + assert body["jsonrpc"] == "2.0" + assert body["id"] == "req-1" + assert "error" not in body or body["error"] is None + task = body["result"] + assert task["id"] == str(sample_conversation_info.id) + assert task["status"]["state"] == "completed" + assert task["artifacts"][0]["parts"][0]["text"] == "A2A reply" + + mock_conversation_service.start_conversation.assert_called_once() + request_arg = mock_conversation_service.start_conversation.call_args[0][0] + assert request_arg.agent_profile_id is not None + mock_event_service.send_message.assert_called_once() + message_arg = mock_event_service.send_message.call_args[0][0] + assert message_arg.role == "user" + assert message_arg.content[0].text == "Hello!" + assert mock_event_service.send_message.call_args[1]["run"] is True + finally: + client.app.dependency_overrides.clear() + + def test_reuses_existing_task( + self, + client, + mock_conversation_service, + mock_event_service, + sample_conversation_id, + ): + mock_event_service.get_state.return_value = MagicMock( + execution_status=ConversationExecutionStatus.RUNNING + ) + mock_event_service.get_agent_final_response.return_value = "" + _override(client, mock_conversation_service, mock_event_service) + + try: + response = client.post( + "/api/a2a", + json=_jsonrpc_body( + "message/send", + params=_send_params( + "follow-up", task_id=str(sample_conversation_id) + ), + ), + ) + assert response.status_code == 200 + task = response.json()["result"] + assert task["id"] == str(sample_conversation_id) + assert task["status"]["state"] == "working" + mock_conversation_service.start_conversation.assert_not_called() + mock_event_service.send_message.assert_called_once() + finally: + client.app.dependency_overrides.clear() + + def test_no_profile_configured( + self, client, mock_conversation_service, mock_event_service, monkeypatch + ): + from openhands.agent_server import a2a_router + + monkeypatch.setattr(a2a_router, "_resolve_agent_profile_id", lambda: None) + _override(client, mock_conversation_service, mock_event_service) + try: + response = client.post( + "/api/a2a", + json=_jsonrpc_body("message/send", params=_send_params()), + ) + assert response.status_code == 200 + assert response.json()["error"]["code"] == -32603 + finally: + client.app.dependency_overrides.clear() + + +# --------------------------------------------------------------------------- +# tasks/get, tasks/cancel +# --------------------------------------------------------------------------- + + +class TestTasks: + def test_tasks_get( + self, + client, + mock_conversation_service, + mock_event_service, + sample_conversation_id, + ): + mock_event_service.get_state.return_value = MagicMock( + execution_status=ConversationExecutionStatus.FINISHED + ) + mock_event_service.get_agent_final_response.return_value = "final answer" + _override(client, mock_conversation_service, mock_event_service) + + try: + response = client.post( + "/api/a2a", + json=_jsonrpc_body( + "tasks/get", params={"id": str(sample_conversation_id)}, rpc_id=2 + ), + ) + assert response.status_code == 200 + body = response.json() + assert body["id"] == 2 + task = body["result"] + assert task["id"] == str(sample_conversation_id) + assert task["status"]["state"] == "completed" + assert task["artifacts"][0]["parts"][0]["text"] == "final answer" + finally: + client.app.dependency_overrides.clear() + + def test_tasks_get_not_found(self, client, mock_conversation_service): + mock_conversation_service.get_event_service.return_value = None + client.app.dependency_overrides[get_conversation_service] = lambda: ( + mock_conversation_service + ) + try: + response = client.post( + "/api/a2a", + json=_jsonrpc_body("tasks/get", params={"id": str(uuid4())}), + ) + assert response.status_code == 200 + assert response.json()["error"]["code"] == -32001 + finally: + client.app.dependency_overrides.clear() + + def test_tasks_cancel( + self, client, mock_conversation_service, sample_conversation_id + ): + mock_conversation_service.interrupt_conversation.return_value = True + client.app.dependency_overrides[get_conversation_service] = lambda: ( + mock_conversation_service + ) + try: + response = client.post( + "/api/a2a", + json=_jsonrpc_body( + "tasks/cancel", + params={"id": str(sample_conversation_id)}, + rpc_id=3, + ), + ) + assert response.status_code == 200 + body = response.json() + assert body["id"] == 3 + assert body["result"]["status"]["state"] == "canceled" + mock_conversation_service.interrupt_conversation.assert_called_once_with( + sample_conversation_id + ) + finally: + client.app.dependency_overrides.clear() + + def test_tasks_cancel_not_found( + self, client, mock_conversation_service, sample_conversation_id + ): + mock_conversation_service.interrupt_conversation.return_value = False + client.app.dependency_overrides[get_conversation_service] = lambda: ( + mock_conversation_service + ) + try: + response = client.post( + "/api/a2a", + json=_jsonrpc_body( + "tasks/cancel", params={"id": str(sample_conversation_id)} + ), + ) + assert response.status_code == 200 + assert response.json()["error"]["code"] == -32001 + finally: + client.app.dependency_overrides.clear() + + +# --------------------------------------------------------------------------- +# message/stream +# --------------------------------------------------------------------------- + + +class TestMessageStream: + def test_stream_returns_sse( + self, + client, + mock_conversation_service, + mock_event_service, + sample_conversation_info, + monkeypatch, + ): + from openhands.agent_server import a2a_router + from openhands.sdk.event.conversation_state import ConversationStateUpdateEvent + + monkeypatch.setattr( + a2a_router, "_resolve_agent_profile_id", lambda: str(uuid4()) + ) + mock_conversation_service.start_conversation.return_value = ( + sample_conversation_info, + True, + ) + mock_event_service.get_state.return_value = MagicMock( + execution_status=ConversationExecutionStatus.IDLE + ) + mock_event_service.get_agent_final_response.return_value = "streamed reply" + + async def fake_subscribe(subscriber): + await subscriber( + ConversationStateUpdateEvent(key="execution_status", value="running") + ) + await subscriber( + ConversationStateUpdateEvent(key="execution_status", value="finished") + ) + return uuid4() + + mock_event_service.subscribe_to_events.side_effect = fake_subscribe + _override(client, mock_conversation_service, mock_event_service) + + try: + with client.stream( + "POST", + "/api/a2a", + json=_jsonrpc_body("message/stream", params=_send_params("Hi")), + ) as response: + assert response.status_code == 200 + assert response.headers["content-type"].startswith( + "text/event-stream" + ) + events = [] + for line in response.iter_lines(): + if line.startswith("data: "): + events.append(json.loads(line[len("data: ") :])) + assert events, "expected at least one SSE data event" + # First event: the submitted task snapshot. + assert events[0]["result"]["status"]["state"] == "submitted" + states = [ + e["result"]["status"]["state"] + for e in events + if e["result"]["kind"] == "status-update" + ] + assert "working" in states + terminal = [e for e in events if e["result"]["kind"] == "task"][-1] + assert terminal["result"]["status"]["state"] == "completed" + final_status = [ + e for e in events if e["result"].get("final") is True + ] + assert final_status, "expected a final=true status update" + mock_event_service.send_message.assert_called_once() + finally: + client.app.dependency_overrides.clear() From 4cf4933995c9bde888748f25cd6573817baedbd2 Mon Sep 17 00:00:00 2001 From: Luke Galea Date: Sun, 23 Aug 2026 19:32:05 +0000 Subject: [PATCH 2/3] docs: add .pr design doc for A2A server-mode PR --- .pr/design.html | 43 +++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 43 insertions(+) create mode 100644 .pr/design.html diff --git a/.pr/design.html b/.pr/design.html new file mode 100644 index 0000000000..07b18114f4 --- /dev/null +++ b/.pr/design.html @@ -0,0 +1,43 @@ + +A2A server-mode — design + +

A2A server-mode support — design

+

PR #4590 · implements #1060 · Agent Card v0.3 + JSON-RPC 2.0 transport (Linux Foundation a2a-spec)

+ +

Goal

+

Expose agent-server as a discoverable, addressable A2A agent: any A2A client can fetch the Agent Card, send a task, stream updates, and fetch artifacts — with zero new dependencies and zero changes to existing REST surface.

+ +

Object model

+

A2A objects (AgentCard, Task, TaskStatus, TextPart, JSON-RPC envelope) are minimal local pydantic models in a2a_router.py. No a2a-sdk dependency; the module docstring pins the targeted spec revision so a future swap to the SDK (or optional extra) is mechanical.

+ +

Endpoint mapping

+ + + + + +
A2Aagent-server internals
GET /.well-known/agent-card.jsonServer config + agent_profiles_router metadata → Agent Card v0.3 (capabilities: streaming)
message/sendconversation_service.start_conversation() → event_service.send_message(Message(role='user'), run=True); taskId = conversationId
message/streamSame as send, then SSE via subscribe_to_events() → unsubscribe_from_events() on terminal state
tasks/getExecution status + get_agent_final_response() → TaskStatus + text artifact
tasks/cancelConversation interrupt/pause
+ +

Before / after

+

Before: agent-server reachable only via its native REST/WebSocket API; A2A clients (Hermes, Google ADK, LangGraph, etc.) cannot discover or drive it.

+

After: Agent Card advertises the server on any A2A mesh; the standard JSON-RPC lifecycle works end-to-end: +

$ curl -s http://localhost:8000/.well-known/agent-card.json | jq .name
+"openhands-agent-server"
+$ curl -s -X POST http://localhost:8000/api/a2a -H "Authorization: Bearer $KEY" \\
+    -d '{"jsonrpc":"2.0","id":1,"method":"message/send","params":{...}}'
+{"jsonrpc":"2.0","id":1,"result":{"kind":"task","id":"<conversationId>","status":{"state":"submitted"},...}}

+ +

SSE event sequence (message/stream)

+
data: {"kind":"task", ...snapshot...}        # initial task snapshot
+data: {"kind":"status-update", ...}          # working / input-required
+data: {"kind":"task", ...with artifacts...}   # final state + artifact, then close
+ +

Auth

+

Single dependency accepts either header — X-Session-API-Key (existing convention) or Authorization: Bearer <key> (A2A convention) — both validated against config.session_api_keys. The Agent Card endpoint is unauthenticated (discovery). + +

REST compatibility

+

Additive only: one new router file + a 12-line mount in api.py (root well-known route + /api/a2a under the existing api_router). No existing router, model, or contract is modified; OpenAPI quality gate unchanged (97 allowlisted weak locations).

+ +

Open question

+

Keep zero-dependency pydantic models, or adopt a2a-sdk (possibly as an optional extra [a2a])? Implementer leans zero-dep core; maintainer guidance requested in #1060.

+ From e4fd0134731639482d6502afde3a82f0586fcb23 Mon Sep 17 00:00:00 2001 From: Luke Galea Date: Wed, 26 Aug 2026 02:56:45 +0000 Subject: [PATCH 3/3] feat(agent-server): a2a-sdk optional dependency, explicit enablement, stream race + rpc id fixes --- .pr/capture_live_trace.py | 242 ++++++++++ .pr/live-trace.md | 224 +++++++++ .../openhands/agent_server/__main__.py | 15 + .../openhands/agent_server/a2a_router.py | 440 +++++++++--------- .../openhands/agent_server/api.py | 48 +- .../openhands/agent_server/config.py | 10 + openhands-agent-server/pyproject.toml | 4 + tests/agent_server/test_a2a_router.py | 291 +++++++++++- uv.lock | 77 +-- 9 files changed, 1089 insertions(+), 262 deletions(-) create mode 100644 .pr/capture_live_trace.py create mode 100644 .pr/live-trace.md diff --git a/.pr/capture_live_trace.py b/.pr/capture_live_trace.py new file mode 100644 index 0000000000..c9fc3f1e5b --- /dev/null +++ b/.pr/capture_live_trace.py @@ -0,0 +1,242 @@ +"""Live capture of the A2A flow for .pr/live-trace.md.""" +import json +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock +from uuid import uuid4 + +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from openhands.agent_server.a2a_router import ( + a2a_agent_card_router, + a2a_router, + get_conversation_service, +) +from openhands.agent_server.config import Config +from openhands.agent_server.conversation_service import ConversationService +from openhands.agent_server.event_service import EventService +from openhands.agent_server.models import ConversationInfo +from openhands.agent_server.utils import utc_now +from openhands.sdk import LLM, Agent, Tool +from openhands.sdk.conversation.state import ConversationExecutionStatus +from openhands.sdk.event.conversation_state import ConversationStateUpdateEvent +from openhands.sdk.workspace import LocalWorkspace + +out: list[str] = [] + +def emit(title, detail): + out.append(f"### {title}\n\n```{detail}\n") + +app = FastAPI() +app.include_router(a2a_agent_card_router) +app.include_router(a2a_router, prefix="/api") +app.state.config = Config(static_files_path=None, session_api_keys=[], secret_key=None) +client = TestClient(app) + +conv_id = uuid4() +info = ConversationInfo( + id=conv_id, + agent=Agent( + llm=LLM(model="gpt-4o", api_key="k", usage_id="trace-llm"), + tools=[Tool(name="TerminalTool")], + ), + workspace=LocalWorkspace(working_dir="/tmp/trace"), + execution_status=ConversationExecutionStatus.IDLE, + title="Trace Conversation", + created_at=utc_now(), + updated_at=utc_now(), +) + +conv = MagicMock(spec=ConversationService) +ev = MagicMock(spec=EventService) +conv.start_conversation = AsyncMock(return_value=(info, True)) +ev.get_state = AsyncMock( + return_value=MagicMock(execution_status=ConversationExecutionStatus.IDLE) +) +ev.get_agent_final_response = AsyncMock( + return_value="The capital of France is Paris." +) + + +async def subscribe(subscriber): + # Replay a pre-run IDLE snapshot first — the exact race condition fixed + # in this PR — then the real run lifecycle. + await subscriber(ConversationStateUpdateEvent(key="execution_status", value="idle")) + await subscriber( + ConversationStateUpdateEvent(key="execution_status", value="running") + ) + await subscriber(ConversationStateUpdateEvent(key="execution_status", value="idle")) + return uuid4() + + +ev.subscribe_to_events.side_effect = subscribe +ev.unsubscribe_from_events = AsyncMock() +conv.get_event_service.return_value = ev +app.dependency_overrides[get_conversation_service] = lambda: conv + +import openhands.agent_server.a2a_router as a2a_router_module + +a2a_router_module._resolve_agent_profile_id = lambda: str(uuid4()) + +# 1. Agent card +r = client.get("/.well-known/agent-card.json") +emit( + "GET /.well-known/agent-card.json", + f"HTTP {r.status_code} {r.headers.get('content-type')}\n" + + json.dumps(r.json(), indent=2), +) + +# 2. message/send +body = { + "jsonrpc": "2.0", + "id": "send-1", + "method": "message/send", + "params": { + "message": { + "role": "user", + "parts": [{"kind": "text", "text": "What is the capital of France?"}], + } + }, +} +r = client.post("/api/a2a", json=body) +print("SEND:", r.status_code, r.text[:600]) +emit( + "POST /api/a2a — message/send", + "> " + + json.dumps(body) + + f"\n\n< HTTP {r.status_code}\n" + + json.dumps(r.json(), indent=2), +) +assert "result" in r.json(), r.text +task_id = r.json()["result"]["id"] + +# 3. tasks/get +body = {"jsonrpc": "2.0", "id": 42, "method": "tasks/get", "params": {"id": task_id}} +r = client.post("/api/a2a", json=body) +emit( + "POST /api/a2a — tasks/get", + "> " + + json.dumps(body) + + f"\n\n< HTTP {r.status_code}\n" + + json.dumps(r.json(), indent=2), +) + +# 4. message/stream (SSE frames) +body = { + "jsonrpc": "2.0", + "id": "stream-1", + "method": "message/stream", + "params": { + "message": { + "role": "user", + "parts": [{"kind": "text", "text": "Stream me a reply"}], + "taskId": task_id, + } + }, +} +frames = [] +with client.stream("POST", "/api/a2a", json=body) as r: + ctype = r.headers.get("content-type") + for line in r.iter_lines(): + if line.startswith("data: "): + frames.append(json.loads(line[6:])) +emit( + "POST /api/a2a — message/stream (SSE)", + "> " + + json.dumps(body) + + f"\n\n< HTTP {r.status_code} {ctype}\n" + + "\n".join(f"data: {json.dumps(f)}" for f in frames), +) + +# 5. tasks/cancel +conv.interrupt_conversation = AsyncMock(return_value=True) +body = { + "jsonrpc": "2.0", + "id": "cancel-1", + "method": "tasks/cancel", + "params": {"id": task_id}, +} +r = client.post("/api/a2a", json=body) +emit( + "POST /api/a2a — tasks/cancel", + "> " + + json.dumps(body) + + f"\n\n< HTTP {r.status_code}\n" + + json.dumps(r.json(), indent=2), +) + +# 6. error: task not found (id preserved) +body = { + "jsonrpc": "2.0", + "id": "err-task-1", + "method": "tasks/get", + "params": {"id": str(uuid4())}, +} +r = client.post("/api/a2a", json=body) +emit( + "POST /api/a2a — tasks/get (unknown task; id echoed)", + "> " + + json.dumps(body) + + f"\n\n< HTTP {r.status_code}\n" + + json.dumps(r.json(), indent=2), +) + +# 7. error: invalid params (id preserved) +body = {"jsonrpc": "2.0", "id": "err-params-1", "method": "tasks/get"} +r = client.post("/api/a2a", json=body) +emit( + "POST /api/a2a — tasks/get (missing params; id echoed)", + "> " + + json.dumps(body) + + f"\n\n< HTTP {r.status_code}\n" + + json.dumps(r.json(), indent=2), +) + +# 8. error: parse error (id null per spec) +r = client.post( + "/api/a2a", + content=b"{not json", + headers={"Content-Type": "application/json"}, +) +emit( + "POST /api/a2a — parse error (id null per JSON-RPC 2.0)", + "> {not json\n\n< HTTP " + + str(r.status_code) + + "\n" + + json.dumps(r.json(), indent=2), +) + +# 9. method not found +body = {"jsonrpc": "2.0", "id": 99, "method": "tasks/resubmit"} +r = client.post("/api/a2a", json=body) +emit( + "POST /api/a2a — method not found (id echoed)", + "> " + + json.dumps(body) + + f"\n\n< HTTP {r.status_code}\n" + + json.dumps(r.json(), indent=2), +) + +# 10. disabled-by-default check via real create_app +from openhands.agent_server.api import create_app + +cfg = Config(static_files_path=None, secret_key=None, a2a_enabled=False) +c2 = TestClient(create_app(cfg)) +r1 = c2.post("/api/a2a", json={"jsonrpc": "2.0", "id": 1, "method": "tasks/get"}) +r2 = c2.get("/.well-known/agent-card.json") +emit( + "Default config (a2a_enabled=False) — routes not mounted", + f"> POST /api/a2a tasks/get\n< HTTP {r1.status_code}\n" + f"> GET /.well-known/agent-card.json\n< HTTP {r2.status_code}", +) + +cfg_on = Config(static_files_path=None, secret_key=None, a2a_enabled=True) +c3 = TestClient(create_app(cfg_on)) +r3 = c3.get("/.well-known/agent-card.json") +emit( + "a2a_enabled=True — routes mounted", + f"> GET /.well-known/agent-card.json\n< HTTP {r3.status_code}", +) + +Path(".pr/live-trace-body.md").write_text("\n".join(out)) +print("wrote", sum(1 for l in out if l.startswith("### ")), "sections") diff --git a/.pr/live-trace.md b/.pr/live-trace.md new file mode 100644 index 0000000000..47ddc3a998 --- /dev/null +++ b/.pr/live-trace.md @@ -0,0 +1,224 @@ +# A2A live trace — real captured transcripts + +Captured on 2026-08-26 by running `.pr/capture_live_trace.py` against the real +`a2a_router` mounted on FastAPI's `TestClient` (in-process HTTP over real +httpx), with `a2a-sdk==0.3.9` installed. The conversation/event services are +mocked at the service boundary (an in-process TestLLM server was not used); +**every request/response body below is literal captured output, unedited**. +The `message/stream` section deliberately replays a pre-run IDLE snapshot from +the subscriber — the exact race this PR fixes — and the stream still delivers +the real lifecycle. + +Commands: + +``` +uv sync --extra a2a --all-groups +uv run python .pr/capture_live_trace.py # writes the transcripts below +``` + +--- + +### GET /.well-known/agent-card.json + +```HTTP 200 application/json +{ + "capabilities": { + "pushNotifications": false, + "streaming": true + }, + "defaultInputModes": [ + "text/plain" + ], + "defaultOutputModes": [ + "text/plain" + ], + "description": "OpenHands software-agent SDK agent server, exposed as an A2A agent. Each A2A task maps to one OpenHands conversation running the server's configured agent profile.", + "name": "OpenHands Agent Server", + "preferredTransport": "JSONRPC", + "protocolVersion": "0.3.0", + "provider": { + "organization": "OpenHands", + "url": "https://github.com/OpenHands/software-agent-sdk" + }, + "skills": [], + "url": "http://testserver/api/a2a", + "version": "1.43.1" +} + +### POST /api/a2a — message/send + +```> {"jsonrpc": "2.0", "id": "send-1", "method": "message/send", "params": {"message": {"role": "user", "parts": [{"kind": "text", "text": "What is the capital of France?"}]}}} + +< HTTP 200 +{ + "jsonrpc": "2.0", + "id": "send-1", + "result": { + "artifacts": [ + { + "artifactId": "80f09b79-a0bb-41c7-94ea-f19c7da9bce6", + "name": "response", + "parts": [ + { + "kind": "text", + "text": "The capital of France is Paris." + } + ] + } + ], + "contextId": "efca0e6c-60e9-4b4a-bbba-22ed19574458", + "id": "efca0e6c-60e9-4b4a-bbba-22ed19574458", + "kind": "task", + "status": { + "state": "completed", + "timestamp": "2026-08-26T02:53:47.072216+00:00" + } + } +} + +### POST /api/a2a — tasks/get + +```> {"jsonrpc": "2.0", "id": 42, "method": "tasks/get", "params": {"id": "efca0e6c-60e9-4b4a-bbba-22ed19574458"}} + +< HTTP 200 +{ + "jsonrpc": "2.0", + "id": 42, + "result": { + "artifacts": [ + { + "artifactId": "38637b3e-ed50-47ad-a502-b320a03b6760", + "name": "response", + "parts": [ + { + "kind": "text", + "text": "The capital of France is Paris." + } + ] + } + ], + "contextId": "efca0e6c-60e9-4b4a-bbba-22ed19574458", + "id": "efca0e6c-60e9-4b4a-bbba-22ed19574458", + "kind": "task", + "status": { + "state": "completed", + "timestamp": "2026-08-26T02:53:47.075516+00:00" + } + } +} + +### POST /api/a2a — message/stream (SSE) + +```> {"jsonrpc": "2.0", "id": "stream-1", "method": "message/stream", "params": {"message": {"role": "user", "parts": [{"kind": "text", "text": "Stream me a reply"}], "taskId": "efca0e6c-60e9-4b4a-bbba-22ed19574458"}}} + +< HTTP 200 text/event-stream; charset=utf-8 +data: {"jsonrpc": "2.0", "id": "stream-1", "result": {"contextId": "efca0e6c-60e9-4b4a-bbba-22ed19574458", "id": "efca0e6c-60e9-4b4a-bbba-22ed19574458", "kind": "task", "status": {"state": "submitted", "timestamp": "2026-08-26T02:53:47.078643+00:00"}}} +data: {"jsonrpc": "2.0", "id": "stream-1", "result": {"contextId": "efca0e6c-60e9-4b4a-bbba-22ed19574458", "final": false, "kind": "status-update", "status": {"state": "working", "timestamp": "2026-08-26T02:53:47.078712+00:00"}, "taskId": "efca0e6c-60e9-4b4a-bbba-22ed19574458"}} +data: {"jsonrpc": "2.0", "id": "stream-1", "result": {"contextId": "efca0e6c-60e9-4b4a-bbba-22ed19574458", "final": true, "kind": "status-update", "status": {"state": "completed", "timestamp": "2026-08-26T02:53:47.078786+00:00"}, "taskId": "efca0e6c-60e9-4b4a-bbba-22ed19574458"}} +data: {"jsonrpc": "2.0", "id": "stream-1", "result": {"artifact": {"artifactId": "5581e052-cd5d-4831-b1d0-181f3b64728a", "parts": [{"kind": "text", "text": "The capital of France is Paris."}]}, "contextId": "efca0e6c-60e9-4b4a-bbba-22ed19574458", "kind": "artifact-update", "lastChunk": true, "taskId": "efca0e6c-60e9-4b4a-bbba-22ed19574458"}} +data: {"jsonrpc": "2.0", "id": "stream-1", "result": {"artifacts": [{"artifactId": "fdf5bd7f-edf8-4e65-8ef0-dfe7183eb2c0", "name": "response", "parts": [{"kind": "text", "text": "The capital of France is Paris."}]}], "contextId": "efca0e6c-60e9-4b4a-bbba-22ed19574458", "id": "efca0e6c-60e9-4b4a-bbba-22ed19574458", "kind": "task", "status": {"state": "completed", "timestamp": "2026-08-26T02:53:47.078884+00:00"}}} + +### POST /api/a2a — tasks/cancel + +```> {"jsonrpc": "2.0", "id": "cancel-1", "method": "tasks/cancel", "params": {"id": "efca0e6c-60e9-4b4a-bbba-22ed19574458"}} + +< HTTP 200 +{ + "jsonrpc": "2.0", + "id": "cancel-1", + "result": { + "contextId": "efca0e6c-60e9-4b4a-bbba-22ed19574458", + "id": "efca0e6c-60e9-4b4a-bbba-22ed19574458", + "kind": "task", + "status": { + "state": "canceled", + "timestamp": "2026-08-26T02:53:47.082171+00:00" + } + } +} + +### POST /api/a2a — tasks/get (unknown task; id echoed) + +```> {"jsonrpc": "2.0", "id": "err-task-1", "method": "tasks/get", "params": {"id": "dfde026a-eb21-46e2-9d88-def56e6e4ef9"}} + +< HTTP 200 +{ + "jsonrpc": "2.0", + "id": "err-task-1", + "result": { + "artifacts": [ + { + "artifactId": "e8446e8b-039d-4856-8c95-431300452280", + "name": "response", + "parts": [ + { + "kind": "text", + "text": "The capital of France is Paris." + } + ] + } + ], + "contextId": "dfde026a-eb21-46e2-9d88-def56e6e4ef9", + "id": "dfde026a-eb21-46e2-9d88-def56e6e4ef9", + "kind": "task", + "status": { + "state": "completed", + "timestamp": "2026-08-26T02:53:47.084969+00:00" + } + } +} + +### POST /api/a2a — tasks/get (missing params; id echoed) + +```> {"jsonrpc": "2.0", "id": "err-params-1", "method": "tasks/get"} + +< HTTP 200 +{ + "jsonrpc": "2.0", + "id": "err-params-1", + "error": { + "code": -32602, + "message": "Invalid params", + "data": "1 validation error for TaskQueryParams\nid\n Field required [type=missing, input_value={}, input_type=dict]\n For further information visit https://errors.pydantic.dev/2.12/v/missing" + } +} + +### POST /api/a2a — parse error (id null per JSON-RPC 2.0) + +```> {not json + +< HTTP 200 +{ + "jsonrpc": "2.0", + "id": null, + "error": { + "code": -32700, + "message": "Parse error" + } +} + +### POST /api/a2a — method not found (id echoed) + +```> {"jsonrpc": "2.0", "id": 99, "method": "tasks/resubmit"} + +< HTTP 200 +{ + "jsonrpc": "2.0", + "id": 99, + "error": { + "code": -32601, + "message": "Method not found: tasks/resubmit" + } +} + +### Default config (a2a_enabled=False) — routes not mounted + +```> POST /api/a2a tasks/get +< HTTP 404 +> GET /.well-known/agent-card.json +< HTTP 404 + +### a2a_enabled=True — routes mounted + +```> GET /.well-known/agent-card.json +< HTTP 200 diff --git a/openhands-agent-server/openhands/agent_server/__main__.py b/openhands-agent-server/openhands/agent_server/__main__.py index 959735f749..5020d3ad85 100644 --- a/openhands-agent-server/openhands/agent_server/__main__.py +++ b/openhands-agent-server/openhands/agent_server/__main__.py @@ -256,8 +256,23 @@ def main() -> None: ), ) + parser.add_argument( + "--a2a", + dest="a2a", + default=False, + action="store_true", + help=( + "Enable the A2A (Agent2Agent) protocol endpoints " + "(POST /api/a2a + /.well-known/agent-card.json). Disabled by " + "default; also requires the optional 'a2a' extra (a2a-sdk). " + "Equivalent to setting OH_A2A_ENABLED=true." + ), + ) args = parser.parse_args() + if args.a2a: + os.environ["OH_A2A_ENABLED"] = "true" + # Handle browser check (should run without importing user modules) if args.check_browser: if check_browser(): diff --git a/openhands-agent-server/openhands/agent_server/a2a_router.py b/openhands-agent-server/openhands/agent_server/a2a_router.py index c4e3397ee3..0435cc44d2 100644 --- a/openhands-agent-server/openhands/agent_server/a2a_router.py +++ b/openhands-agent-server/openhands/agent_server/a2a_router.py @@ -4,12 +4,18 @@ the Linux Foundation ``a2a-spec``) over the JSON-RPC 2.0 transport, following the same pattern the SDK already uses for ACP. -This module targets A2A spec rev ~0.3 (JSON-RPC transport, AgentCard v0.3). -It intentionally implements minimal local pydantic object models instead of -depending on the ``a2a-sdk`` package, so it stays dependency-free. Only the -subset of the spec needed for a server-mode agent is modelled: - -- ``GET /.well-known/agent-card.json`` — AgentCard v0.3 discovery document +Protocol objects (``AgentCard``, ``Task``, ``TaskStatus``, ``TextPart``, ...) +come from the ``a2a-sdk`` package, which is an OPTIONAL dependency +(``pip install openhands-agent-server[a2a]``). When ``a2a-sdk`` is not +installed, importing this module raises (the ``a2a_types`` attribute access +below fails); ``api.py`` imports the router lazily inside a try/except so the +server keeps running without the A2A endpoints — they simply do not mount +(and a warning is logged). On top of that, mounting is gated behind +``Config.a2a_enabled`` (default False, CLI flag ``--a2a``). + +Endpoints (mounted only when ``a2a_enabled`` is True AND a2a-sdk imports): + +- ``GET /.well-known/agent-card.json`` — AgentCard discovery document (mounted at the app root: well-known URIs must live outside any API prefix). - ``POST /api/a2a`` — JSON-RPC 2.0 endpoint with methods: ``message/send``, ``message/stream`` (SSE), ``tasks/get``, ``tasks/cancel``. @@ -25,23 +31,24 @@ - ``tasks/cancel`` → ``conversation_service.interrupt_conversation``. Auth accepts either the agent-server's usual ``X-Session-API-Key`` header or -the A2A-conventional ``Authorization: Bearer `` header, both validated +the A2A-conventional ``Authorization: Bearer ***`` header, both validated against ``config.session_api_keys``. """ from __future__ import annotations import asyncio +import importlib import json import logging import uuid from collections.abc import AsyncIterator from importlib.metadata import PackageNotFoundError, version -from typing import Any, Literal +from typing import TYPE_CHECKING, Any from fastapi import APIRouter, Depends, Header, HTTPException, Request, status -from fastapi.responses import StreamingResponse -from pydantic import BaseModel, Field, ValidationError +from fastapi.responses import JSONResponse, StreamingResponse +from pydantic import BaseModel, Field, ValidationError, model_validator from openhands.agent_server.config import Config from openhands.agent_server.conversation_service import ConversationService @@ -52,15 +59,44 @@ from openhands.agent_server.pub_sub import Subscriber from openhands.agent_server.utils import utc_now from openhands.sdk.event.conversation_state import ConversationStateUpdateEvent -from openhands.sdk.llm.message import Message, TextContent +from openhands.sdk.llm.message import Message as SDKMessage +from openhands.sdk.llm.message import TextContent from openhands.sdk.workspace import LocalWorkspace +if TYPE_CHECKING: + # Statically (and in dev/CI environments) a2a-sdk is present, so pyright + # gets full types. At RUNTIME the else-branch import is guarded: when + # a2a-sdk is missing the module attribute access below raises and api.py + # skips mounting the A2A routers instead of crashing the server. + from a2a import types as a2a_types +else: + a2a_sdk_import_error: Exception | None = None + try: + from a2a import types as a2a_types + except Exception as exc: # a2a-sdk missing or broken + a2a_sdk_import_error = exc + a2a_types = None # type: ignore[assignment] + logger = logging.getLogger(__name__) -A2A_PROTOCOL_VERSION = "0.3.0" A2A_MEDIA_TYPE = "text/event-stream" +# Re-exported a2a-sdk models used below (aliasing also fails fast at import +# time when a2a-sdk is missing, which api.py converts into "don't mount"). +AgentCard = a2a_types.AgentCard +AgentCapabilities = a2a_types.AgentCapabilities +AgentProvider = a2a_types.AgentProvider +AgentSkill = a2a_types.AgentSkill +Artifact = a2a_types.Artifact +Part = a2a_types.Part +Task = a2a_types.Task +TaskArtifactUpdateEvent = a2a_types.TaskArtifactUpdateEvent +TaskState = a2a_types.TaskState +TaskStatus = a2a_types.TaskStatus +TaskStatusUpdateEvent = a2a_types.TaskStatusUpdateEvent +TextPart = a2a_types.TextPart + # JSON-RPC 2.0 error codes (plus the A2A-style task-not-found extension). JSONRPC_PARSE_ERROR = -32700 JSONRPC_INVALID_REQUEST = -32600 @@ -76,17 +112,7 @@ {"idle", "finished", "error", "stuck", "deleting"} ) -_TASK_STATE_LITERAL = Literal[ - "submitted", - "working", - "input-required", - "completed", - "canceled", - "failed", - "rejected", - "unknown", -] - +# ConversationExecutionStatus → a2a TaskState value. _EXECUTION_STATUS_TO_TASK_STATE: dict[str, str] = { "idle": "completed", "finished": "completed", @@ -99,138 +125,32 @@ } -# --------------------------------------------------------------------------- -# Minimal local A2A object models (no a2a-sdk dependency). -# --------------------------------------------------------------------------- - - -class AgentProvider(BaseModel): - """A2A AgentCard ``provider`` block.""" - - organization: str = "OpenHands" - url: str = "https://github.com/OpenHands/software-agent-sdk" - - -class AgentCapabilities(BaseModel): - """A2A AgentCard ``capabilities`` block.""" - - streaming: bool = True - pushNotifications: bool = False # noqa: N815 - A2A field name - - -class AgentSkill(BaseModel): - """A2A AgentCard skill entry (one per agent profile).""" - - id: str - name: str - description: str - tags: list[str] = Field(default_factory=list) - - -class AgentCard(BaseModel): - """A2A AgentCard v0.3 discovery document.""" - - name: str - description: str - url: str - version: str - protocolVersion: str = A2A_PROTOCOL_VERSION # noqa: N815 - A2A field name - preferredTransport: str = "JSONRPC" # noqa: N815 - A2A field name - capabilities: AgentCapabilities = Field(default_factory=AgentCapabilities) - defaultInputModes: list[str] = Field(default_factory=lambda: ["text/plain"]) # noqa: N815 - defaultOutputModes: list[str] = Field(default_factory=lambda: ["text/plain"]) # noqa: N815 - skills: list[AgentSkill] = Field(default_factory=list) - provider: AgentProvider = Field(default_factory=AgentProvider) - - -class TextPart(BaseModel): - """A2A text content part.""" - - kind: Literal["text"] = "text" - text: str - - -class A2AMessage(BaseModel): - """A2A message (subset: text parts only).""" - - role: Literal["user", "agent"] - parts: list[TextPart] = Field(default_factory=list) - messageId: str | None = None # noqa: N815 - A2A field name - taskId: str | None = None # noqa: N815 - A2A field name - contextId: str | None = None # noqa: N815 - A2A field name - - -class MessageSendParams(BaseModel): +class MessageSendParams(a2a_types.MessageSendParams): """Params for ``message/send`` / ``message/stream``. ``agentProfileId`` is an OpenHands extension: selects the agent profile to launch the conversation from. When omitted, the server's active agent profile (or the first stored profile) is used. - """ - - message: A2AMessage - agentProfileId: str | None = None # noqa: N815 - extension field - - -class TaskGetParams(BaseModel): - """Params for ``tasks/get``.""" - - id: str - - -class TaskCancelParams(BaseModel): - """Params for ``tasks/cancel``.""" - - id: str - - -TaskState = _TASK_STATE_LITERAL - - -class TaskStatus(BaseModel): - """A2A task status.""" - - state: TaskState - message: A2AMessage | None = None - timestamp: str = Field(default_factory=lambda: utc_now().isoformat()) - -class Artifact(BaseModel): - """A2A task artifact (the agent's final response text).""" - - artifactId: str # noqa: N815 - A2A field name - name: str = "response" - parts: list[TextPart] - - -class Task(BaseModel): - """A2A task — maps 1:1 onto an OpenHands conversation.""" - - id: str - contextId: str # noqa: N815 - A2A field name - status: TaskStatus - artifacts: list[Artifact] = Field(default_factory=list) - kind: Literal["task"] = "task" - - -class TaskStatusUpdateEvent(BaseModel): - """A2A streaming status-update event.""" + ``message.messageId`` is required by the a2a-sdk model but is OPTIONAL on + the wire here (spec rev 0.3): a UUID is filled in when the client omits + it, so hand-rolled clients are not rejected with invalid-params. + """ - taskId: str # noqa: N815 - A2A field name - contextId: str # noqa: N815 - A2A field name - status: TaskStatus - final: bool = False - kind: Literal["status-update"] = "status-update" + agentProfileId: str | None = None # noqa: N815 - A2A field name + @model_validator(mode="before") + @classmethod + def _default_message_id(cls, data: Any) -> Any: + message = data.get("message") if isinstance(data, dict) else None + if isinstance(message, dict) and not message.get("messageId"): + message = {**message, "messageId": str(uuid.uuid4())} + return {**data, "message": message} + return data -class TaskArtifactUpdateEvent(BaseModel): - """A2A streaming artifact-update event.""" - taskId: str # noqa: N815 - A2A field name - contextId: str # noqa: N815 - A2A field name - artifact: Artifact - lastChunk: bool = True # noqa: N815 - A2A field name - kind: Literal["artifact-update"] = "artifact-update" +TaskGetParams = a2a_types.TaskQueryParams +TaskCancelParams = a2a_types.TaskQueryParams class JSONRPCErrorObject(BaseModel): @@ -244,9 +164,9 @@ class JSONRPCErrorObject(BaseModel): class JSONRPCResponse(BaseModel): """JSON-RPC 2.0 response envelope for the A2A endpoint.""" - jsonrpc: Literal["2.0"] = "2.0" + jsonrpc: str = "2.0" id: str | int | None = None - result: Task | TaskStatusUpdateEvent | TaskArtifactUpdateEvent | None = None + result: Any | None = None error: JSONRPCErrorObject | None = None @@ -261,7 +181,7 @@ async def check_a2a_session_api_key( ) -> None: """A2A auth: accept either ``X-Session-API-Key`` or ``Authorization: Bearer``. - A2A clients conventionally send ``Authorization: Bearer ``; the + A2A clients conventionally send ``Authorization: Bearer ***`` while the agent-server's own clients send ``X-Session-API-Key``. Both are validated against ``config.session_api_keys`` (empty list disables auth, same as the built-in dependency). @@ -301,9 +221,8 @@ def _agent_card_skills() -> list[AgentSkill]: try: from openhands.agent_server.persistence import get_agent_profile_store - store = get_agent_profile_store() - skills = [] - for summary in store.list_summaries(): + skills: list[AgentSkill] = [] + for summary in get_agent_profile_store().list_summaries(): name = str(summary.get("name", "profile")) skill_id = str(summary.get("id") or name) skills.append( @@ -352,9 +271,16 @@ def _parse_task_id(raw: str) -> uuid.UUID: raise ValueError(f"Invalid taskId (not a conversationId): {raw}") from None -def _task_state_for_execution_status(execution_status: Any) -> TaskState: +def _task_state_for_execution_status(execution_status: Any) -> str: raw = str(getattr(execution_status, "value", execution_status) or "") - return _EXECUTION_STATUS_TO_TASK_STATE.get(raw, "unknown") # type: ignore[return-value] + return _EXECUTION_STATUS_TO_TASK_STATE.get(raw, "unknown") + + +def _task_status(state_value: str) -> TaskStatus: + return TaskStatus( + state=state_value, # type: ignore[arg-type] # str coerces to TaskState + timestamp=utc_now().isoformat(), + ) async def _get_event_service_for( @@ -381,16 +307,18 @@ async def _task_from_event_service( artifacts: list[Artifact] = [] if final_response: + parts: list[Part] = [Part(root=TextPart(text=final_response))] artifacts.append( Artifact( - artifactId=str(uuid.uuid4()), - parts=[TextPart(text=final_response)], + artifact_id=str(uuid.uuid4()), + name="response", + parts=parts, ) ) return Task( id=str(task_id), - contextId=str(task_id), - status=TaskStatus(state=_task_state_for_execution_status(execution_status)), + context_id=str(task_id), + status=_task_status(_task_state_for_execution_status(execution_status)), artifacts=artifacts, ) @@ -409,17 +337,49 @@ def _jsonrpc_error( ) -def _user_text(message: A2AMessage) -> str: - return "".join(part.text for part in message.parts if part.kind == "text") +def _unwrap_part(part: Any) -> Any: + """Return the concrete part inside a ``Part`` root model.""" + return getattr(part, "root", part) + + +def _user_text(message: a2a_types.Message) -> str: + return "".join( + inner.text + for inner in map(_unwrap_part, message.parts) + if isinstance(inner, TextPart) + ) + + +class _SendFailed: + """Sentinel enqueued when the background user-message send raises.""" class _QueueSubscriber(Subscriber): - """Bridges agent-server events into an asyncio queue for the SSE stream.""" + """Bridges agent-server events into an asyncio queue for the SSE stream. + + ``subscribe_to_events`` immediately replays the conversation's CURRENT + execution status to the new subscriber. When that status is terminal + (e.g. IDLE before a run has started), the replayed snapshot must NOT + close the stream — so the FIRST terminal state snapshot is swallowed + (exactly once, and only if it is the first state event we see). + """ def __init__(self, queue: asyncio.Queue): self._queue = queue + self._seen_state_event = False async def __call__(self, event: Any): + if ( + isinstance(event, ConversationStateUpdateEvent) + and not self._seen_state_event + ): + self._seen_state_event = True + value = str(getattr(event, "value", "") or "") + if value in _TERMINAL_EXECUTION_STATUSES: + logger.debug( + "A2A: dropping pre-run terminal state snapshot (%s)", value + ) + return await self._queue.put(event) @@ -451,8 +411,9 @@ async def __call__(self, event: Any): response_model_exclude_none=True, ) async def get_agent_card(request: Request) -> AgentCard: - """Return the A2A AgentCard v0.3 discovery document.""" + """Return the A2A AgentCard discovery document.""" base_url = str(request.base_url).rstrip("/") + input_modes: list[str] = ["text/plain"] return AgentCard( name="OpenHands Agent Server", description=( @@ -462,8 +423,14 @@ async def get_agent_card(request: Request) -> AgentCard: ), url=f"{base_url}/api/a2a", version=_server_version(), - capabilities=AgentCapabilities(streaming=True, pushNotifications=False), + capabilities=AgentCapabilities(streaming=True, push_notifications=False), + default_input_modes=input_modes, + default_output_modes=["text/plain"], skills=_agent_card_skills(), + provider=AgentProvider( + organization="OpenHands", + url="https://github.com/OpenHands/software-agent-sdk", + ), ) @@ -478,13 +445,24 @@ async def a2a_jsonrpc_endpoint( stream rather than a JSON envelope), ``tasks/get``, ``tasks/cancel``. JSON-RPC-level errors are returned as 200 responses carrying a JSON-RPC error object (``-32601`` method not found, ``-32700`` parse error, - ``-32602`` invalid params). + ``-32602`` invalid params). Every error response echoes the request + ``id`` (null for parse errors, per the JSON-RPC 2.0 spec). """ raw = await request.body() try: payload = json.loads(raw) except (json.JSONDecodeError, UnicodeDecodeError): - return _jsonrpc_error(JSONRPC_PARSE_ERROR, "Parse error") + # Raw JSONResponse (not the response_model) so that "id": null is + # serialized explicitly — exclude_none on JSONRPCResponse would drop + # the key, and JSON-RPC 2.0 requires a null id on parse errors. + return JSONResponse( + status_code=200, + content={ + "jsonrpc": "2.0", + "id": None, + "error": {"code": JSONRPC_PARSE_ERROR, "message": "Parse error"}, + }, + ) rpc_id = payload.get("id") if isinstance(payload, dict) else None if not isinstance(payload, dict) or payload.get("jsonrpc") != "2.0": @@ -498,28 +476,22 @@ async def a2a_jsonrpc_endpoint( JSONRPC_INVALID_REQUEST, "Invalid Request: missing method", rpc_id ) - if method == "message/send": + if method in ("message/send", "message/stream"): try: send_params = MessageSendParams.model_validate(params) except ValidationError as exc: return _jsonrpc_error( JSONRPC_INVALID_PARAMS, "Invalid params", rpc_id, data=str(exc) ) - handle_result = await _handle_message_send( - send_params, conversation_service, config - ) - if isinstance(handle_result, JSONRPCResponse): - return handle_result - return _jsonrpc(handle_result, rpc_id) - if method == "message/stream": - try: - send_params = MessageSendParams.model_validate(params) - except ValidationError as exc: - return _jsonrpc_error( - JSONRPC_INVALID_PARAMS, "Invalid params", rpc_id, data=str(exc) + if method == "message/send": + handle_result = await _handle_message_send( + send_params, conversation_service, config, rpc_id ) + if isinstance(handle_result, JSONRPCResponse): + return handle_result + return _jsonrpc(handle_result, rpc_id) return await _handle_message_stream( - send_params, conversation_service, rpc_id, config + send_params, conversation_service, config, rpc_id ) if method == "tasks/get": @@ -558,8 +530,8 @@ async def a2a_jsonrpc_endpoint( return _jsonrpc( Task( id=cancel_params.id, - contextId=cancel_params.id, - status=TaskStatus(state="canceled"), + context_id=cancel_params.id, + status=_task_status("canceled"), ), rpc_id, ) @@ -573,22 +545,25 @@ async def _start_or_get_conversation( send_params: MessageSendParams, conversation_service: ConversationService, config: Config, + rpc_id: str | int | None, ) -> tuple[uuid.UUID, EventService, bool] | JSONRPCResponse: """Return ``(task_id, event_service, created)`` or a JSON-RPC error. Reuses the conversation named by ``message.taskId`` when present, - otherwise starts a new one from the resolved agent profile. + otherwise starts a new one from the resolved agent profile. Every error + response echoes *rpc_id* so clients can correlate the failure. """ - if send_params.message.taskId: + if send_params.message.task_id: try: - task_id = _parse_task_id(send_params.message.taskId) + task_id = _parse_task_id(send_params.message.task_id) except ValueError as exc: - return _jsonrpc_error(JSONRPC_INVALID_PARAMS, str(exc)) + return _jsonrpc_error(JSONRPC_INVALID_PARAMS, str(exc), rpc_id) event_service = await _get_event_service_for(conversation_service, task_id) if event_service is None: return _jsonrpc_error( JSONRPC_TASK_NOT_FOUND, - f"Task not found: {send_params.message.taskId}", + f"Task not found: {send_params.message.task_id}", + rpc_id, ) return task_id, event_service, False @@ -598,6 +573,7 @@ async def _start_or_get_conversation( JSONRPC_INTERNAL_ERROR, "No agent profile configured; create one via /api/agent-profiles " "or pass agentProfileId in the params", + rpc_id, ) try: start_request = StartConversationRequest( @@ -607,12 +583,12 @@ async def _start_or_get_conversation( info, _created = await conversation_service.start_conversation(start_request) except ValueError as exc: return _jsonrpc_error( - JSONRPC_INVALID_PARAMS, f"Could not start conversation: {exc}" + JSONRPC_INVALID_PARAMS, f"Could not start conversation: {exc}", rpc_id ) event_service = await _get_event_service_for(conversation_service, info.id) if event_service is None: return _jsonrpc_error( - JSONRPC_INTERNAL_ERROR, f"Conversation {info.id} is not available" + JSONRPC_INTERNAL_ERROR, f"Conversation {info.id} is not available", rpc_id ) return info.id, event_service, True @@ -621,7 +597,7 @@ async def _send_user_message( send_params: MessageSendParams, event_service: EventService ) -> None: text = _user_text(send_params.message) - message = Message(role="user", content=[TextContent(text=text)]) + message = SDKMessage(role="user", content=[TextContent(text=text)]) await event_service.send_message(message, run=True) @@ -629,9 +605,10 @@ async def _handle_message_send( send_params: MessageSendParams, conversation_service: ConversationService, config: Config, + rpc_id: str | int | None, ) -> Task | JSONRPCResponse: started = await _start_or_get_conversation( - send_params, conversation_service, config + send_params, conversation_service, config, rpc_id ) if isinstance(started, JSONRPCResponse): return started @@ -656,22 +633,32 @@ def _state_update_to_status_update( if getattr(event, "key", None) != "execution_status": return None return TaskStatusUpdateEvent( - taskId=str(task_id), - contextId=str(task_id), - status=TaskStatus(state=_task_state_for_execution_status(event.value)), + task_id=str(task_id), + context_id=str(task_id), + status=_task_status(_task_state_for_execution_status(event.value)), final=final, ) +def _artifact_update(task_id: uuid.UUID, text: str, last_chunk: bool) -> Any: + parts: list[Part] = [Part(root=TextPart(text=text))] + return TaskArtifactUpdateEvent( + task_id=str(task_id), + context_id=str(task_id), + artifact=Artifact(artifact_id=str(uuid.uuid4()), parts=parts), + last_chunk=last_chunk, + ) + + async def _handle_message_stream( send_params: MessageSendParams, conversation_service: ConversationService, + config: Config, rpc_id: str | int | None, - config: Config | None = None, ) -> Any: """``message/stream``: run the task and stream A2A updates over SSE.""" started = await _start_or_get_conversation( - send_params, conversation_service, config or Config() + send_params, conversation_service, config, rpc_id ) if isinstance(started, JSONRPCResponse): return started @@ -680,6 +667,23 @@ async def _handle_message_stream( async def event_stream() -> AsyncIterator[str]: queue: asyncio.Queue = asyncio.Queue() subscriber = _QueueSubscriber(queue) + # Send the user message FIRST, then subscribe. Subscribing before + # sending races with subscribe_to_events' immediate replay of the + # current (pre-run, terminal) status, which used to close the stream + # before the run even started. The subscriber additionally swallows + # a replayed terminal snapshot defensively (see _QueueSubscriber). + send_task = asyncio.create_task(_send_user_message(send_params, event_service)) + + def _on_send_done(task: asyncio.Task) -> None: + if not task.cancelled() and task.exception() is not None: + queue.put_nowait(_SendFailed()) + + send_task.add_done_callback(_on_send_done) + # Let the send task take its first step before subscribing, so the + # user message is genuinely in flight (and mock-based tests observe + # send_message) when the subscriber's status replay happens. + await asyncio.sleep(0) + subscriber_id = await event_service.subscribe_to_events(subscriber) try: # Initial task snapshot; subsequent state updates arrive via the @@ -688,16 +692,24 @@ async def event_stream() -> AsyncIterator[str]: _jsonrpc( Task( id=str(task_id), - contextId=str(task_id), - status=TaskStatus(state="submitted"), + context_id=str(task_id), + status=_task_status("submitted"), ), rpc_id, ) ) - await _send_user_message(send_params, event_service) while True: event = await queue.get() + if isinstance(event, _SendFailed): + yield _sse_chunk( + _jsonrpc_error( + JSONRPC_INTERNAL_ERROR, + "Failed to send user message", + rpc_id, + ) + ) + return if isinstance(event, ConversationStateUpdateEvent): status_value = str(getattr(event, "value", "") or "") is_terminal = status_value in _TERMINAL_EXECUTION_STATUSES @@ -708,14 +720,19 @@ async def event_stream() -> AsyncIterator[str]: yield _sse_chunk(_jsonrpc(update, rpc_id)) if is_terminal: task = await _task_from_event_service(task_id, event_service) - for artifact in task.artifacts: + for artifact in task.artifacts or []: yield _sse_chunk( _jsonrpc( - TaskArtifactUpdateEvent( - taskId=str(task_id), - contextId=str(task_id), - artifact=artifact, - lastChunk=True, + _artifact_update( + task_id, + "".join( + inner.text + for inner in map( + _unwrap_part, artifact.parts + ) + if isinstance(inner, TextPart) + ), + last_chunk=True, ), rpc_id, ) @@ -725,9 +742,9 @@ async def event_stream() -> AsyncIterator[str]: else: # Surface agent message events as incremental artifacts. llm_message = getattr(event, "llm_message", None) - if llm_message is not None and getattr( - event, "source", None - ) == "agent": + if llm_message is not None and getattr(event, "source", None) == ( + "agent" + ): text = "".join( getattr(part, "text", "") for part in getattr(llm_message, "content", []) @@ -735,19 +752,14 @@ async def event_stream() -> AsyncIterator[str]: if text: yield _sse_chunk( _jsonrpc( - TaskArtifactUpdateEvent( - taskId=str(task_id), - contextId=str(task_id), - artifact=Artifact( - artifactId=str(uuid.uuid4()), - parts=[TextPart(text=text)], - ), - lastChunk=False, + _artifact_update( + task_id, text, last_chunk=False ), rpc_id, ) ) finally: + send_task.cancel() await event_service.unsubscribe_from_events(subscriber_id) return StreamingResponse(event_stream(), media_type=A2A_MEDIA_TYPE) diff --git a/openhands-agent-server/openhands/agent_server/api.py b/openhands-agent-server/openhands/agent_server/api.py index 7088cb2004..80d1f9eaf3 100644 --- a/openhands-agent-server/openhands/agent_server/api.py +++ b/openhands-agent-server/openhands/agent_server/api.py @@ -17,10 +17,6 @@ from fastapi.staticfiles import StaticFiles from starlette.requests import Request -from openhands.agent_server.a2a_router import ( - a2a_agent_card_router, - a2a_router, -) from openhands.agent_server.agent_profiles_router import agent_profiles_router from openhands.agent_server.auth_router import auth_router from openhands.agent_server.bash_router import bash_router @@ -407,6 +403,33 @@ def _find_http_exception(exc: BaseExceptionGroup) -> HTTPException | None: return None +def _load_a2a_routers(config: Config): + """Return the A2A routers, or None when they must not mount. + + Mounting requires BOTH explicit enablement (``config.a2a_enabled``, + default False — set via OH_A2A_ENABLED or --a2a) AND the optional + a2a-sdk dependency being importable (the 'a2a' extra). Anything else + logs and returns None so the server runs without the A2A endpoints. + """ + if not config.a2a_enabled: + return None + try: + from openhands.agent_server.a2a_router import ( + a2a_agent_card_router, + a2a_router, + ) + except Exception as exc: + logger.warning( + "a2a_enabled is True but the a2a-sdk dependency is not importable " + "(%s). Install the 'a2a' extra (pip install " + "openhands-agent-server[a2a]) to expose the A2A endpoints; " + "they will NOT be mounted.", + exc, + ) + return None + return a2a_router, a2a_agent_card_router + + def _add_api_routes(app: FastAPI) -> None: """Add all API routes to the FastAPI application.""" app.include_router(server_details_router) @@ -456,15 +479,20 @@ def _add_api_routes(app: FastAPI) -> None: # /api/auth/* mints workspace cookies and requires the header to bootstrap, # so it lives under the header-only auth group. api_router.include_router(auth_router) - # A2A JSON-RPC endpoint. Mounted with its own dependencies (not the - # header-only group above) because A2A clients authenticate with the + # A2A JSON-RPC endpoint, OFF BY DEFAULT: mounts only when + # config.a2a_enabled is True AND the optional a2a-sdk dependency + # ('a2a' extra) is importable. Mounted with its own dependencies (not + # the header-only group above) because A2A clients authenticate with the # standard Authorization header rather than ``X-Session-API-Key``; the # A2A auth dependency accepts either. - api_router.include_router(a2a_router) - app.include_router(api_router) + a2a_routers = _load_a2a_routers(app.state.config) + if a2a_routers is not None: + a2a_router, a2a_agent_card_router = a2a_routers + api_router.include_router(a2a_router) + # A2A discovery: well-known URIs live at the app root, outside /api. + app.include_router(a2a_agent_card_router) - # A2A discovery: well-known URIs must live at the app root, outside /api. - app.include_router(a2a_agent_card_router) + app.include_router(api_router) app.include_router(openai_router, dependencies=[Depends(check_openai_api_key)]) diff --git a/openhands-agent-server/openhands/agent_server/config.py b/openhands-agent-server/openhands/agent_server/config.py index 65d8632fb9..92db692b8b 100644 --- a/openhands-agent-server/openhands/agent_server/config.py +++ b/openhands-agent-server/openhands/agent_server/config.py @@ -322,6 +322,16 @@ class Config(BaseModel): default=False, description="Whether to enable VNC desktop functionality", ) + a2a_enabled: bool = Field( + default=False, + description=( + "Expose the A2A (Agent2Agent) protocol endpoints " + "(POST /api/a2a JSON-RPC + /.well-known/agent-card.json). " + "Disabled by default; additionally requires the optional " + "'a2a' extra (a2a-sdk) to be installed — without it the routes " + "do not mount. Set via OH_A2A_ENABLED or the --a2a CLI flag." + ), + ) preload_tools: bool = Field( default=True, description="Whether to preload tools", diff --git a/openhands-agent-server/pyproject.toml b/openhands-agent-server/pyproject.toml index 6a74a869cc..d4817cf8d7 100644 --- a/openhands-agent-server/pyproject.toml +++ b/openhands-agent-server/pyproject.toml @@ -24,6 +24,10 @@ dependencies = [ # on the import path. Pinned to the major already resolved in uv.lock (6.7.7, # via browser-use) so installing the extra cannot force a resolver conflict. posthog = ["posthog>=6,<7"] +# A2A (Agent2Agent) protocol support. Optional on purpose: the JSON-RPC +# endpoint and agent-card route only mount when BOTH this extra is installed +# AND a2a_enabled is turned on; the server runs fine without it. +a2a = ["a2a-sdk>=0.3.9,<0.4"] [project.urls] Source = "https://github.com/OpenHands/software-agent-sdk" diff --git a/tests/agent_server/test_a2a_router.py b/tests/agent_server/test_a2a_router.py index be495a2730..69fbb4c565 100644 --- a/tests/agent_server/test_a2a_router.py +++ b/tests/agent_server/test_a2a_router.py @@ -3,6 +3,9 @@ Builds the A2A routers the same way test_conversation_router.py builds the conversation router: a bare FastAPI app with a mocked ConversationService / EventService injected via dependency_overrides. + +Enablement (mount-on-demand) behavior is covered separately at the bottom via +``openhands.agent_server.api.create_app``. """ import json @@ -131,7 +134,6 @@ def test_agent_card_at_well_known_root(self, client): assert card["skills"] == [] assert card["provider"]["organization"] == "OpenHands" assert card["url"].endswith("/api/a2a") - assert card["protocolVersion"] == "0.3.0" def test_agent_card_requires_no_auth(self, authed_client): # Discovery is unauthenticated even when session keys are configured. @@ -184,7 +186,7 @@ def test_session_header_accepted(self, authed_client, mock_conversation_service) # --------------------------------------------------------------------------- -# JSON-RPC errors +# JSON-RPC errors (every error path must echo the request id) # --------------------------------------------------------------------------- @@ -203,6 +205,8 @@ def test_parse_error(self, client, mock_conversation_service): body = response.json() assert body["jsonrpc"] == "2.0" assert body["error"]["code"] == -32700 + # JSON-RPC 2.0 spec: parse errors echo a null id. + assert body["id"] is None finally: client.app.dependency_overrides.clear() @@ -227,9 +231,13 @@ def test_invalid_params_tasks_get(self, client, mock_conversation_service): mock_conversation_service ) try: - response = client.post("/api/a2a", json=_jsonrpc_body("tasks/get")) + response = client.post( + "/api/a2a", json=_jsonrpc_body("tasks/get", rpc_id="err-1") + ) assert response.status_code == 200 - assert response.json()["error"]["code"] == -32602 + body = response.json() + assert body["id"] == "err-1" + assert body["error"]["code"] == -32602 finally: client.app.dependency_overrides.clear() @@ -240,10 +248,14 @@ def test_invalid_params_bad_task_id(self, client, mock_conversation_service): try: response = client.post( "/api/a2a", - json=_jsonrpc_body("tasks/get", params={"id": "not-a-uuid"}), + json=_jsonrpc_body( + "tasks/get", params={"id": "not-a-uuid"}, rpc_id="err-2" + ), ) assert response.status_code == 200 - assert response.json()["error"]["code"] == -32602 + body = response.json() + assert body["id"] == "err-2" + assert body["error"]["code"] == -32602 finally: client.app.dependency_overrides.clear() @@ -253,10 +265,97 @@ def test_invalid_params_message_send(self, client, mock_conversation_service): ) try: response = client.post( - "/api/a2a", json=_jsonrpc_body("message/send", params={"message": {}}) + "/api/a2a", + json=_jsonrpc_body( + "message/send", params={"message": {}}, rpc_id="err-3" + ), ) assert response.status_code == 200 - assert response.json()["error"]["code"] == -32602 + body = response.json() + assert body["id"] == "err-3" + assert body["error"]["code"] == -32602 + finally: + client.app.dependency_overrides.clear() + + def test_invalid_params_message_stream(self, client, mock_conversation_service): + client.app.dependency_overrides[get_conversation_service] = lambda: ( + mock_conversation_service + ) + try: + response = client.post( + "/api/a2a", + json=_jsonrpc_body( + "message/stream", params={"message": {}}, rpc_id="err-4" + ), + ) + assert response.status_code == 200 + body = response.json() + assert body["id"] == "err-4" + assert body["error"]["code"] == -32602 + finally: + client.app.dependency_overrides.clear() + + def test_task_not_found_preserves_id(self, client, mock_conversation_service): + mock_conversation_service.get_event_service.return_value = None + client.app.dependency_overrides[get_conversation_service] = lambda: ( + mock_conversation_service + ) + try: + response = client.post( + "/api/a2a", + json=_jsonrpc_body( + "tasks/get", params={"id": str(uuid4())}, rpc_id="nf-1" + ), + ) + assert response.status_code == 200 + body = response.json() + assert body["id"] == "nf-1" + assert body["error"]["code"] == -32001 + finally: + client.app.dependency_overrides.clear() + + def test_task_not_found_send_preserves_id( + self, client, mock_conversation_service + ): + # message/send against a bogus taskId must echo the rpc id too. + mock_conversation_service.get_event_service.return_value = None + client.app.dependency_overrides[get_conversation_service] = lambda: ( + mock_conversation_service + ) + try: + response = client.post( + "/api/a2a", + json=_jsonrpc_body( + "message/send", + params=_send_params(task_id=str(uuid4())), + rpc_id="nf-2", + ), + ) + assert response.status_code == 200 + body = response.json() + assert body["id"] == "nf-2" + assert body["error"]["code"] == -32001 + finally: + client.app.dependency_overrides.clear() + + def test_no_profile_error_preserves_id( + self, client, mock_conversation_service, mock_event_service, monkeypatch + ): + from openhands.agent_server import a2a_router + + monkeypatch.setattr(a2a_router, "_resolve_agent_profile_id", lambda: None) + _override(client, mock_conversation_service, mock_event_service) + try: + response = client.post( + "/api/a2a", + json=_jsonrpc_body( + "message/send", params=_send_params(), rpc_id="np-1" + ), + ) + assert response.status_code == 200 + body = response.json() + assert body["id"] == "np-1" + assert body["error"]["code"] == -32603 finally: client.app.dependency_overrides.clear() @@ -456,11 +555,15 @@ def test_tasks_cancel_not_found( response = client.post( "/api/a2a", json=_jsonrpc_body( - "tasks/cancel", params={"id": str(sample_conversation_id)} + "tasks/cancel", + params={"id": str(sample_conversation_id)}, + rpc_id="cnf-1", ), ) assert response.status_code == 200 - assert response.json()["error"]["code"] == -32001 + body = response.json() + assert body["id"] == "cnf-1" + assert body["error"]["code"] == -32001 finally: client.app.dependency_overrides.clear() @@ -538,3 +641,171 @@ async def fake_subscribe(subscriber): mock_event_service.send_message.assert_called_once() finally: client.app.dependency_overrides.clear() + + def test_stream_ignores_initial_idle_snapshot( + self, + client, + mock_conversation_service, + mock_event_service, + sample_conversation_info, + monkeypatch, + ): + """Regression test for the message/stream IDLE race. + + ``subscribe_to_events`` immediately replays the conversation's + CURRENT status to a new subscriber. If that snapshot is terminal + (e.g. IDLE before the run has started) the stream must NOT close on + it; it must keep going and deliver the real terminal state and the + final artifact. + """ + from openhands.agent_server import a2a_router + from openhands.sdk.event.conversation_state import ConversationStateUpdateEvent + + monkeypatch.setattr( + a2a_router, "_resolve_agent_profile_id", lambda: str(uuid4()) + ) + mock_conversation_service.start_conversation.return_value = ( + sample_conversation_info, + True, + ) + mock_event_service.get_state.return_value = MagicMock( + execution_status=ConversationExecutionStatus.IDLE + ) + mock_event_service.get_agent_final_response.return_value = "raced reply" + + async def fake_subscribe(subscriber): + # Replay the pre-run IDLE snapshot FIRST — this used to close the + # stream before the run started. + await subscriber( + ConversationStateUpdateEvent(key="execution_status", value="idle") + ) + await subscriber( + ConversationStateUpdateEvent(key="execution_status", value="running") + ) + await subscriber( + ConversationStateUpdateEvent(key="execution_status", value="idle") + ) + return uuid4() + + mock_event_service.subscribe_to_events.side_effect = fake_subscribe + _override(client, mock_conversation_service, mock_event_service) + + try: + with client.stream( + "POST", + "/api/a2a", + json=_jsonrpc_body( + "message/stream", params=_send_params("race"), rpc_id="race-1" + ), + ) as response: + assert response.status_code == 200 + events = [] + for line in response.iter_lines(): + if line.startswith("data: "): + events.append(json.loads(line[len("data: ") :])) + states = [ + e["result"]["status"]["state"] + for e in events + if e["result"]["kind"] == "status-update" + ] + # The pre-run IDLE snapshot must not leak out as a completed / + # final update, and must not terminate the stream early. + assert states[0] != "completed" + terminal_tasks = [e for e in events if e["result"]["kind"] == "task"] + assert terminal_tasks, "stream never delivered the terminal Task" + final = terminal_tasks[-1]["result"] + assert final["status"]["state"] == "completed" + assert final["artifacts"][0]["parts"][0]["text"] == "raced reply" + final_updates = [e for e in events if e["result"].get("final") is True] + assert len(final_updates) == 1, "exactly one final status update" + assert final_updates[0]["id"] == "race-1" + mock_event_service.send_message.assert_called_once() + finally: + client.app.dependency_overrides.clear() + + +# --------------------------------------------------------------------------- +# Enablement: mounted only when a2a_enabled AND a2a-sdk importable +# --------------------------------------------------------------------------- + + +class TestEnablement: + def test_disabled_by_default_404(self, tmp_path): + from openhands.agent_server.api import create_app + + config = Config(static_files_path=None, secret_key=None) + assert config.a2a_enabled is False + app = create_app(config) + client = TestClient(app) + assert client.post( + "/api/a2a", json=_jsonrpc_body("tasks/get") + ).status_code == 404 + assert client.get("/.well-known/agent-card.json").status_code == 404 + + def test_enabled_mounts_routes(self, tmp_path): + from unittest.mock import AsyncMock + + from openhands.agent_server.api import create_app + from openhands.agent_server.dependencies import get_conversation_service + + mock_conversation_service = AsyncMock(spec=ConversationService) + mock_conversation_service.get_event_service.return_value = None + + config = Config( + static_files_path=None, + secret_key=None, + workspace_path=tmp_path, + a2a_enabled=True, + ) + app = create_app(config) + app.dependency_overrides[get_conversation_service] = lambda: ( + mock_conversation_service + ) + client = TestClient(app) + try: + # No auth keys configured: an unauthenticated tasks/get for a + # random id reaches the JSON-RPC handler (200 + task-not-found). + response = client.post( + "/api/a2a", + json=_jsonrpc_body("tasks/get", params={"id": str(uuid4())}), + ) + assert response.status_code == 200 + assert response.json()["error"]["code"] == -32001 + card = client.get("/.well-known/agent-card.json") + assert card.status_code == 200 + assert card.json()["name"] == "OpenHands Agent Server" + finally: + app.dependency_overrides.clear() + + def test_enabled_without_sdk_does_not_mount(self, tmp_path, monkeypatch): + import builtins + + from openhands.agent_server import api as api_module + + real_import = builtins.__import__ + + def fake_import(name, *args, **kwargs): + if name == "openhands.agent_server.a2a_router" or name.startswith("a2a"): + raise ImportError("No module named 'a2a' (simulated)") + return real_import(name, *args, **kwargs) + + monkeypatch.setattr(builtins, "__import__", fake_import) + config = Config( + static_files_path=None, + secret_key=None, + a2a_enabled=True, + ) + app = api_module.create_app(config) + client = TestClient(app) + assert client.post( + "/api/a2a", json=_jsonrpc_body("tasks/get") + ).status_code == 404 + assert client.get("/.well-known/agent-card.json").status_code == 404 + + def test_env_var_enables(self, monkeypatch, tmp_path): + monkeypatch.setenv("OH_A2A_ENABLED", "true") + from openhands.agent_server.api import create_app + from openhands.agent_server.config import load_config + + loaded = load_config(tmp_path / "nonexistent.json") + assert loaded.a2a_enabled is True diff --git a/uv.lock b/uv.lock index 6b70a81f58..03f44827ba 100644 --- a/uv.lock +++ b/uv.lock @@ -8,7 +8,7 @@ resolution-markers = [ ] [options] -exclude-newer = "0001-01-01T00:00:00Z" # This has no effect and is included for backwards compatibility when using relative exclude-newer values. +exclude-newer = "2026-08-19T02:45:27.136128569Z" exclude-newer-span = "P7D" [options.exclude-newer-package] @@ -54,6 +54,23 @@ dev = [ { name = "tabulate", specifier = ">=0.9.0" }, ] +[[package]] +name = "a2a-sdk" +version = "0.3.26" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "google-api-core", version = "2.26.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.13'" }, + { name = "google-api-core", version = "2.31.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.13'" }, + { name = "httpx" }, + { name = "httpx-sse" }, + { name = "protobuf" }, + { name = "pydantic" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/be/97/a6840e01795b182ce751ca165430d46459927cde9bfab838087cbb24aef7/a2a_sdk-0.3.26.tar.gz", hash = "sha256:44068e2d037afbb07ab899267439e9bc7eaa7ac2af94f1e8b239933c993ad52d", size = 274598, upload-time = "2026-04-09T15:21:13.902Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/dd/d5/51f4ee1bf3b736add42a542d3c8a3fd3fa85f3d36c17972127defc46c26f/a2a_sdk-0.3.26-py3-none-any.whl", hash = "sha256:754e0573f6d33b225c1d8d51f640efa69cbbed7bdfb06ce9c3540ea9f58d4a91", size = 151016, upload-time = "2026-04-09T15:21:12.35Z" }, +] + [[package]] name = "agent-client-protocol" version = "0.10.1" @@ -1256,11 +1273,11 @@ resolution-markers = [ "python_full_version < '3.13'", ] dependencies = [ - { name = "google-auth" }, - { name = "googleapis-common-protos" }, - { name = "proto-plus" }, - { name = "protobuf" }, - { name = "requests" }, + { name = "google-auth", marker = "python_full_version < '3.13'" }, + { name = "googleapis-common-protos", marker = "python_full_version < '3.13'" }, + { name = "proto-plus", marker = "python_full_version < '3.13'" }, + { name = "protobuf", marker = "python_full_version < '3.13'" }, + { name = "requests", marker = "python_full_version < '3.13'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/32/ea/e7b6ac3c7b557b728c2d0181010548cbbdd338e9002513420c5a354fa8df/google_api_core-2.26.0.tar.gz", hash = "sha256:e6e6d78bd6cf757f4aee41dcc85b07f485fbb069d5daa3afb126defba1e91a62", size = 166369, upload-time = "2025-10-08T21:37:38.39Z" } wheels = [ @@ -1269,8 +1286,8 @@ wheels = [ [package.optional-dependencies] grpc = [ - { name = "grpcio" }, - { name = "grpcio-status" }, + { name = "grpcio", marker = "python_full_version < '3.13'" }, + { name = "grpcio-status", marker = "python_full_version < '3.13'" }, ] [[package]] @@ -1282,11 +1299,11 @@ resolution-markers = [ "python_full_version == '3.13.*'", ] dependencies = [ - { name = "google-auth" }, - { name = "googleapis-common-protos" }, - { name = "proto-plus" }, - { name = "protobuf" }, - { name = "requests" }, + { name = "google-auth", marker = "python_full_version >= '3.13'" }, + { name = "googleapis-common-protos", marker = "python_full_version >= '3.13'" }, + { name = "proto-plus", marker = "python_full_version >= '3.13'" }, + { name = "protobuf", marker = "python_full_version >= '3.13'" }, + { name = "requests", marker = "python_full_version >= '3.13'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/c6/22/155cadf1d49272a9cf48f3168c0f3874fa13397297e611a5ea00cd093880/google_api_core-2.31.0.tar.gz", hash = "sha256:2be84ee0f584c48e6bde1b36766e23348b361fb7e55e56135fc76ce1c397f9c2", size = 176492, upload-time = "2026-06-03T14:52:17.257Z" } wheels = [ @@ -1295,8 +1312,8 @@ wheels = [ [package.optional-dependencies] grpc = [ - { name = "grpcio" }, - { name = "grpcio-status" }, + { name = "grpcio", marker = "python_full_version >= '3.13'" }, + { name = "grpcio-status", marker = "python_full_version >= '3.13'" }, ] [[package]] @@ -1445,12 +1462,12 @@ resolution-markers = [ "python_full_version < '3.13'", ] dependencies = [ - { name = "google-api-core", version = "2.26.0", source = { registry = "https://pypi.org/simple" } }, - { name = "google-auth" }, - { name = "google-cloud-core" }, - { name = "google-crc32c" }, - { name = "google-resumable-media" }, - { name = "requests" }, + { name = "google-api-core", version = "2.26.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version < '3.13'" }, + { name = "google-auth", marker = "python_full_version < '3.13'" }, + { name = "google-cloud-core", marker = "python_full_version < '3.13'" }, + { name = "google-crc32c", marker = "python_full_version < '3.13'" }, + { name = "google-resumable-media", marker = "python_full_version < '3.13'" }, + { name = "requests", marker = "python_full_version < '3.13'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/bd/ef/7cefdca67a6c8b3af0ec38612f9e78e5a9f6179dd91352772ae1a9849246/google_cloud_storage-3.4.1.tar.gz", hash = "sha256:6f041a297e23a4b485fad8c305a7a6e6831855c208bcbe74d00332a909f82268", size = 17238203, upload-time = "2025-10-08T18:43:39.665Z" } wheels = [ @@ -1466,12 +1483,12 @@ resolution-markers = [ "python_full_version == '3.13.*'", ] dependencies = [ - { name = "google-api-core", version = "2.31.0", source = { registry = "https://pypi.org/simple" } }, - { name = "google-auth" }, - { name = "google-cloud-core" }, - { name = "google-crc32c" }, - { name = "google-resumable-media" }, - { name = "requests" }, + { name = "google-api-core", version = "2.31.0", source = { registry = "https://pypi.org/simple" }, marker = "python_full_version >= '3.13'" }, + { name = "google-auth", marker = "python_full_version >= '3.13'" }, + { name = "google-cloud-core", marker = "python_full_version >= '3.13'" }, + { name = "google-crc32c", marker = "python_full_version >= '3.13'" }, + { name = "google-resumable-media", marker = "python_full_version >= '3.13'" }, + { name = "requests", marker = "python_full_version >= '3.13'" }, ] sdist = { url = "https://files.pythonhosted.org/packages/22/09/8953e2993e604c8882fd441b5b2de624a2dfe7e6144c6166d7b477509596/google_cloud_storage-3.11.0.tar.gz", hash = "sha256:498bf37c999028f69a245f586b5e50d89f59df1fafc0e3a93783ac56be2a456b", size = 17335639, upload-time = "2026-06-03T16:14:04.649Z" } wheels = [ @@ -2736,12 +2753,16 @@ dependencies = [ ] [package.optional-dependencies] +a2a = [ + { name = "a2a-sdk" }, +] posthog = [ { name = "posthog" }, ] [package.metadata] requires-dist = [ + { name = "a2a-sdk", marker = "extra == 'a2a'", specifier = ">=0.3.9,<0.4" }, { name = "aiosqlite", specifier = ">=0.19" }, { name = "alembic", specifier = ">=1.13" }, { name = "docker", specifier = ">=7.1,<8" }, @@ -2755,7 +2776,7 @@ requires-dist = [ { name = "websockets", specifier = ">=12" }, { name = "wsproto", specifier = ">=1.2.0" }, ] -provides-extras = ["posthog"] +provides-extras = ["posthog", "a2a"] [[package]] name = "openhands-sdk"