feat(agent-server): A2A protocol server-mode support (Agent Card + JSON-RPC/SSE)

.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")

.pr/design.html

@@ -0,0 +1,43 @@
+<!DOCTYPE html>
+<html><head><meta charset="utf-8"><title>A2A server-mode — design</title>
+<style>body{font-family:system-ui,sans-serif;max-width:860px;margin:2rem auto;padding:0 1rem;line-height:1.5}h1{font-size:1.5rem}h2{margin-top:2rem;font-size:1.15rem}table{border-collapse:collapse;width:100%;margin:1rem 0}td,th{border:1px solid #ccc;padding:.4rem .6rem;text-align:left;font-size:.9rem}code,pre{background:#f4f4f4;padding:.1rem .3rem;font-size:.85rem}pre{padding:.8rem;overflow-x:auto}</style></head><body>
+<h1>A2A server-mode support — design</h1>
+<p>PR #4590 · implements <a href="https://github.com/OpenHands/software-agent-sdk/issues/1060">#1060</a> · Agent Card v0.3 + JSON-RPC 2.0 transport (Linux Foundation a2a-spec)</p>
+
+<h2>Goal</h2>
+<p>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.</p>
+
+<h2>Object model</h2>
+<p>A2A objects (AgentCard, Task, TaskStatus, TextPart, JSON-RPC envelope) are minimal local pydantic models in <code>a2a_router.py</code>. No <code>a2a-sdk</code> dependency; the module docstring pins the targeted spec revision so a future swap to the SDK (or optional extra) is mechanical.</p>
+
+<h2>Endpoint mapping</h2>
+<table><tr><th>A2A</th><th>agent-server internals</th></tr>
+<tr><td><code>GET /.well-known/agent-card.json</code></td><td>Server config + <code>agent_profiles_router</code> metadata → Agent Card v0.3 (capabilities: streaming)</td></tr>
+<tr><td><code>message/send</code></td><td><code>conversation_service.start_conversation()</code> → <code>event_service.send_message(Message(role='user'), run=True)</code>; taskId = conversationId</td></tr>
+<tr><td><code>message/stream</code></td><td>Same as send, then SSE via <code>subscribe_to_events()</code> → <code>unsubscribe_from_events()</code> on terminal state</td></tr>
+<tr><td><code>tasks/get</code></td><td>Execution status + <code>get_agent_final_response()</code> → TaskStatus + text artifact</td></tr>
+<tr><td><code>tasks/cancel</code></td><td>Conversation interrupt/pause</td></tr></table>
+
+<h2>Before / after</h2>
+<p><b>Before:</b> agent-server reachable only via its native REST/WebSocket API; A2A clients (Hermes, Google ADK, LangGraph, etc.) cannot discover or drive it.</p>
+<p><b>After:</b> Agent Card advertises the server on any A2A mesh; the standard JSON-RPC lifecycle works end-to-end:
+<pre>$ 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":"&lt;conversationId&gt;","status":{"state":"submitted"},...}}</pre></p>
+
+<h2>SSE event sequence (message/stream)</h2>
+<pre>data: {"kind":"task", ...snapshot...}        # initial task snapshot
+data: {"kind":"status-update", ...}          # working / input-required
+data: {"kind":"task", ...with artifacts...}   # final state + artifact, then close</pre>
+
+<h2>Auth</h2>
+<p>Single dependency accepts either header — <code>X-Session-API-Key</code> (existing convention) or <code>Authorization: Bearer &lt;key&gt;</code> (A2A convention) — both validated against <code>config.session_api_keys</code>. The Agent Card endpoint is unauthenticated (discovery).
+
+<h2>REST compatibility</h2>
+<p>Additive only: one new router file + a 12-line mount in <code>api.py</code> (root well-known route + <code>/api/a2a</code> under the existing api_router). No existing router, model, or contract is modified; OpenAPI quality gate unchanged (97 allowlisted weak locations).</p>
+
+<h2>Open question</h2>
+<p>Keep zero-dependency pydantic models, or adopt <code>a2a-sdk</code> (possibly as an optional extra <code>[a2a]</code>)? Implementer leans zero-dep core; maintainer guidance requested in #1060.</p>
+</body></html>