Skip to content

A2AAgentSpec

A2AAgentSpec — specification for a registered A2A peer agent.

A2AAgentSpec

Specification for a registered A2A peer agent.

Each A2AAgentSpec entry registers a remote A2A-compliant peer under name, derived directly from agent_card.name — there is no separate local alias, since the card is remote/peer-controlled data like the rest of the spec's inputs. name serves as the registry key on the coordinator and as the enum value UseA2AAgentTool presents to the LLM for dispatch. Discovery (fetching the peer's AgentCard) is eager, at spec construction — the spec holds a fully resolved card, not a lazy reference to one.

The spec is pure data: it never constructs or holds a live SDK Client. Connecting to the peer is UseA2AAgentTool's job, done fresh on each dispatch from this spec's url/headers/ agent_card.

A plain class, not a pydantic BaseModel: agent_card is a non-pydantic SDK type, so BaseModel bought only arbitrary_types_allowed=True overhead here.

url is likewise derived from the card rather than passed independently: it should match agent_card.supported_interfaces[0].url, not the URL from_url fetched the card from — the two can legitimately differ (that's the reason supported_interfaces exists as a separate list rather than a single top-level field). A card declaring more than one interface is a real possibility the protocol allows for, but this spec doesn't attempt to disambiguate between them — it always takes the first. Our own server (LLMAgentA2AExecutor) only ever publishes one, so this is a deliberate simplification, not an oversight.

Attributes:

Name Type Description
name

Registry key for this A2A agent, taken from agent_card.name. Appears as an enum value in UseA2AAgentTool's dispatch schema.
url

Base URL of the remote A2A peer. Should match agent_card.supported_interfaces[0].url.
agent_card

The peer's resolved AgentCard, fetched eagerly at construction time.
headers

Optional HTTP headers (e.g. auth) sent on requests to this peer, both for card resolution and for dispatch, mirroring MCPToolProvider.streamable_http_headers. May carry credentials (e.g. Authorization); stored as a plain str, not masked. Production use should wrap values in pydantic's SecretStr so a stray repr/log/dump can't leak them — left as a callout rather than built in, to keep this an educational framework rather than a production-hardened one.
timeout

Seconds UseA2AAgentTool allows a dispatch to this peer before timing out. Defaults to 60.0 rather than httpx's own default (5.0s, applied to connect/read/ write/pool combined) — too low for a peer that makes one or more LLM calls per task. Explicitly setting this to None disables the timeout entirely (unbounded), since it is passed straight through to httpx.AsyncClient.
Source code in src/llm_agents_from_scratch/a2a/client/spec.py 
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
class A2AAgentSpec:
    """Specification for a registered A2A peer agent.

    Each ``A2AAgentSpec`` entry registers a remote A2A-compliant peer under
    ``name``, derived directly from ``agent_card.name`` — there is no
    separate local alias, since the card is remote/peer-controlled data
    like the rest of the spec's inputs. ``name`` serves as the registry key
    on the coordinator and as the enum value ``UseA2AAgentTool`` presents
    to the LLM for dispatch. Discovery (fetching the peer's ``AgentCard``)
    is eager, at spec construction — the spec holds a fully resolved card,
    not a lazy reference to one.

    The spec is pure data: it never constructs or holds a live SDK
    ``Client``. Connecting to the peer is ``UseA2AAgentTool``'s job, done
    fresh on each dispatch from this spec's ``url``/``headers``/
    ``agent_card``.

    A plain class, not a pydantic ``BaseModel``: ``agent_card`` is a
    non-pydantic SDK type, so ``BaseModel`` bought only
    ``arbitrary_types_allowed=True`` overhead here.

    ``url`` is likewise derived from the card rather than passed
    independently: it should match
    ``agent_card.supported_interfaces[0].url``, not the URL ``from_url``
    fetched the card from — the two can legitimately differ (that's the
    reason ``supported_interfaces`` exists as a separate list rather than
    a single top-level field). A card declaring more than one interface is
    a real possibility the protocol allows for, but this spec doesn't
    attempt to disambiguate between them — it always takes the first. Our
    own server (``LLMAgentA2AExecutor``) only ever publishes one, so
    this is a deliberate simplification, not an oversight.

    Attributes:
        name: Registry key for this A2A agent, taken from
            ``agent_card.name``. Appears as an enum value in
            ``UseA2AAgentTool``'s dispatch schema.
        url: Base URL of the remote A2A peer. Should match
            ``agent_card.supported_interfaces[0].url``.
        agent_card: The peer's resolved ``AgentCard``, fetched eagerly at
            construction time.
        headers: Optional HTTP headers (e.g. auth) sent on requests to
            this peer, both for card resolution and for dispatch, mirroring
            ``MCPToolProvider.streamable_http_headers``. May carry
            credentials (e.g. ``Authorization``); stored as a plain
            ``str``, not masked. Production use should wrap values in
            pydantic's ``SecretStr`` so a stray repr/log/dump can't leak
            them — left as a callout rather than built in, to keep this
            an educational framework rather than a production-hardened
            one.
        timeout: Seconds ``UseA2AAgentTool`` allows a dispatch to this
            peer before timing out. Defaults to 60.0 rather than
            ``httpx``'s own default (5.0s, applied to connect/read/
            write/pool combined) — too low for a peer that makes one or
            more LLM calls per task. Explicitly setting this to
            ``None`` disables the timeout entirely (unbounded), since
            it is passed straight through to ``httpx.AsyncClient``.
    """

    def __init__(
        self,
        name: str,
        url: str,
        agent_card: AgentCard,
        headers: dict[str, str] | None = None,
        timeout: float | None = 60.0,
    ) -> None:
        """Initialize an A2AAgentSpec.

        Args:
            name: Registry key for this A2A agent, taken from
                ``agent_card.name``.
            url: Base URL of the remote A2A peer.
            agent_card: The peer's resolved ``AgentCard``.
            headers: Optional HTTP headers sent on requests to this
                peer. Defaults to ``None``.
            timeout: Seconds ``UseA2AAgentTool`` allows a dispatch to
                this peer before timing out. ``None`` disables the
                timeout entirely. Defaults to ``60.0``.
        """
        self.name = name
        self.url = url
        self.agent_card = agent_card
        self.headers = headers
        self.timeout = timeout

    def __repr__(self) -> str:
        """Readable repr listing every field."""
        return (
            f"{type(self).__name__}("
            f"name={self.name!r}, "
            f"url={self.url!r}, "
            f"headers={self.headers!r}, "
            f"agent_card={self.agent_card!r}, "
            f"timeout={self.timeout!r})"
        )

    @classmethod
    def from_agent_card(
        cls,
        agent_card: AgentCard,
        headers: dict[str, str] | None = None,
        timeout: float | None = 60.0,
    ) -> Self:
        """Builds a spec from an ``AgentCard`` already in hand.

        Sync — covers cached cards, self-built cards, and test fixtures,
        with no network access performed here.

        Args:
            agent_card: The peer's already-resolved ``AgentCard``.
            headers: Optional HTTP headers sent on requests to this peer.
            timeout: Seconds ``UseA2AAgentTool`` allows a dispatch to
                this peer before timing out.

        Returns:
            A2AAgentSpec: The constructed spec.

        Raises:
            A2AAgentCardMissingInterfaceError: If ``agent_card`` declares
                no ``supported_interfaces`` to derive a dispatch URL from.
        """
        if not agent_card.supported_interfaces:
            raise A2AAgentCardMissingInterfaceError(
                f"AgentCard '{agent_card.name}' declares no "
                "supported_interfaces; cannot determine a dispatch URL.",
            )
        return cls(
            name=agent_card.name,
            url=agent_card.supported_interfaces[0].url,
            agent_card=agent_card,
            headers=headers,
            timeout=timeout,
        )

    @classmethod
    async def from_url(
        cls,
        url: str,
        headers: dict[str, str] | None = None,
        agent_card_path: str | None = None,
        timeout: float | None = 60.0,
    ) -> Self:
        """Fetches the peer's ``AgentCard`` from ``url``, then builds a spec.

        Async — resolves the card over the wire via ``A2ACardResolver``
        before delegating to ``from_agent_card``. An unreachable peer
        raises the underlying ``httpx`` error to the caller. ``url`` here
        is only the card-resolution endpoint — the constructed spec's own
        ``url`` comes from the fetched card's ``supported_interfaces``
        instead, which can legitimately differ.

        Args:
            url: Base URL to resolve the peer's well-known ``AgentCard``
                from.
            headers: Optional HTTP headers sent on requests to this peer.
            agent_card_path: Optional override for the well-known agent
                card path. ``None`` uses the SDK's own default.
            timeout: Timeout in seconds, applied both to this card
                resolution and, via the returned spec, to every future
                ``UseA2AAgentTool`` dispatch to this peer.

        Returns:
            A2AAgentSpec: The constructed spec.

        Raises:
            A2AAgentCardMissingInterfaceError: If the fetched card
                declares no ``supported_interfaces``.
        """
        async with httpx.AsyncClient(
            headers=headers,
            timeout=timeout,
        ) as httpx_client:
            resolver_kwargs: dict[str, str] = {}
            if agent_card_path is not None:
                resolver_kwargs["agent_card_path"] = agent_card_path
            resolver = A2ACardResolver(
                httpx_client=httpx_client,
                base_url=url,
                **resolver_kwargs,
            )
            agent_card = await resolver.get_agent_card()

        return cls.from_agent_card(
            agent_card=agent_card,
            headers=headers,
            timeout=timeout,
        )

    def catalog(self) -> str:
        """Return XML entry for this spec in the a2a agents catalog.

        Nests the peer's declared ``AgentSkill``s (its ``agent_card.skills``)
        as an ``<a2a_skills>`` block, giving the coordinator finer-grained
        routing signal than the top-level description alone. Omitted
        entirely when the peer declares no skills.
        """
        skills_block = ""
        if self.agent_card.skills:
            entries = "\n".join(
                CATALOG_INDIVIDUAL_A2A_SKILL_TEMPLATE.format(name=skill.name)
                for skill in self.agent_card.skills
            )
            skills_block = CATALOG_A2A_SKILLS_TEMPLATE.format(skills=entries)
        return CATALOG_SPEC_TEMPLATE.format(
            name=self.name,
            description=self.agent_card.description,
            skills=skills_block,
        )

__init__ 

__init__(name, url, agent_card, headers=None, timeout=60.0)

Initialize an A2AAgentSpec.

Parameters:

Name Type Description Default
name str

Registry key for this A2A agent, taken from agent_card.name.

 required
url str

Base URL of the remote A2A peer.

 required
agent_card AgentCard

The peer's resolved AgentCard.

 required
headers dict[str, str] | None

Optional HTTP headers sent on requests to this peer. Defaults to None.

 None
timeout float | None

Seconds UseA2AAgentTool allows a dispatch to this peer before timing out. None disables the timeout entirely. Defaults to 60.0.

 60.0
Source code in src/llm_agents_from_scratch/a2a/client/spec.py 
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def __init__(
    self,
    name: str,
    url: str,
    agent_card: AgentCard,
    headers: dict[str, str] | None = None,
    timeout: float | None = 60.0,
) -> None:
    """Initialize an A2AAgentSpec.

    Args:
        name: Registry key for this A2A agent, taken from
            ``agent_card.name``.
        url: Base URL of the remote A2A peer.
        agent_card: The peer's resolved ``AgentCard``.
        headers: Optional HTTP headers sent on requests to this
            peer. Defaults to ``None``.
        timeout: Seconds ``UseA2AAgentTool`` allows a dispatch to
            this peer before timing out. ``None`` disables the
            timeout entirely. Defaults to ``60.0``.
    """
    self.name = name
    self.url = url
    self.agent_card = agent_card
    self.headers = headers
    self.timeout = timeout

__repr__ 

__repr__()

Readable repr listing every field.

Source code in src/llm_agents_from_scratch/a2a/client/spec.py 
102
103
104
105
106
107
108
109
110
111
def __repr__(self) -> str:
    """Readable repr listing every field."""
    return (
        f"{type(self).__name__}("
        f"name={self.name!r}, "
        f"url={self.url!r}, "
        f"headers={self.headers!r}, "
        f"agent_card={self.agent_card!r}, "
        f"timeout={self.timeout!r})"
    )

from_agent_card classmethod 

from_agent_card(agent_card, headers=None, timeout=60.0)

Builds a spec from an AgentCard already in hand.

Sync — covers cached cards, self-built cards, and test fixtures, with no network access performed here.

Parameters:

Name Type Description Default
agent_card AgentCard

The peer's already-resolved AgentCard.

 required
headers dict[str, str] | None

Optional HTTP headers sent on requests to this peer.

 None
timeout float | None

Seconds UseA2AAgentTool allows a dispatch to this peer before timing out.

 60.0

Returns:

Name Type Description
A2AAgentSpec Self

The constructed spec.

Raises:

Type Description
A2AAgentCardMissingInterfaceError

If agent_card declares no supported_interfaces to derive a dispatch URL from.
Source code in src/llm_agents_from_scratch/a2a/client/spec.py 
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
@classmethod
def from_agent_card(
    cls,
    agent_card: AgentCard,
    headers: dict[str, str] | None = None,
    timeout: float | None = 60.0,
) -> Self:
    """Builds a spec from an ``AgentCard`` already in hand.

    Sync — covers cached cards, self-built cards, and test fixtures,
    with no network access performed here.

    Args:
        agent_card: The peer's already-resolved ``AgentCard``.
        headers: Optional HTTP headers sent on requests to this peer.
        timeout: Seconds ``UseA2AAgentTool`` allows a dispatch to
            this peer before timing out.

    Returns:
        A2AAgentSpec: The constructed spec.

    Raises:
        A2AAgentCardMissingInterfaceError: If ``agent_card`` declares
            no ``supported_interfaces`` to derive a dispatch URL from.
    """
    if not agent_card.supported_interfaces:
        raise A2AAgentCardMissingInterfaceError(
            f"AgentCard '{agent_card.name}' declares no "
            "supported_interfaces; cannot determine a dispatch URL.",
        )
    return cls(
        name=agent_card.name,
        url=agent_card.supported_interfaces[0].url,
        agent_card=agent_card,
        headers=headers,
        timeout=timeout,
    )

from_url async classmethod 

from_url(
    url, headers=None, agent_card_path=None, timeout=60.0
)

Fetches the peer's AgentCard from url, then builds a spec.

Async — resolves the card over the wire via A2ACardResolver before delegating to from_agent_card. An unreachable peer raises the underlying httpx error to the caller. url here is only the card-resolution endpoint — the constructed spec's own url comes from the fetched card's supported_interfaces instead, which can legitimately differ.

Parameters:

Name Type Description Default
url str

Base URL to resolve the peer's well-known AgentCard from.

 required
headers dict[str, str] | None

Optional HTTP headers sent on requests to this peer.

 None
agent_card_path str | None

Optional override for the well-known agent card path. None uses the SDK's own default.

 None
timeout float | None

Timeout in seconds, applied both to this card resolution and, via the returned spec, to every future UseA2AAgentTool dispatch to this peer.

 60.0

Returns:

Name Type Description
A2AAgentSpec Self

The constructed spec.

Raises:

Type Description
A2AAgentCardMissingInterfaceError

If the fetched card declares no supported_interfaces.
Source code in src/llm_agents_from_scratch/a2a/client/spec.py 
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
@classmethod
async def from_url(
    cls,
    url: str,
    headers: dict[str, str] | None = None,
    agent_card_path: str | None = None,
    timeout: float | None = 60.0,
) -> Self:
    """Fetches the peer's ``AgentCard`` from ``url``, then builds a spec.

    Async — resolves the card over the wire via ``A2ACardResolver``
    before delegating to ``from_agent_card``. An unreachable peer
    raises the underlying ``httpx`` error to the caller. ``url`` here
    is only the card-resolution endpoint — the constructed spec's own
    ``url`` comes from the fetched card's ``supported_interfaces``
    instead, which can legitimately differ.

    Args:
        url: Base URL to resolve the peer's well-known ``AgentCard``
            from.
        headers: Optional HTTP headers sent on requests to this peer.
        agent_card_path: Optional override for the well-known agent
            card path. ``None`` uses the SDK's own default.
        timeout: Timeout in seconds, applied both to this card
            resolution and, via the returned spec, to every future
            ``UseA2AAgentTool`` dispatch to this peer.

    Returns:
        A2AAgentSpec: The constructed spec.

    Raises:
        A2AAgentCardMissingInterfaceError: If the fetched card
            declares no ``supported_interfaces``.
    """
    async with httpx.AsyncClient(
        headers=headers,
        timeout=timeout,
    ) as httpx_client:
        resolver_kwargs: dict[str, str] = {}
        if agent_card_path is not None:
            resolver_kwargs["agent_card_path"] = agent_card_path
        resolver = A2ACardResolver(
            httpx_client=httpx_client,
            base_url=url,
            **resolver_kwargs,
        )
        agent_card = await resolver.get_agent_card()

    return cls.from_agent_card(
        agent_card=agent_card,
        headers=headers,
        timeout=timeout,
    )

catalog 

catalog()

Return XML entry for this spec in the a2a agents catalog.

Nests the peer's declared AgentSkills (its agent_card.skills) as an <a2a_skills> block, giving the coordinator finer-grained routing signal than the top-level description alone. Omitted entirely when the peer declares no skills.

Source code in src/llm_agents_from_scratch/a2a/client/spec.py 
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
def catalog(self) -> str:
    """Return XML entry for this spec in the a2a agents catalog.

    Nests the peer's declared ``AgentSkill``s (its ``agent_card.skills``)
    as an ``<a2a_skills>`` block, giving the coordinator finer-grained
    routing signal than the top-level description alone. Omitted
    entirely when the peer declares no skills.
    """
    skills_block = ""
    if self.agent_card.skills:
        entries = "\n".join(
            CATALOG_INDIVIDUAL_A2A_SKILL_TEMPLATE.format(name=skill.name)
            for skill in self.agent_card.skills
        )
        skills_block = CATALOG_A2A_SKILLS_TEMPLATE.format(skills=entries)
    return CATALOG_SPEC_TEMPLATE.format(
        name=self.name,
        description=self.agent_card.description,
        skills=skills_block,
    )