Skip to content

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.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
import uvicorn

from autogen.beta import Agent
from autogen.beta.a2a import A2AServer, build_card
from autogen.beta.config import AnthropicConfig
from autogen.beta.tools import tool

@tool(description="Add two integers and return the sum as a string.")
async def calc_add(a: int, b: int) -> str:
    return f"{a + b}"

async def main() -> None:
    agent = Agent(
        name="claude",
        config=AnthropicConfig(model="claude-sonnet-4-6"),
        tools=[calc_add],
    )
    server = A2AServer(agent)
    card = build_card(agent, url="http://127.0.0.1:8000")
    asgi = server.build_jsonrpc(url="http://127.0.0.1:8000", card=card)

    await uvicorn.Server(uvicorn.Config(asgi, host="127.0.0.1", port=8000)).serve()

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.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
from autogen.beta import Agent
from autogen.beta.a2a import A2AServer
from autogen.beta.config import AnthropicConfig
from autogen.beta.tools import tool
from autogen.beta.tools.builtin import WebSearchTool

agent = Agent(
    name="claude",
    config=AnthropicConfig(model="claude-sonnet-4-6"),
    tools=[
        WebSearchTool(),  # built-in provider tool — runs on Anthropic's side
        calc_add,         # @tool — runs on this server
    ],
)
server = A2AServer(agent)

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#

1
2
3
card = build_card(agent, url="http://127.0.0.1:8001", transports=("rest",))
rest_app = server.build_rest(url="http://127.0.0.1:8001", card=card)
await uvicorn.Server(uvicorn.Config(rest_app, host="127.0.0.1", port=8001)).serve()

build_rest(path_prefix="/v1") mounts the routes under a sub-path; both the card and the dispatcher respect it.

gRPC#

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
card = build_card(
    agent,
    url="grpc://127.0.0.1:50051",
    transports=("grpc",),
    grpc_url="grpc://127.0.0.1:50051",
)
grpc_server = server.build_grpc(
    bind="127.0.0.1:50051",
    grpc_url="grpc://127.0.0.1:50051",
    card=card,
)
await grpc_server.start()
await grpc_server.wait_for_termination()

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.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
from a2a.server.tasks import InMemoryPushNotificationConfigStore

server = A2AServer(agent, push_config_store=InMemoryPushNotificationConfigStore())
card = build_card(
    agent,
    url="http://127.0.0.1:8000",
    transports=("jsonrpc", "rest", "grpc"),
    rest_url="http://127.0.0.1:8001",
    grpc_url="grpc://127.0.0.1:50051",
)

asgi = server.build_jsonrpc(url="http://127.0.0.1:8000", card=card)
rest = server.build_rest(url="http://127.0.0.1:8001", card=card)
grpc = server.build_grpc(bind="127.0.0.1:50051", grpc_url="grpc://127.0.0.1:50051", card=card)

await grpc.start()
await asyncio.gather(
    uvicorn.Server(uvicorn.Config(asgi, host="127.0.0.1", port=8000)).serve(),
    uvicorn.Server(uvicorn.Config(rest, host="127.0.0.1", port=8001)).serve(),
    grpc.wait_for_termination(),
)

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#

autogen.beta.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.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
from autogen.beta.a2a import A2AServer, build_card
from autogen.beta.a2a.security import (
    bearer_scheme,
    api_key_scheme,
    require,
)

bearer = bearer_scheme(name="bearer", bearer_format="JWT")
api_key = api_key_scheme(name="x_api_key", key_name="X-API-Key", location="header")

card = build_card(
    agent,
    url="http://127.0.0.1:8000",
    security=[require(bearer), require(api_key)],
)
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):

1
2
3
4
security=[
    require(bearer),
    require(api_key),
]
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):

1
2
3
security=[
    require(bearer, api_key),
]
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):

1
2
3
security=[
    require(bearer, oauth.with_scopes("read", "write")),
]

Scheme binding names are arbitrary strings — pass any value to name=, including non-identifier forms like "X-My-Scheme":

1
2
custom = bearer_scheme(name="X-My-Scheme")
require(custom)

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:

1
2
3
4
from starlette.middleware.cors import CORSMiddleware

asgi = server.build_jsonrpc(url="http://127.0.0.1:8000")
asgi.add_middleware(CORSMiddleware, allow_origins=["*"])

For gRPC, attach interceptors when constructing the channel via grpc.aio.Server options on build_grpc(options=...).