feat: add A2A protocol support via serve_a2a

[tejaskash]

Summary

• Adds serve_a2a, build_a2a_app, and BedrockCallContextBuilder to bedrock_agentcore.runtime
• ~130 lines of Bedrock-specific glue delegating protocol handling to the official a2a-sdk
• Optional dependency: pip install "bedrock-agentcore[a2a]"
• Auto-builds AgentCard from StrandsA2AExecutor when not provided
• Auto-populates agent_card.url from AGENTCORE_RUNTIME_URL env var on deploy
• Docker/container host detection for 0.0.0.0 vs 127.0.0.1
• /ping health check endpoint with Bedrock-compatible response format
• Propagates Bedrock session, request, token, and custom headers via BedrockAgentCoreContext contextvars

Why

PR #217 reimplemented A2A from scratch (~1700 lines). The industry (ADK, Strands, CrewAI) has converged on the official a2a-sdk. This replaces that approach with minimal glue that delegates protocol handling to the SDK.

Strands

from strands import Agent
from strands.multiagent.a2a.executor import StrandsA2AExecutor

from bedrock_agentcore.runtime import serve_a2a

agent = Agent(
    name="Calculator Agent",
    description="A calculator agent that can perform basic arithmetic operations.",
    callback_handler=None,
)

if __name__ == "__main__":
    serve_a2a(StrandsA2AExecutor(agent))

Google ADK

from google.adk.a2a.executor.a2a_agent_executor import A2aAgentExecutor
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService

from a2a.types import AgentCapabilities, AgentCard, AgentSkill
from bedrock_agentcore.runtime import serve_a2a

agent = Agent(
    name="calculator_agent",
    model="gemini-2.0-flash",
    description="A calculator agent that can perform basic arithmetic operations.",
    instruction="You are a helpful calculator. Answer math questions clearly and concisely.",
)

runner = Runner(
    app_name=agent.name,
    agent=agent,
    session_service=InMemorySessionService(),
)

card = AgentCard(
    name=agent.name,
    description=agent.description,
    url="http://localhost:9000/",
    version="0.1.0",
    capabilities=AgentCapabilities(streaming=True),
    skills=[AgentSkill(id="calc", name="calculator", description="Arithmetic operations", tags=["math"])],
    default_input_modes=["text"],
    default_output_modes=["text"],
)

if __name__ == "__main__":
    serve_a2a(A2aAgentExecutor(runner=runner), card)

LangGraph

from langchain_aws import ChatBedrockConverse
from langgraph.prebuilt import create_react_agent

from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.tasks import TaskUpdater
from a2a.types import AgentCapabilities, AgentCard, AgentSkill, Part, TextPart
from a2a.utils import new_task
from bedrock_agentcore.runtime import serve_a2a

llm = ChatBedrockConverse(model="us.anthropic.claude-sonnet-4-20250514")

graph = create_react_agent(llm, tools=[], prompt="You are a helpful calculator.")


class LangGraphA2AExecutor(AgentExecutor):
    """Wraps a LangGraph CompiledGraph as an a2a-sdk AgentExecutor."""

    def __init__(self, graph):
        self.graph = graph

    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        task = context.current_task or new_task(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()
        result = await self.graph.ainvoke({"messages": [("user", user_text)]})
        response = result["messages"][-1].content

        await updater.add_artifact([Part(root=TextPart(text=response))])
        await updater.complete()

    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
        pass


card = AgentCard(
    name="langgraph-agent",
    description="A LangGraph agent on Bedrock AgentCore",
    url="http://localhost:9000/",
    version="0.1.0",
    capabilities=AgentCapabilities(streaming=True),
    skills=[AgentSkill(id="calc", name="calculator", description="Arithmetic operations", tags=["math"])],
    default_input_modes=["text"],
    default_output_modes=["text"],
)

if __name__ == "__main__":
    serve_a2a(LangGraphA2AExecutor(graph), card)

Test plan

• 21 unit tests in tests/bedrock_agentcore/runtime/test_a2a.py
• 10 integration tests in tests/integration/runtime/test_a2a_integration.py
• Full test suite passes (1112 passed, 1 skipped)
• Deployed and verified on Bedrock AgentCore with Strands, ADK, and LangGraph executors
• Agent card URL auto-populated correctly from AGENTCORE_RUNTIME_URL
• ruff check / ruff format clean

[sundargthb]

Your proposal listed three components: serve_a2a, BedrockCallContextBuilder, and build_runtime_url(agent_arn, region).

The third one isn’t here. Intentionally dropped, or coming in a follow-up? Customers deploying to AgentCore need to construct the runtime URL from their ARN — without this utility they’ll do it themselves (and can get the URL-encoding of the ARN wrong).

[sundargthb]

Minor: if future Starlette version compiles routes eagerly in build(), will this break ? I am good if we can rely on this implementation detail.