A2AServer(agent, *, extended_card=None, card_modifier=None, extended_card_modifier=None, task_store=None, push_config_store=None, push_sender=None, executor=None)
Wrap an AG2 Agent as an A2A endpoint.
Holds transport-agnostic state (executor, task/push stores, extended card, per-card modifier hooks). Transport-specific parameters (URL, ports, paths) live on the build_* methods — one server can be exposed on different URLs through different transports.
extended_card, when supplied, is served via the JSON-RPC GetExtendedAgentCard method; the public card automatically flips capabilities.extended_agent_card when an extended card is provided.
Each builder returns a ready-to-serve transport object: :py:meth:build_jsonrpc and :py:meth:build_rest return a Starlette ASGI app; :py:meth:build_grpc returns a grpc.aio.Server.
A2A spec doesn't define middleware — attach cross-cutting concerns (CORS, auth, tracing) to the returned transport object directly.
Source code in autogen/beta/a2a/server.py
| def __init__(
self,
agent: Agent,
*,
extended_card: AgentCard | None = None,
card_modifier: CardModifier | None = None,
extended_card_modifier: ExtendedCardModifier | None = None,
task_store: TaskStore | None = None,
push_config_store: PushNotificationConfigStore | None = None,
push_sender: PushNotificationSender | None = None,
executor: A2AAgentExecutorBase | None = None,
) -> None:
self._agent = agent
self._extended_card = extended_card
self._card_modifier = card_modifier
self._extended_card_modifier = extended_card_modifier
# Materialise the store eagerly so multi-transport setups (same
# server exposed via JSON-RPC + REST + gRPC) all share one task
# store. Otherwise each builder defaults to its own.
self._task_store = task_store or InMemoryTaskStore()
self._push_config_store = push_config_store
self._push_sender = push_sender
# ``executor`` is escape-hatch for tests / advanced use cases that
# need a custom ``AgentExecutor``. Default wraps the supplied agent.
self._executor = executor if executor is not None else AgentExecutor(agent)
|
property
The shared task store used across all transport builders.
build_jsonrpc(*, url, card=None, rpc_url='/', card_url=DEFAULT_AGENT_CARD_PATH, legacy_card_url=LEGACY_AGENT_CARD_PATH)
Starlette ASGI app exposing JSON-RPC routes + agent card.
Source code in autogen/beta/a2a/server.py
| def build_jsonrpc(
self,
*,
url: str,
card: AgentCard | None = None,
rpc_url: str = "/",
card_url: str = DEFAULT_AGENT_CARD_PATH,
legacy_card_url: str | None = LEGACY_AGENT_CARD_PATH,
) -> "Starlette":
"""Starlette ASGI app exposing JSON-RPC routes + agent card."""
resolved_card = card or build_card(
self._agent,
url=url,
transports=("jsonrpc",),
push_notifications=self._push_config_store is not None,
)
return build_jsonrpc_asgi(
agent_card=resolved_card,
rpc_url=rpc_url,
card_url=card_url,
legacy_card_url=legacy_card_url,
**self._shared_kwargs(include_card_modifier=True),
)
|
build_rest(*, url, card=None, path_prefix='', card_url=DEFAULT_AGENT_CARD_PATH, legacy_card_url=LEGACY_AGENT_CARD_PATH)
Starlette ASGI app exposing REST routes + agent card.
path_prefix mounts REST under a sub-path (e.g. "/v1"); both the AgentCard interface URL and the dispatcher respect it.
Source code in autogen/beta/a2a/server.py
| def build_rest(
self,
*,
url: str,
card: AgentCard | None = None,
path_prefix: str = "",
card_url: str = DEFAULT_AGENT_CARD_PATH,
legacy_card_url: str | None = LEGACY_AGENT_CARD_PATH,
) -> "Starlette":
"""Starlette ASGI app exposing REST routes + agent card.
``path_prefix`` mounts REST under a sub-path (e.g. ``"/v1"``); both
the AgentCard interface URL and the dispatcher respect it.
"""
resolved_card = card or build_card(
self._agent,
url=url,
transports=("rest",),
rest_path_prefix=path_prefix,
push_notifications=self._push_config_store is not None,
)
return build_rest_asgi(
agent_card=resolved_card,
path_prefix=path_prefix,
card_url=card_url,
legacy_card_url=legacy_card_url,
**self._shared_kwargs(include_card_modifier=True),
)
|
build_grpc(*, bind, grpc_url, card=None, options=())
grpc.aio.Server bound to bind; caller starts/awaits it.
bind is the listener address (e.g. "0.0.0.0:50051"). grpc_url is the public URL clients will connect to (used in the AgentCard interface entry — usually identical to bind, but not when behind a load balancer). Insecure binding only.
card_modifier does not apply: A2A v1.x has no GetAgentCard gRPC method — the public card is served over HTTP only. extended_card_modifier does apply (gRPC has GetExtendedAgentCard).
Source code in autogen/beta/a2a/server.py
| def build_grpc(
self,
*,
bind: str,
grpc_url: str,
card: AgentCard | None = None,
options: Sequence[tuple[str, Any]] = (),
) -> "Server":
"""``grpc.aio.Server`` bound to ``bind``; caller starts/awaits it.
``bind`` is the listener address (e.g. ``"0.0.0.0:50051"``).
``grpc_url`` is the public URL clients will connect to (used in
the AgentCard interface entry — usually identical to ``bind``,
but not when behind a load balancer). Insecure binding only.
``card_modifier`` does not apply: A2A v1.x has no ``GetAgentCard``
gRPC method — the public card is served over HTTP only.
``extended_card_modifier`` does apply (gRPC has ``GetExtendedAgentCard``).
"""
resolved_card = card or build_card(
self._agent,
url=grpc_url,
transports=("grpc",),
grpc_url=grpc_url,
push_notifications=self._push_config_store is not None,
)
return build_grpc_server(
agent_card=resolved_card,
bind=bind,
options=options,
**self._shared_kwargs(include_card_modifier=False),
)
|