Skip to content

ant_ai.a2a.agent

A2AAgentTool pydantic-model

Bases: Tool

A tool that provides an interface to interact with the Agent via the A2A protocol.

Show JSON schema: 
{
  "$defs": {
    "A2AConfig": {
      "description": "Configuration used to set the connection with the A2A server.",
      "properties": {
        "endpoint": {
          "description": "The URL of the A2A server to connect to.",
          "title": "Endpoint",
          "type": "string"
        },
        "timeout": {
          "$ref": "#/$defs/TimeoutTypes",
          "description": "Timeout configuration for the A2A client. Can be a float (total timeout) or a dict with connect/read/write/pool timeouts. See httpx.Timeout for more details."
        },
        "agent_card_path": {
          "default": "/.well-known/agent-card.json",
          "description": "The path, on the remote url, to the agent card.",
          "title": "Agent Card Path",
          "type": "string"
        },
        "supported_protocol_bindings": {
          "default": [
            "JSONRPC",
            "HTTP+JSON"
          ],
          "description": "The supported A2A protocol bindings.",
          "items": {
            "type": "string"
          },
          "title": "Supported Protocol Bindings",
          "type": "array"
        },
        "streaming": {
          "default": true,
          "description": "Whether to enable streaming.",
          "title": "Streaming",
          "type": "boolean"
        },
        "propagate_trace_context": {
          "default": true,
          "description": "Whether to inject the current trace context into outbound A2A requests. Set to False when calling third-party agents you do not own.",
          "title": "Propagate Trace Context",
          "type": "boolean"
        }
      },
      "required": [
        "endpoint"
      ],
      "title": "A2AConfig",
      "type": "object"
    },
    "TimeoutTypes": {
      "anyOf": [
        {
          "type": "number"
        },
        {
          "maxItems": 4,
          "minItems": 4,
          "prefixItems": [
            {
              "anyOf": [
                {
                  "type": "number"
                },
                {
                  "type": "null"
                }
              ]
            },
            {
              "anyOf": [
                {
                  "type": "number"
                },
                {
                  "type": "null"
                }
              ]
            },
            {
              "anyOf": [
                {
                  "type": "number"
                },
                {
                  "type": "null"
                }
              ]
            },
            {
              "anyOf": [
                {
                  "type": "number"
                },
                {
                  "type": "null"
                }
              ]
            }
          ],
          "type": "array"
        },
        {
          "type": "null"
        }
      ]
    }
  },
  "description": "A tool that provides an interface to interact with the Agent via the A2A protocol.",
  "properties": {
    "name": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Tool name.",
      "title": "Name"
    },
    "description": {
      "anyOf": [
        {
          "type": "string"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "Tool description. Is used by the LLM to decide whether to call or not the specific tool.",
      "title": "Description"
    },
    "parameters": {
      "anyOf": [
        {
          "additionalProperties": true,
          "type": "object"
        },
        {
          "type": "null"
        }
      ],
      "default": null,
      "description": "The parameters needed by the tool. This is a self-constructed field.",
      "title": "Parameters"
    },
    "config": {
      "$ref": "#/$defs/A2AConfig",
      "description": "Configuration for the A2A client."
    },
    "agent_input_description": {
      "default": "Message to send to the agent. Contains all the necessary information to answer the question in a clear way.",
      "description": "Description of the input to the agent. This description will be used by the agent to generate the prompt for remote agent request.",
      "title": "Agent Input Description",
      "type": "string"
    }
  },
  "required": [
    "config"
  ],
  "title": "A2AAgentTool",
  "type": "object"
}

Fields:

Validators:

  • _set_defaults
Source code in src/ant_ai/a2a/agent.py 
 16
 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
class A2AAgentTool(Tool):
    """
    A tool that provides an interface to interact with the Agent via the A2A protocol.
    """

    config: A2AConfig = Field(..., description="Configuration for the A2A client.")
    agent_input_description: str = Field(
        default="Message to send to the agent. Contains all the necessary information to answer the question in a clear way.",
        description="Description of the input to the agent. This description will be used by the agent to generate the prompt for remote agent request.",
    )

    _a2a: A2AClient | None = PrivateAttr(default=None)
    _initialized: bool = PrivateAttr(default=False)
    _agent_card: AgentCard | None = PrivateAttr(default=None)
    _last_task_id: str | None = PrivateAttr(default=None)

    @model_validator(mode="after")
    def _set_defaults(self) -> A2AAgentTool:
        return self

    def _ensure_a2a(self) -> None:
        if self._a2a is None:
            self._a2a = A2AClient(config=self.config)
            self._a2a._agent_card: AgentCard | None = self._agent_card

    def _init_metadata(self, agent_card: AgentCard) -> None:
        """Set Tool metadata exactly once from an AgentCard."""
        if not self.name:
            self.name: str = agent_card.name
        if not self.description:
            self.description: str = self._create_agent_description(agent_card)

        self.parameters: dict[str, Any] = {
            "type": "object",
            "properties": {
                "message": {
                    "type": "string",
                    "description": (self.agent_input_description),
                }
            },
            "required": ["message"],
        }

    async def _ensure_initialized(self) -> None:
        """Fetch AgentCard (if needed) and set metadata/_func exactly once."""
        if self._initialized:
            return

        self._ensure_a2a()
        if self._a2a is None:
            raise RuntimeError("A2A client not initialized")

        agent_card: AgentCard = self._agent_card or await self._a2a.get_agent_card()
        self._agent_card: AgentCard = agent_card
        self._init_metadata(agent_card)

        self._attach_func()
        self._initialized = True

    def _attach_func(self) -> None:
        """Attach the call function to the _func (single callable Tool)."""

        async def _call_remote(message: str) -> str:
            await self._ensure_initialized()
            self._ensure_a2a()
            if self._a2a is None:
                raise RuntimeError("A2A client not initialized")

            last_text: str = ""
            async for ev in self._a2a.send_message(
                message, context_id=current_session_id.get(None)
            ):
                if ev.content:
                    last_text: str = ev.content
                if isinstance(ev, (FinalAnswerEvent, ClarificationNeededEvent)):
                    break
            return last_text

        self._func = _call_remote

    @overload
    @classmethod
    def from_config(cls, config: A2AConfig, agent_card: AgentCard) -> A2AAgentTool:
        """Creates an A2A agent tool from a configuration and an agent card.

        Returns:
            An A2A agent tool.
        """
        ...

    @overload
    @classmethod
    def from_config(
        cls, config: A2AConfig, agent_card: None = None
    ) -> Awaitable[A2AAgentTool]:
        """Creates an A2A agent tool from a configuration.

        Args:
            config: _description_
            agent_card: _description_. Defaults to None.

        Returns:
            An awaitable of an A2A agent tool.
        """
        ...

    @classmethod
    def from_config(
        cls, config: A2AConfig, agent_card: AgentCard | None = None
    ) -> A2AAgentTool | Awaitable[A2AAgentTool]:
        """Creates an A2A agent tool from a configuration and an optional agent card. If no agent card is provided, the tool will be initialized asynchronously and the tool will be returned as an awaitable tool.

        Args:
            config: _description_
            agent_card: _description_. Defaults to None.

        Returns:
            An A2A agent tool or an awaitable of an A2A agent tool.
        """
        tool: A2AAgentTool = cls(name=None, description=None, config=config)

        if agent_card is not None:
            tool._agent_card: AgentCard = agent_card
            tool._ensure_a2a()
            tool._init_metadata(agent_card)
            tool._attach_func()
            tool._initialized = True
            return tool

        async def _build() -> A2AAgentTool:
            await tool._ensure_initialized()
            return tool

        return _build()

    def _create_agent_description(self, agent_card: AgentCard) -> str:
        parts: list[str] = [
            agent_card.description,
        ]

        if agent_card.skills:
            parts += ["", "### Available Skills", ""]
            for skill in agent_card.skills:
                parts.append(f"**{skill.name}**")
                parts.append(skill.description)
                if skill.tags:
                    parts.append(f"Tags: {', '.join(skill.tags)}")
                examples = list(skill.examples) if skill.examples else []
                if examples:
                    parts.append("Examples:")
                    for ex in examples:
                        parts.append(f"  - {ex}")
                parts.append("")

        return "\n".join(parts)

    @property
    def is_namespace(self) -> Literal[False]:
        return False

    def _sid(self) -> str:
        return current_session_id.get()

config pydantic-field 

config: A2AConfig

Configuration for the A2A client.

agent_input_description pydantic-field 

agent_input_description: str = "Message to send to the agent. Contains all the necessary information to answer the question in a clear way."

Description of the input to the agent. This description will be used by the agent to generate the prompt for remote agent request.

from_config classmethod 

from_config(
    config: A2AConfig, agent_card: AgentCard
) -> A2AAgentTool

from_config(
    config: A2AConfig, agent_card: None = None
) -> Awaitable[A2AAgentTool]

from_config(
    config: A2AConfig, agent_card: AgentCard | None = None
) -> A2AAgentTool | Awaitable[A2AAgentTool]

Creates an A2A agent tool from a configuration and an optional agent card. If no agent card is provided, the tool will be initialized asynchronously and the tool will be returned as an awaitable tool.

Parameters:

Name Type Description Default
config A2AConfig

description

 required
agent_card AgentCard | None

description. Defaults to None.

 None

Returns:

Type Description
A2AAgentTool | Awaitable[A2AAgentTool]

An A2A agent tool or an awaitable of an A2A agent tool.
Source code in src/ant_ai/a2a/agent.py 
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_config(
    cls, config: A2AConfig, agent_card: AgentCard | None = None
) -> A2AAgentTool | Awaitable[A2AAgentTool]:
    """Creates an A2A agent tool from a configuration and an optional agent card. If no agent card is provided, the tool will be initialized asynchronously and the tool will be returned as an awaitable tool.

    Args:
        config: _description_
        agent_card: _description_. Defaults to None.

    Returns:
        An A2A agent tool or an awaitable of an A2A agent tool.
    """
    tool: A2AAgentTool = cls(name=None, description=None, config=config)

    if agent_card is not None:
        tool._agent_card: AgentCard = agent_card
        tool._ensure_a2a()
        tool._init_metadata(agent_card)
        tool._attach_func()
        tool._initialized = True
        return tool

    async def _build() -> A2AAgentTool:
        await tool._ensure_initialized()
        return tool

    return _build()