Skip to content

A2AConfig

autogen.beta.a2a.config.A2AConfig dataclass #

A2AConfig(card_url, prefer=None, streaming=True, headers=None, timeout=60.0, max_reconnects=3, reconnect_backoff=0.5, polling_interval=0.5, input_required_timeout=None, httpx_client_factory=None, interceptors=(), grpc_channel_factory=None, preset_card=None, tenant=None, history_length=None)

Bases: ModelConfig

Connection config for an A2A agent acting as an LLM provider.

card_url is the HTTP(S) URL where the agent card is published (fetched from {card_url}/.well-known/agent-card.json per spec). The actual transport endpoint for the request/response exchange is read from card.supported_interfaces — the user does not pass it.

prefer selects a transport when the server card declares more than one binding. None (default) auto-picks: if exactly one interface matches card_url it is used; otherwise the first server-listed interface wins. Pass "jsonrpc" | "rest" | "grpc" to force a specific binding.

polling_interval is used when the server card declares capabilities.streaming=False or when the user opts into streaming=False: Task state is polled via get_task every polling_interval seconds until terminal.

input_required_timeout caps how long the client waits on the HITL hook when the server transitions a task into TASK_STATE_INPUT_REQUIRED. None means wait indefinitely (matches ConversationContext.input).

grpc_channel_factory builds a grpc.aio.Channel for a given URL when the resolved transport is gRPC. Optional — defaults to insecure_channel via default_grpc_channel_factory.

tenant scopes every outgoing request to a specific tenant on the remote server (A2A multi-tenancy: a single shared backend can isolate data per tenant). Per-call override is available via context.variables["a2a:tenant"].

history_length truncates the server-side Task.history echoed back on get_task / list operations to the most recent N messages. Pure server-side hint — does not change what the client uploads.

card_url instance-attribute #

card_url

prefer class-attribute instance-attribute #

prefer = None

streaming class-attribute instance-attribute #

streaming = True

headers class-attribute instance-attribute #

headers = None

timeout class-attribute instance-attribute #

timeout = 60.0

max_reconnects class-attribute instance-attribute #

max_reconnects = 3

reconnect_backoff class-attribute instance-attribute #

reconnect_backoff = 0.5

polling_interval class-attribute instance-attribute #

polling_interval = 0.5

input_required_timeout class-attribute instance-attribute #

input_required_timeout = None

httpx_client_factory class-attribute instance-attribute #

httpx_client_factory = field(default=None, repr=False)

interceptors class-attribute instance-attribute #

interceptors = ()

grpc_channel_factory class-attribute instance-attribute #

grpc_channel_factory = field(default=None, repr=False)

preset_card class-attribute instance-attribute #

preset_card = field(default=None, repr=False)

tenant class-attribute instance-attribute #

tenant = None

history_length class-attribute instance-attribute #

history_length = None

copy #

copy(**overrides)
Source code in autogen/beta/a2a/config.py 
97
98
def copy(self, /, **overrides: Unpack[A2AConfigOverrides]) -> Self:
    return replace(self, **overrides)

from_card classmethod #

from_card(card, *, card_url=None, **overrides)

Construct a config from a pre-fetched AgentCard.

Useful when the card has already been resolved (e.g. via a discovery service) and a network round-trip on connect can be skipped. card_url defaults to the first interface declared on the card; raises A2AInvalidCardError if neither is available.

Source code in autogen/beta/a2a/config.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
@classmethod
def from_card(
    cls,
    card: AgentCard,
    *,
    card_url: str | None = None,
    **overrides: Any,
) -> Self:
    """Construct a config from a pre-fetched ``AgentCard``.

    Useful when the card has already been resolved (e.g. via a
    discovery service) and a network round-trip on connect can be
    skipped. ``card_url`` defaults to the first interface declared
    on the card; raises ``A2AInvalidCardError`` if neither is
    available.
    """
    resolved_url = card_url or _first_interface_url(card)
    if not resolved_url:
        raise A2AInvalidCardError(
            "AgentCard has no supported_interfaces and no `card_url` override was provided",
        )
    return cls(card_url=resolved_url, preset_card=card, **overrides)

create #

create()
Source code in autogen/beta/a2a/config.py 
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
def create(self) -> A2AClient:
    return A2AClient(
        card_url=self.card_url,
        prefer=self.prefer,
        streaming=self.streaming,
        headers=dict(self.headers) if self.headers else None,
        timeout=self.timeout,
        max_reconnects=self.max_reconnects,
        reconnect_backoff=self.reconnect_backoff,
        polling_interval=self.polling_interval,
        input_required_timeout=self.input_required_timeout,
        httpx_client_factory=self.httpx_client_factory,
        interceptors=tuple(self.interceptors),
        grpc_channel_factory=self.grpc_channel_factory,
        preset_card=self.preset_card,
        tenant=self.tenant,
        history_length=self.history_length,
    )

create_files_client #

create_files_client()
Source code in autogen/beta/config/config.py 
20
21
def create_files_client(self) -> "FilesClient":
    raise NotImplementedError(f"{type(self).__name__} does not support Files API.")