Exposing an Agent as an A2A Server - AG2
Server
A2AServer wraps an existing Agent and produces a transport object you can serve directly. JSON-RPC is the default; the same A2AServer instance can also build REST and gRPC transports that share one task store.
Minimal Server
The smallest end-to-end setup: an Agent with a tool, served over JSON-RPC on a single port via uvicorn.
<br> 1<br> 2<br> 3<br> 4<br> 5<br> 6<br> 7<br> 8<br> 9<br>10<br>11<br>12<br>13<br>14<br>15<br>16<br>17<br>18<br>19<br>20<br>21<br>22<br> |
<br>import uvicorn<br>from ag2 import Agent<br>from ag2.a2a import A2AServer, build_card<br>from ag2.config import AnthropicConfig<br>from ag2.tools import tool<br>@tool(description="Add two integers and return the sum as a string.")<br>async def calc_add(a: int, b: int) -> str:<br> return f"{a + b}"<br>async def main() -> None:<br> agent = Agent(<br> name="claude",<br> config=AnthropicConfig(model="claude-sonnet-4-6"),<br> tools=[calc_add],<br> )<br> server = A2AServer(agent)<br> card = build_card(agent, url="http://127.0.0.1:8000")<br> asgi = server.build_jsonrpc(url="http://127.0.0.1:8000", card=card)<br> await uvicorn.Server(uvicorn.Config(asgi, host="127.0.0.1", port=8000)).serve()<br> |
After startup the agent card is reachable at http://127.0.0.1:8000/.well-known/agent-card.json. A client connects by passing that base URL to A2AConfig(card_url=...) — see the Client page.
What A2AServer Holds
A2AServer.__init__ materialises transport-agnostic state — the executor, the task store, optional push notifications. Transport-specific parameters (URL, paths, ports) live on the build_* methods.
| Constructor argument | Purpose |
|---|---|
agent |
The AG2 Agent exposed over A2A |
task_store |
Shared TaskStore. Defaults to a single InMemoryTaskStore reused across every build_* call |
push_config_store |
Enables push-notifications CRUD (see Tasks & Push). Optional |
push_sender |
Custom delivery sender. Defaults to no-op when not set |
extended_card |
Auth-aware extra metadata returned via GetExtendedAgentCard |
card_modifier / extended_card_modifier |
Per-request hooks that mutate the card before it's served |
executor |
Escape hatch — drop in a custom AgentExecutor (see Advanced) |
Note
The default InMemoryTaskStore is materialised once at __init__ time. This is what makes JSON-RPC, REST and gRPC bound to the same A2AServer see each other's tasks — a single store, three transports.
Server-side Tools
Tools attached to the wrapped Agent execute on the server, just as they would for a local agent. The remote LLM picks them up automatically.
<br> 1<br> 2<br> 3<br> 4<br> 5<br> 6<br> 7<br> 8<br> 9<br>10<br>11<br>12<br>13<br>14<br>15<br> |
<br>from ag2 import Agent<br>from ag2.a2a import A2AServer<br>from ag2.config import AnthropicConfig<br>from ag2.tools import tool<br>from ag2.tools.builtin import WebSearchTool<br>agent = Agent(<br> name="claude",<br> config=AnthropicConfig(model="claude-sonnet-4-6"),<br> tools=[<br> WebSearchTool(), # built-in provider tool — runs on Anthropic's side<br> calc_add, # @tool — runs on this server<br> ],<br>)<br>server = A2AServer(agent)<br> |
For client-side tools — declared on the caller and forwarded back from the server when the LLM calls them — see Client → Local Tools.
Choosing a Transport
The default build_jsonrpc(...) is the right choice for most setups. JSON-RPC is the most widely-supported A2A binding, works over any HTTP infrastructure (proxies, gateways, load balancers), and is what every A2A client implementation speaks first.
Reach for an alternative transport when:
| Transport | Use when |
|---|---|
build_jsonrpc (default) |
You want the simplest, most portable HTTP binding. Recommended start. |
build_rest |
You need a HTTP+JSON REST surface with stable URLs (logging, cache control, route-level auth in a gateway). |
build_grpc |
You need bidirectional streaming with low overhead, or your infra is gRPC-native. |
REST
<br>1<br>2<br>3<br> |
<br>card = build_card(agent, url="http://127.0.0.1:8001", transports=("rest",))<br>rest_app = server.build_rest(url="http://127.0.0.1:8001", card=card)<br>await uvicorn.Server(uvicorn.Config(rest_app, host="127.0.0.1", port=8001)).serve()<br> |
build_rest(path_prefix="/v1") mounts the routes under a sub-path; both the card and the dispatcher respect it.
gRPC
<br> 1<br> 2<br> 3<br> 4<br> 5<br> 6<br> 7<br> 8<br> 9<br>10<br>11<br>12<br>13<br> |
<br>card = build_card(<br> agent,<br> url="grpc://127.0.0.1:50051",<br> transports=("grpc",),<br> grpc_url="grpc://127.0.0.1:50051",<br>)<br>grpc_server = server.build_grpc(<br> bind="127.0.0.1:50051",<br> grpc_url="grpc://127.0.0.1:50051",<br> card=card,<br>)<br>await grpc_server.start()<br>await grpc_server.wait_for_termination()<br> |
build_grpc returns an unstarted grpc.aio.Server — the caller is responsible for start() and wait_for_termination(). bind is the listener address, grpc_url is the URL declared in the card (they're usually identical, but differ when the server sits behind a load balancer).
Note
A2A v1.x has no GetAgentCard gRPC method — the public card is always served over HTTP at /.well-known/agent-card.json. So even for a gRPC-only server clients fetch the card via HTTP first, then switch to gRPC for the actual exchange. Plan your card URL accordingly.
One Server, Three Transports
The same A2AServer instance can back any combination of transports. Build a single multi-transport AgentCard and call each build_* against it — they share one task store.
<br> 1<br> 2<br> 3<br> 4<br> 5<br> 6<br> 7<br> 8<br> 9<br>10<br>11<br>12<br>13<br>14<br>15<br>16<br>17<br>18<br>19<br>20<br>21<br> |
<br>from a2a.server.tasks import InMemoryPushNotificationConfigStore<br>server = A2AServer(agent, push_config_store=InMemoryPushNotificationConfigStore())<br>card = build_card(<br> agent,<br> url="http://127.0.0.1:8000",<br> transports=("jsonrpc", "rest", "grpc"),<br> rest_url="http://127.0.0.1:8001",<br> grpc_url="grpc://127.0.0.1:50051",<br>)<br>asgi = server.build_jsonrpc(url="http://127.0.0.1:8000", card=card)<br>rest = server.build_rest(url="http://127.0.0.1:8001", card=card)<br>grpc = server.build_grpc(bind="127.0.0.1:50051", grpc_url="grpc://127.0.0.1:50051", card=card)<br>await grpc.start()<br>await asyncio.gather(<br> uvicorn.Server(uvicorn.Config(asgi, host="127.0.0.1", port=8000)).serve(),<br> uvicorn.Server(uvicorn.Config(rest, host="127.0.0.1", port=8001)).serve(),<br> grpc.wait_for_termination(),<br>)<br> |
Customising the AgentCard
build_card(agent, url=...) accepts a handful of optional kwargs to enrich the published card with discovery metadata and auth declarations.
| Argument | Purpose |
|---|---|
version |
Card version (defaults to "1.0.0") |
description |
Free-form description. Defaults to the first entry of the agent's system prompt |
skills |
Explicit Sequence[AgentSkill]. When None, build_card walks agent.tools for any SkillsToolkit and publishes its local skills automatically; falls back to a single agent-derived skill if none are found |
push_notifications |
Toggles capabilities.push_notifications on the card |
provider |
AgentProvider block (organization, URL) |
documentation_url / icon_url |
Discovery metadata |
security |
Auth declarations — see below |
tenants |
Mapping[TransportName, str] — surface a per-transport tenant on the corresponding AgentInterface.tenant |
rest_url / rest_path_prefix / grpc_url |
Per-transport URL overrides for multi-transport cards |
Declaring Authentication
ag2.a2a.security ships factories for every A2A-recognised scheme. Each factory returns a typed Scheme object that carries its card-level binding name. Pass them to require(...) to build Requirement entries; build_card auto-derives the card's security_schemes from the schemes referenced in security= — no duplicate declarations.
<br> 1<br> 2<br> 3<br> 4<br> 5<br> 6<br> 7<br> 8<br> 9<br>10<br>11<br>12<br>13<br>14<br>15<br> |
<br>from ag2.a2a import A2AServer, build_card<br>from ag2.a2a.security import (<br> bearer_scheme,<br> api_key_scheme,<br> require,<br>)<br>bearer = bearer_scheme(name="bearer", bearer_format="JWT")<br>api_key = api_key_scheme(name="x_api_key", key_name="X-API-Key", location="header")<br>card = build_card(<br> agent,<br> url="http://127.0.0.1:8000",<br> security=[require(bearer), require(api_key)],<br>)<br> |
| Helper | Scheme |
|---|---|
bearer_scheme(name=..., bearer_format=..., description=...) |
HTTP Bearer (e.g. JWT) |
http_auth_scheme(name=..., scheme=..., ...) |
Any other HTTP auth scheme (basic, digest, custom bearer formats) |
api_key_scheme(name=..., key_name=..., location=...) |
API key in header / query / cookie. name is the card binding; key_name is the header/query/cookie key sent by the client. |
oauth2_scheme(name=..., flows=..., oauth2_metadata_url=...) |
OAuth2 wrapping a pre-built OAuthFlows |
open_id_connect_scheme(name=..., url=...) |
OpenID Connect discovery URL |
mtls_scheme(name=...) |
Mutual TLS client-cert auth |
Combining requirements: AND vs OR
The security= list holds independent rules — clients only need to satisfy one of them (entries are OR-ed). Inside a single require(...) call, all passed schemes must be presented together (arguments are AND-ed). Attach OAuth2/OIDC scopes via scheme.with_scopes(...).
Example A — accept Bearer OR API-key (two separate require() calls):
<br>1<br>2<br>3<br>4<br> |
<br>security=[<br> require(bearer),<br> require(api_key),<br>]<br> |
| Request headers | Accepted? |
|---|---|
Authorization: Bearer <jwt> |
✅ matches first rule |
X-API-Key: <key> |
✅ matches second rule |
| both headers present | ✅ either rule alone is enough |
| neither | ❌ no rule satisfied |
Example B — require Bearer AND API-key together (one require() with two args):
<br>1<br>2<br>3<br> |
<br>security=[<br> require(bearer, api_key),<br>]<br> |
| Request headers | Accepted? |
|---|---|
only Authorization: Bearer <jwt> |
❌ missing API key |
only X-API-Key: <key> |
❌ missing Bearer |
| both headers present | ✅ both args inside the same require() satisfied |
Example C — mixing scopes (OAuth2 needs scopes, Bearer doesn't):
<br>1<br>2<br>3<br> |
<br>security=[<br> require(bearer, oauth.with_scopes("read", "write")),<br>]<br> |
Scheme binding names are arbitrary strings — pass any value to name=, including non-identifier forms like "X-My-Scheme":
<br>1<br>2<br> |
<br>custom = bearer_scheme(name="X-My-Scheme")<br>require(custom)<br> |
Note
build_card only declares auth on the card — it does not enforce it. Wire the actual check into the ASGI app (Starlette middleware, gateway, reverse proxy) or the gRPC server's interceptors.
Adding Cross-cutting Middleware
A2A doesn't define server-side middleware. Attach CORS, auth or tracing directly to the returned transport object:
<br>1<br>2<br>3<br>4<br> |
<br>from starlette.middleware.cors import CORSMiddleware<br>asgi = server.build_jsonrpc(url="http://127.0.0.1:8000")<br>asgi.add_middleware(CORSMiddleware, allow_origins=["*"])<br> |