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 ag2 import Agent
from ag2.a2a import A2AServer, build_card
from ag2.config import AnthropicConfig
from ag2.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.

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 ag2 import Agent
from ag2.a2a import A2AServer
from ag2.config import AnthropicConfig
from ag2.tools import tool
from ag2.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:

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).

TLS#

build_grpc binds an insecure port by default for backwards compatibility. Pass a grpc.ServerCredentials object as server_credentials= to bind a secure port, and publish the interface with a grpcs:// or grpc+tls:// URL:

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

from pathlib import Path

import grpc

from ag2.a2a import build_card

private_key = Path("certs/server-key.pem").read_bytes()
certificate = Path("certs/server.pem").read_bytes()
credentials = grpc.ssl_server_credentials([(private_key, certificate)])

grpc_url = "grpcs://agents.example.com:50051"
card = build_card(agent, url="https://agents.example.com", transports=("grpc",), grpc_url=grpc_url)
grpc_server = server.build_grpc(
    bind="0.0.0.0:50051",
    grpc_url=grpc_url,
    card=card,
    server_credentials=credentials,
)
await grpc_server.start()
await grpc_server.wait_for_termination()

AG2 does not read certificate files or alter the credentials; certificate loading, rotation and mTLS policy remain under application control. To require client certificates, create the server credentials with root_certificates=<client CA> and require_client_auth=True. See gRPC TLS on the client for public-CA, private-CA and mTLS channels.

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.

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.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

from ag2.a2a import A2AServer, build_card
from ag2.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)],
)

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),
]

Example B — require Bearer AND API-key together (one require() with two args):

1
2
3

security=[
    require(bearer, api_key),
]

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)

Declaring Extensions#

A2A extensions let an agent advertise protocol behaviour beyond the base spec. AG2 always declares its own urn:ag2:client-tools:v1 as required=False; pass extensions= to declare your own alongside it.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

from a2a.types import AgentExtension

from ag2.a2a import build_card

card = build_card(
    agent,
    url="http://127.0.0.1:8000",
    extensions=[
        AgentExtension(
            uri="urn:acme:billing:v1",
            description="Attaches a cost breakdown to every reply",
            required=True,
        ),
    ],
)

required=True declares that the exchange genuinely breaks without the extension. AG2's client honours the flag: it refuses to connect to a card requiring a URI it hasn't activated, rather than proceeding and misbehaving mid-task. Note this is client-side only — build_card declares the requirement, it does not enforce it, and a client that ignores the flag is still served. Use required=False to let extension-unaware clients keep working.

Duplicate URIs raise ValueError, including redeclaring urn:ag2:client-tools:v1 — build_card always declares that one itself, so a card can never ship two entries for the same extension.

See Extensions for activating them on the client.

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=...).