feat(a2a): migrate runtime integration to a2a-sdk v1

pyproject.toml

@@ -154,13 +154,13 @@ dev = [
     "langchain>=1.0.0",
     "langgraph>=1.0.0",
     "langchain-mcp-adapters>=0.1.0",
-    "a2a-sdk[http-server]>=0.3,<1.0",
+    "a2a-sdk[http-server]>=1.0.1,<2.0",
     "ag-ui-protocol>=0.1.10",
     "mcp-proxy-for-aws>=0.1.0",
 ]
 
 [project.optional-dependencies]
-a2a = ["a2a-sdk[http-server]>=0.3,<1.0"]
+a2a = ["a2a-sdk[http-server]>=1.0.1,<2.0"]
 ag-ui = ["ag-ui-protocol>=0.1.10"]
 strands-agents = [
     "strands-agents>=1.20.0",

src/bedrock_agentcore/runtime/a2a.py

@@ -39,12 +39,12 @@ def _check_a2a_sdk() -> None:
 
 
 def _build_agent_card(executor: Any, url: str) -> Any:
-    """Build an AgentCard by introspecting a StrandsA2AExecutor.
+    """Build an AgentCard by introspecting an executor.
 
     Extracts name/description from ``executor.agent``. Falls back to generic
     defaults for other executors.
     """
-    from a2a.types import AgentCapabilities, AgentCard, AgentSkill
+    from a2a.types import AgentCapabilities, AgentCard, AgentInterface, AgentSkill
 
     name = "agent"
     description = "A Bedrock AgentCore agent"
@@ -57,12 +57,36 @@ def _build_agent_card(executor: Any, url: str) -> Any:
     return AgentCard(
         name=name,
         description=description,
-        url=url,
         version="0.1.0",
         capabilities=AgentCapabilities(streaming=True),
         skills=[AgentSkill(id="main", name=name, description=description, tags=["main"])],
         default_input_modes=["text"],
         default_output_modes=["text"],
+        supported_interfaces=[
+            AgentInterface(
+                protocol_binding="JSONRPC",
+                protocol_version="1.0",
+                url=url,
+            )
+        ],
+    )
+
+
+def _set_jsonrpc_url(agent_card: Any, url: str) -> None:
+    """Set the runtime URL on the card's JSON-RPC interface."""
+    from a2a.types import AgentInterface
+
+    for interface in agent_card.supported_interfaces:
+        if interface.protocol_binding == "JSONRPC":
+            interface.url = url
+            return
+
+    agent_card.supported_interfaces.append(
+        AgentInterface(
+            protocol_binding="JSONRPC",
+            protocol_version="1.0",
+            url=url,
+        )
     )
 
 
@@ -97,7 +121,7 @@ def build_runtime_url(agent_arn: str, region: Optional[str] = None) -> str:
 class BedrockCallContextBuilder:
     """Extracts Bedrock runtime headers and propagates them into BedrockAgentCoreContext.
 
-    Implements the a2a-sdk CallContextBuilder ABC so the A2A server
+    Implements the a2a-sdk ServerCallContextBuilder ABC so the A2A server
     automatically calls ``build()`` on every incoming request.
     """
 
@@ -160,6 +184,8 @@ def build(self, request: Any) -> Any:
         _ensure_baggage_processor_registered()
 
         state = {
+            "headers": dict(headers),
+            "bedrock_request_id": request_id,
             "request_id": request_id,
             "session_id": session_id,
         }
@@ -174,9 +200,9 @@ def build(self, request: Any) -> Any:
 # Register as a virtual subclass so isinstance checks pass without
 # requiring a2a-sdk to be importable at class-definition time.
 try:
-    from a2a.server.apps import CallContextBuilder
+    from a2a.server.routes import ServerCallContextBuilder
 
-    CallContextBuilder.register(BedrockCallContextBuilder)
+    ServerCallContextBuilder.register(BedrockCallContextBuilder)
 except Exception:  # pragma: no cover
     pass
 
@@ -196,7 +222,7 @@ def build_a2a_app(
         agent_card: Optional ``a2a.types.AgentCard`` describing the agent.
             If ``None``, one is built automatically by introspecting the executor.
         task_store: Optional ``TaskStore``; defaults to ``InMemoryTaskStore``.
-        context_builder: Optional ``CallContextBuilder``; defaults to
+        context_builder: Optional ``ServerCallContextBuilder``; defaults to
             ``BedrockCallContextBuilder``.
         ping_handler: Optional callback returning a ``PingStatus``.
 
@@ -207,8 +233,8 @@ def build_a2a_app(
 
     _check_a2a_sdk()
 
-    from a2a.server.apps import A2AStarletteApplication
     from a2a.server.request_handlers import DefaultRequestHandler
+    from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
     from a2a.server.tasks import InMemoryTaskStore
     from starlette.applications import Starlette
     from starlette.responses import JSONResponse
@@ -219,7 +245,7 @@ def build_a2a_app(
     if agent_card is None:
         agent_card = _build_agent_card(executor, runtime_url)
     elif os.environ.get(AGENTCORE_RUNTIME_URL_ENV):
-        agent_card.url = runtime_url
+        _set_jsonrpc_url(agent_card, runtime_url)
 
     if task_store is None:
         task_store = InMemoryTaskStore()
@@ -229,12 +255,17 @@ def build_a2a_app(
     http_handler = DefaultRequestHandler(
         agent_executor=executor,
         task_store=task_store,
+        agent_card=agent_card,
     )
 
-    a2a_app = A2AStarletteApplication(
-        agent_card=agent_card,
-        http_handler=http_handler,
-        context_builder=context_builder,
+    routes = create_agent_card_routes(agent_card)
+    routes.extend(
+        create_jsonrpc_routes(
+            request_handler=http_handler,
+            rpc_url="/",
+            context_builder=context_builder,
+            enable_v0_3_compat=True,
+        )
     )
 
     def _handle_ping(request: Any) -> JSONResponse:
@@ -248,12 +279,8 @@ def _handle_ping(request: Any) -> JSONResponse:
             status = PingStatus.HEALTHY
         return JSONResponse({"status": status.value})
 
-    # Build the Starlette app with /ping included upfront, then add A2A routes,
-    # so we don't depend on mutating app.routes after build().
-    app = Starlette(routes=[Route("/ping", _handle_ping, methods=["GET"])])
-    a2a_app.add_routes_to_app(app)
-
-    return app
+    routes.insert(0, Route("/ping", _handle_ping, methods=["GET"]))
+    return Starlette(routes=routes)
 
 
 def serve_a2a(
@@ -276,7 +303,7 @@ def serve_a2a(
         port: Port to serve on (default 9000).
         host: Host to bind to; auto-detected if ``None``.
         task_store: Optional ``TaskStore``; defaults to ``InMemoryTaskStore``.
-        context_builder: Optional ``CallContextBuilder``; defaults to
+        context_builder: Optional ``ServerCallContextBuilder``; defaults to
             ``BedrockCallContextBuilder``.
         ping_handler: Optional callback returning a ``PingStatus``.
         **kwargs: Additional arguments forwarded to ``uvicorn.run()``.

tests/bedrock_agentcore/runtime/test_a2a.py

@@ -2,11 +2,11 @@
 import uuid
 from unittest.mock import patch
 
+from a2a.helpers import new_task_from_user_message
 from a2a.server.agent_execution import AgentExecutor, RequestContext
 from a2a.server.events import EventQueue
 from a2a.server.tasks import InMemoryTaskStore, TaskUpdater
-from a2a.types import AgentCapabilities, AgentCard, AgentSkill, Part, TextPart
-from a2a.utils import new_task
+from a2a.types import AgentCapabilities, AgentCard, AgentInterface, AgentSkill, Part
 from starlette.testclient import TestClient
 
 from bedrock_agentcore.runtime.a2a import (
@@ -27,12 +27,12 @@ def __init__(self):
 
     async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
         self.last_call_context = context.call_context
-        task = context.current_task or new_task(context.message)
+        task = context.current_task or new_task_from_user_message(context.message)
         if not context.current_task:
             await event_queue.enqueue_event(task)
         updater = TaskUpdater(event_queue, task.id, task.context_id)
         user_text = context.get_user_input()
-        await updater.add_artifact([Part(root=TextPart(text=f"echo: {user_text}"))])
+        await updater.add_artifact([Part(text=f"echo: {user_text}")])
         await updater.complete()
 
     async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
@@ -43,12 +43,18 @@ def _make_agent_card() -> AgentCard:
     return AgentCard(
         name="test-agent",
         description="A test agent",
-        url="http://localhost:9000",
         version="1.0.0",
         capabilities=AgentCapabilities(streaming=True),
         skills=[AgentSkill(id="echo", name="echo", description="Echoes input", tags=["echo"])],
         default_input_modes=["text"],
         default_output_modes=["text"],
+        supported_interfaces=[
+            AgentInterface(
+                protocol_binding="JSONRPC",
+                protocol_version="1.0",
+                url="http://localhost:9000",
+            )
+        ],
     )
 
 
@@ -62,9 +68,9 @@ def _jsonrpc_request(method: str, params: dict | None = None) -> dict:
 def _send_message_params(text: str = "hello") -> dict:
     return {
         "message": {
-            "message_id": str(uuid.uuid4()),
-            "role": "user",
-            "parts": [{"kind": "text", "text": text}],
+            "messageId": str(uuid.uuid4()),
+            "role": "ROLE_USER",
+            "parts": [{"text": text}],
         }
     }
 
@@ -117,47 +123,64 @@ def test_agent_card_returns_card_data(self):
     def test_message_send_executes_and_returns_completed_task(self):
         """Verify the full RPC path: JSON-RPC request -> executor -> completed task."""
         app = build_a2a_app(_EchoExecutor(), _make_agent_card())
-        client = TestClient(app, raise_server_exceptions=False)
-        resp = client.post("/", json=_jsonrpc_request("message/send", _send_message_params("unit-test")))
+        client = TestClient(app, raise_server_exceptions=False, headers={"A2A-Version": "1.0"})
+        resp = client.post("/", json=_jsonrpc_request("SendMessage", _send_message_params("unit-test")))
         assert resp.status_code == 200
         body = resp.json()
         assert "result" in body
-        task = body["result"]
-        assert task["status"]["state"] == "completed"
+        task = body["result"]["task"]
+        assert task["status"]["state"] == "TASK_STATE_COMPLETED"
         assert task["artifacts"][0]["parts"][0]["text"] == "echo: unit-test"
 
+    def test_v03_message_send_remains_compatible(self):
+        app = build_a2a_app(_EchoExecutor(), _make_agent_card())
+        client = TestClient(app, raise_server_exceptions=False)
+        params = {
+            "message": {
+                "message_id": str(uuid.uuid4()),
+                "role": "user",
+                "parts": [{"kind": "text", "text": "compat"}],
+            }
+        }
+
+        resp = client.post("/", json=_jsonrpc_request("message/send", params))
+
+        assert resp.status_code == 200
+        assert resp.json()["result"]["status"]["state"] == "completed"
+        assert resp.json()["result"]["artifacts"][0]["parts"][0]["text"] == "echo: compat"
+
     def test_custom_task_store_is_used_for_persistence(self):
         """Verify that a custom task_store actually stores the task."""
         store = InMemoryTaskStore()
         app = build_a2a_app(_EchoExecutor(), _make_agent_card(), task_store=store)
-        client = TestClient(app, raise_server_exceptions=False)
+        client = TestClient(app, raise_server_exceptions=False, headers={"A2A-Version": "1.0"})
 
         # Send a message so a task gets created
-        resp = client.post("/", json=_jsonrpc_request("message/send", _send_message_params("store-test")))
-        task_id = resp.json()["result"]["id"]
+        resp = client.post("/", json=_jsonrpc_request("SendMessage", _send_message_params("store-test")))
+        task_id = resp.json()["result"]["task"]["id"]
 
         # Retrieve from the same store via tasks/get
-        resp2 = client.post("/", json=_jsonrpc_request("tasks/get", {"id": task_id}))
+        resp2 = client.post("/", json=_jsonrpc_request("GetTask", {"id": task_id}))
         assert resp2.json()["result"]["id"] == task_id
 
     def test_agent_card_url_auto_populated_from_env(self):
-        """When AGENTCORE_RUNTIME_URL is set, agent_card.url is overridden."""
+        """When AGENTCORE_RUNTIME_URL is set, the JSON-RPC interface URL is overridden."""
         card = _make_agent_card()
-        assert card.url == "http://localhost:9000"
+        assert card.supported_interfaces[0].url == "http://localhost:9000"
         with patch.dict("os.environ", {"AGENTCORE_RUNTIME_URL": "https://deployed.example.com/"}):
             build_a2a_app(_EchoExecutor(), card)
-        assert card.url == "https://deployed.example.com/"
+        assert card.supported_interfaces[0].url == "https://deployed.example.com/"
 
     def test_agent_card_url_unchanged_without_env(self):
-        """When AGENTCORE_RUNTIME_URL is not set, agent_card.url stays as-is."""
+        """When AGENTCORE_RUNTIME_URL is not set, the JSON-RPC interface URL stays as-is."""
         card = _make_agent_card()
-        original_url = card.url
+        original_url = card.supported_interfaces[0].url
         with patch.dict("os.environ", {}, clear=False):
             import os
 
             os.environ.pop("AGENTCORE_RUNTIME_URL", None)
             build_a2a_app(_EchoExecutor(), card)
-        assert card.url == original_url
+        assert card.supported_interfaces[0].url == original_url
 
     def test_auto_builds_card_when_none_provided(self):
         """When agent_card is omitted, a default card is built automatically."""
@@ -167,7 +190,7 @@ def test_auto_builds_card_when_none_provided(self):
         assert resp.status_code == 200
         body = resp.json()
         assert body["name"] == "agent"
-        assert body["url"] == "http://localhost:9000/"
+        assert body["supportedInterfaces"][0]["url"] == "http://localhost:9000/"
 
     def test_auto_builds_card_from_strands_executor(self):
         """When executor has .agent with name/description, card is built from it."""
@@ -192,7 +215,7 @@ def test_auto_card_uses_runtime_url_env(self):
             app = build_a2a_app(_EchoExecutor())
         client = TestClient(app)
         resp = client.get("/.well-known/agent-card.json")
-        assert resp.json()["url"] == "https://prod.example.com/"
+        assert resp.json()["supportedInterfaces"][0]["url"] == "https://prod.example.com/"
 
 
 class TestBedrockCallContextBuilder:
@@ -334,11 +357,11 @@ def _test():
             executor = _EchoExecutor()
             builder = BedrockCallContextBuilder()
             app = build_a2a_app(executor, _make_agent_card(), context_builder=builder)
-            client = TestClient(app, raise_server_exceptions=False)
+            client = TestClient(app, raise_server_exceptions=False, headers={"A2A-Version": "1.0"})
 
             client.post(
                 "/",
-                json=_jsonrpc_request("message/send", _send_message_params("ctx-test")),
+                json=_jsonrpc_request("SendMessage", _send_message_params("ctx-test")),
                 headers={
                     "X-Amzn-Bedrock-AgentCore-Runtime-Session-Id": "unit-sess",
                     "X-Amzn-Bedrock-AgentCore-Runtime-Request-Id": "unit-req",
@@ -349,7 +372,7 @@ def _test():
             ctx = executor.last_call_context
             assert ctx is not None
             assert ctx.state["session_id"] == "unit-sess"
-            assert ctx.state["request_id"] == "unit-req"
+            assert ctx.state["bedrock_request_id"] == "unit-req"
             assert ctx.state["workload_access_token"] == "unit-token"
 
         self._run_in_isolated_context(_test)