Client

A2AConfig is a ModelConfig — pass it to a regular Agent and the remote A2A server becomes that agent's LLM provider. Conversation history, tool calls and streaming are negotiated through the protocol; calling code keeps the familiar agent.ask(...) / reply.ask(...) shape.

Minimal Client#

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

from ag2 import Agent
from ag2.a2a import A2AConfig

async def main() -> None:
    remote = Agent(
        "remote",
        config=A2AConfig(card_url="http://127.0.0.1:8000"),
    )
    reply = await remote.ask("Add 17 and 25 with calc_add. Just the number.")
    print(reply.response.content)  # -> "42"

card_url is the HTTP(S) base where the server publishes /.well-known/agent-card.json. The client fetches the card on first use, picks a binding from supported_interfaces, and uses the URL declared in the card for every subsequent request — you don't pass transport-specific URLs.

Selecting a Transport#

When the card declares multiple bindings, prefer=... forces a choice:

1
2
3
4

config = A2AConfig(
    card_url="http://127.0.0.1:8000",
    prefer="grpc",  # one of "jsonrpc" | "rest" | "grpc"
)

prefer=None (default) auto-picks: if exactly one declared interface URL matches card_url it wins; otherwise the first server-listed interface is used.

gRPC TLS#

The default gRPC channel factory selects security from the interface URL published in the agent card. grpcs:// and grpc+tls:// create a TLS channel using the system CA roots; grpc://, grpc+insecure:// and a bare host:port keep the backwards-compatible insecure channel.

For a server certificate issued by a public CA, the client needs no TLS-specific configuration. It fetches the card over HTTPS and follows the secure gRPC interface declared there:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11

from ag2 import Agent
from ag2.a2a import A2AConfig

remote = Agent(
    "remote",
    config=A2AConfig(
        card_url="https://agents.example.com",
        prefer="grpc",
    ),
)
reply = await remote.ask("Summarize the latest report.")

Private CAs, self-signed certificates and mTLS need custom grpc.ChannelCredentials. Wrap those credentials with secure_grpc_channel_factory(...) and pass the returned callable through the existing grpc_channel_factory escape hatch:

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

from pathlib import Path

import grpc

from ag2 import Agent
from ag2.a2a import A2AConfig, secure_grpc_channel_factory

ca_certificate = Path("certs/ca.pem").read_bytes()
credentials = grpc.ssl_channel_credentials(root_certificates=ca_certificate)
remote = Agent(
    "remote",
    config=A2AConfig(
        card_url="https://agents.internal.example",
        prefer="grpc",
        grpc_channel_factory=secure_grpc_channel_factory(credentials),
    ),
)
reply = await remote.ask("Run the internal health check.")

The secure factory always creates a TLS channel, regardless of whether the card uses a secure, insecure or bare gRPC URL spelling. It also accepts gRPC channel options via options=. For mTLS, construct the credentials with root_certificates=, private_key= and certificate_chain=; AG2 passes the resulting object to gRPC unchanged.

Multi-turn — reply.ask#

A2A servers are stateless from AG2's perspective: every call ships the full conversation history as a application/vnd.ag2.history+json DataPart attached to the outgoing message. The continuation API is the regular reply.ask(...):

1
2
3
4
5

remote = Agent("remote", config=A2AConfig(card_url="http://127.0.0.1:8000"))

reply = await remote.ask("Remember the number 42.")
reply = await reply.ask("What number did I ask you to remember?")
print(reply.response.content)

The remote agent recovers the entire prior context on every turn — there is no server-side session id to manage.

Local Tools (forwarded from the server)#

Tools declared on the client are advertised to the server in the AG2 client-tools extension. When the remote LLM picks one, the call is routed back to the client and executed locally; the tool result is sent back into the server-side LLM loop. The server LLM never sees your local environment.

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

from datetime import datetime

from ag2 import Agent
from ag2.a2a import A2AConfig
from ag2.tools import tool

@tool(description="Return the user's local wall-clock time as ISO-8601.")
def get_local_time() -> str:
    return datetime.now().isoformat(timespec="seconds")

remote = Agent(
    "remote",
    config=A2AConfig(card_url="http://127.0.0.1:8000"),
    tools=[get_local_time],
)
reply = await remote.ask("What time is it on my machine? Use get_local_time.")

The same agent can mix client-side and server-side tools — server tools execute remotely, client tools execute locally, and the LLM picks freely between them within one turn.

Remote Agent as a Sub-tool — as_tool()#

A remote A2A agent plugs into a local Agent like any other delegate via Agent.as_tool(). The local LLM decides when to delegate; the wrapper exposes a task_<name> tool that takes an objective (and optional context).

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

from ag2 import Agent
from ag2.a2a import A2AConfig
from ag2.config import AnthropicConfig

researcher = Agent(
    "researcher",
    config=A2AConfig(card_url="http://research.internal:8000"),
)

writer = Agent(
    "writer",
    prompt="Use the researcher tool to gather facts before writing a draft.",
    config=AnthropicConfig(model="claude-sonnet-4-6"),
    tools=[researcher.as_tool(description="Delegate research questions to the remote researcher.")],
)

reply = await writer.ask("Write a 3-paragraph brief on the latest A2A spec changes.")

This composes naturally with several remotes — give each as_tool() a distinct name= and let the local LLM route by capability. See Subagents for the general as_tool() semantics.

Extensions#

A2A extensions are activated per connection by URI. List them on A2AConfig.extensions; every URI must be advertised in the server card's capabilities.extensions.

1
2
3
4
5
6
7
8

from ag2 import Agent
from ag2.a2a import A2AConfig

remote = Agent(
    "remote",
    config=A2AConfig(card_url="http://127.0.0.1:8000", extensions=["urn:acme:billing:v1"]),
)
reply = await remote.ask("invoice summary for Q3")

Activated URIs travel on both channels the spec allows — the Message.extensions field and the A2A-Extensions header (gRPC metadata on the gRPC binding) — and stay attached across continuation legs, so a tool-result or HITL round-trip does not silently deactivate them mid-task.

The card and the activation list are reconciled before the first request goes out, so a mismatch surfaces as A2AExtensionNotSupportedError at connect time instead of an opaque failure mid-task. It is raised in both directions:

• a URI the card does not advertise — a client-side mistake;
• an extension the card marks required=True that the client did not activate — the server expects behaviour AG2 will not provide.

urn:ag2:client-tools:v1 is the exception to the second rule: AG2 implements it natively, so a card may require it without the client listing it explicitly.

Activation covers every request the config makes, not just ask. The one-shot task and push-notification helpers open their own connection from the same A2AConfig, so they activate the same URIs and run the same reconciliation — a server enforcing a required=True extension sees a consistent client on the conversational and the administrative path alike.

1
2
3
4
5
6

from ag2.a2a.errors import A2AExtensionNotSupportedError

try:
    await remote.ask("invoice summary for Q3")
except A2AExtensionNotSupportedError as err:
    print(err.uris)  # the URIs that could not be reconciled

See Declaring Extensions for the server side.

A2AConfig Reference#

Constructing From a Pre-fetched Card#

When the card has already been resolved (discovery service, on-disk cache), A2AConfig.from_card(...) skips the network round-trip on connect:

1
2
3

from ag2.a2a import A2AConfig

config = A2AConfig.from_card(card, prefer="jsonrpc", timeout=30.0)

card_url defaults to the first interface URL on the card; pass card_url=... to override.