Skip to content

a2a_protocol_examples.md

Latest commit

 

History

History
226 lines (163 loc) · 8.33 KB

a2a_protocol_examples.md

File metadata and controls

226 lines (163 loc) · 8.33 KB
Raw

A2A Protocol Support

This document explains how to serve your agent using the A2A (Agent-to-Agent) protocol on Bedrock AgentCore Runtime.

Installation

Use the a2a extra for a2a-sdk 0.3, including the current Strands A2A integration:

pip install "bedrock-agentcore[a2a]"

Use the mutually exclusive a2a-v1 extra to opt in to A2A protocol v1:

pip install "bedrock-agentcore[a2a-v1]"

The runtime adapter detects the installed SDK major version. Do not install both extras in the same environment.

Quick Start

Strands Agent

Strands provides a built-in StrandsA2AExecutor that wraps a Strands Agent as an A2A executor. When no AgentCard is provided, one is auto-built from the agent's name and description.

from strands import Agent
from strands.a2a import StrandsA2AExecutor
from bedrock_agentcore.runtime import serve_a2a

agent = Agent(
    model="us.anthropic.claude-sonnet-4-20250514",
    system_prompt="You are a helpful calculator.",
)

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

LangGraph Agent

LangGraph requires a thin AgentExecutor wrapper (~15 lines) and an explicit AgentCard:

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):
    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", tags=["math"])],
    default_input_modes=["text"],
    default_output_modes=["text"],
)

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

Google ADK Agent

Google ADK provides A2aAgentExecutor built-in. You supply an explicit AgentCard:

from google.adk.agents import LlmAgent
from google.adk.runners import Runner
from google.adk.a2a import A2aAgentExecutor

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

agent = LlmAgent(
    model="gemini-2.0-flash",
    name="calculator",
    description="A calculator agent",
    instruction="You are a helpful calculator.",
)
runner = Runner(agent=agent, app_name="calculator", session_service=None)

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

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

API Reference

serve_a2a(executor, agent_card=None, *, port=None, host=None, ...)

Starts a Bedrock-compatible A2A server with uvicorn.

Parameter Type Default Description
executor AgentExecutor required An a2a-sdk AgentExecutor that implements the agent logic
agent_card AgentCard None Agent metadata. Auto-built from executor if omitted (works best with Strands)
port int | None None Port to serve on. Uses A2A_PORT, then 9000, when omitted. The generic PORT is ignored — see Ports
host str None Host to bind to. Auto-detected: 0.0.0.0 in Docker, 127.0.0.1 otherwise
task_store TaskStore None Custom task store; defaults to InMemoryTaskStore
context_builder CallContextBuilder None Custom context builder; defaults to BedrockCallContextBuilder
ping_handler Callable[[], PingStatus] None Custom health check callback
**kwargs Additional arguments forwarded to uvicorn.run()

build_a2a_app(executor, agent_card=None, *, task_store=None, context_builder=None, ping_handler=None)

Builds a Starlette ASGI application without starting a server. Useful for testing or embedding in a larger app.

Returns a Starlette application with routes:

  • POST / — A2A JSON-RPC endpoint (message/send, message/stream, tasks/get, tasks/cancel)
  • GET /.well-known/agent-card.json — Agent card discovery
  • GET /ping — Bedrock health check

build_runtime_url(agent_arn, region=None)

Builds the Bedrock AgentCore runtime invocation URL from an agent ARN.

from bedrock_agentcore.runtime import build_runtime_url

url = build_runtime_url("arn:aws:bedrock-agentcore:us-east-1:123456789012:runtime/my-agent-abc123")
# https://bedrock-agentcore.us-east-1.amazonaws.com/runtimes/arn%3Aaws%3A.../invocations

BedrockCallContextBuilder

Extracts Bedrock runtime headers from incoming requests and propagates them into BedrockAgentCoreContext contextvars. This is the default context_builder used by build_a2a_app and serve_a2a.

Headers extracted:

  • X-Amzn-Bedrock-AgentCore-Runtime-Session-Id — session ID
  • X-Amzn-Bedrock-AgentCore-Runtime-Request-Id — request ID (auto-generated UUID if missing)
  • WorkloadAccessToken — workload access token
  • OAuth2CallbackUrl — OAuth2 callback URL
  • Authorization — authorization header
  • X-Amzn-Bedrock-AgentCore-Runtime-Custom-* — custom headers

Behavior Details

Ports

The AgentCore Runtime service contract fixes the A2A container port at 9000 (HTTP uses 8080, MCP uses 8000). The runtime proxies invocations to 9000 only, so an A2A server bound elsewhere is unreachable and every invocation fails with HTTP 424 (RuntimeClientError).

serve_a2a therefore defaults to 9000 and ignores the generic PORT environment variable. PORT is a widespread convention (Heroku, Cloud Run, App Runner) and is commonly already set to another protocol's port — notably in images shared across an HTTP and an A2A runtime, where PORT=8080 is correct for HTTP and fatal for A2A.

To override the port for local development, use the protocol-scoped A2A_PORT, or pass port= explicitly:

A2A_PORT=9001 python main.py
serve_a2a(executor, port=9001)

Precedence is port= argument, then A2A_PORT, then 9000. When the resolved port is not 9000, serve_a2a logs a warning, since that configuration cannot work in a deployed runtime.

Agent Card Auto-Population

When deployed on Bedrock AgentCore, the AGENTCORE_RUNTIME_URL environment variable is set automatically. The agent card's url field is updated to match, so you don't need to hardcode the deployed URL.

Docker Host Detection

When host is not specified, serve_a2a automatically binds to 0.0.0.0 inside Docker containers (detected via /.dockerenv or DOCKER_CONTAINER env var) and 127.0.0.1 otherwise.

Custom Ping Handler

from bedrock_agentcore.runtime import serve_a2a
from bedrock_agentcore.runtime.models import PingStatus

def my_ping():
    if is_overloaded():
        return PingStatus.HEALTHY_BUSY
    return PingStatus.HEALTHY

serve_a2a(executor, ping_handler=my_ping)

If the ping handler raises an exception, the server falls back to PingStatus.HEALTHY.