agentscope.agentA2AAgent
class A2AAgentagentscope/agent/_a2a_agent.py:189–661A2A 智能体，驱动智能体的推理、工具调用与会话执行。位于 agentscope.agent 包。核心成员包括 aclose、reply_stream、reply、compress_context、observe。
This class intentionally provides Agent-like interaction methods without
inheriting :class:agentscope.agent.Agent. A local Agent owns a model,
toolkit, state, and reasoning loop; this adapter delegates those concerns
to the remote A2A server and owns only the remote conversation
(context_id) and the Task the next message continues (task_id),
both held in :class:agentscope.state.A2AAgentState.
Each A2A Part of a response becomes one content block: text Parts become
:class:~agentscope.message.TextBlock, raw byte and URL Parts become
:class:~agentscope.message.DataBlock.
The adapter owns its A2A client and closes it in :meth:aclose, so it is
single-use: once closed it cannot be reopened.
构造器摘要Constructor Summary
| 构造器Constructor | 说明Description |
|---|---|
def __init__(agent_card: AgentCard, *, client: Client | None = None, state: A2AAgentState | None = None) -> None | 初始化 `A2AAgent` 实例。 Initialize the A2A agent adapter. |
方法摘要Method Summary
| 方法Method | 说明Description |
|---|---|
async def __aenter__() -> A2AAgentasync | 进入异步上下文并返回自身。 Enter the asynchronous context manager. |
async def __aexit__(*_: Any) -> Noneasync | 退出异步上下文并清理资源。 Close the owned client when leaving the context manager. |
async def aclose() -> Noneasync | 返回 `aclose` 结果。 Close the owned A2A client. Repeated calls are safe. |
async def reply_stream(inputs: Msg | list[Msg] | None = None, yield_final_msg: bool = False) -> AsyncGenerator[AgentEve…async | 流式回复输入消息，并逐个产出事件或消息。 Send input and stream the remote reply as AgentScope events. |
async def reply(inputs: Msg | list[Msg] | None = None) -> Msgasync | 回复输入消息并返回最终消息。 Send input and return the final assistant message. |
async def compress_context(*args: Any, **kwargs: Any) -> Noneasync | 在上下文超过阈值时压缩历史消息。 Do nothing because the remote A2A server owns its context. |
async def observe(msgs: Msg | list[Msg] | None = None) -> Noneasync | 接收外部观察消息并写入上下文。 Cache messages to send ahead of the next reply's own input. |
构造器详情Constructor Detail
A2AAgent
def __init__(agent_card: AgentCard, *, client: Client | None = None, state: A2AAgentState | None = None) -> None初始化 A2AAgent 实例。
Initialize the A2A agent adapter.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
agent_card | AgentCard | 智能体卡片（类型 `AgentCard`）。The remote Agent Card, used both to identify the peer (its `name` becomes this adapter's `name`) and to select a transport. The SDK picks the newest protocol version the card advertises for the chosen binding, falling back to its A2A 0.3 compatibility transport when that is all the peer offers. |
clientkw默认default None | Client | None | 客户端（类型 `Client | None`）。An official SDK client, e.g. one configured for gRPC or with custom auth. If omitted, a streaming client is built from the card, which then requires a `JSONRPC` or `HTTP+JSON` interface. The adapter owns the client either way and closes it in :meth:`aclose`. |
statekw默认default None | A2AAgentState | None | 状态对象（类型 `A2AAgentState | None`）。An existing state to resume, e.g. one saved from an earlier adapter, so its `context_id` continues the same remote conversation. A fresh state is created when omitted. |
agentscope/agent/_a2a_agent.py:207–261方法详情Method Detail
__aenter__
async def __aenter__() -> A2AAgent进入异步上下文并返回自身。
Leaving the block closes the owned client, and nothing reopens it, so an adapter can only be entered before it is closed.
agentscope/agent/_a2a_agent.py:263–271__aexit__
async def __aexit__(*_: Any) -> None退出异步上下文并清理资源。
Close the owned client when leaving the context manager.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
_ | Any | 组件（类型 `Any`）。The component. |
agentscope/agent/_a2a_agent.py:273–275aclose
async def aclose() -> None返回 aclose 结果。
Close the owned A2A client.
agentscope/agent/_a2a_agent.py:277–281reply_stream
async def reply_stream(inputs: Msg | list[Msg] | None = None, yield_final_msg: bool = False) -> AsyncGenerator[AgentEvent | Msg, None]流式回复输入消息，并逐个产出事件或消息。
Send input and stream the remote reply as AgentScope events.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
inputs默认default None | Msg | list[Msg] | None | 输入消息（类型 `Msg | list[Msg] | None`）。The messages to send, preceded by the observed ones. |
yield_final_msg默认default False | bool | yield 最终消息（类型 `bool`）。Whether to yield the final reply message after the events. |
agentscope/agent/_a2a_agent.py:283–303reply
async def reply(inputs: Msg | list[Msg] | None = None) -> Msg回复输入消息并返回最终消息。
Send input and return the final assistant message.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
inputs默认default None | Msg | list[Msg] | None | 输入消息（类型 `Msg | list[Msg] | None`）。The messages to send, preceded by the observed ones. |
agentscope/agent/_a2a_agent.py:305–322compress_context
async def compress_context(*args: Any, **kwargs: Any) -> None在上下文超过阈值时压缩历史消息。
The arguments are accepted for interface compatibility with :class:agentscope.agent.Agent.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
args | Any | 参数（类型 `Any`）。The args. |
kwargs | Any | 关键字参数（类型 `Any`）。The kwargs. |
agentscope/agent/_a2a_agent.py:630–644observe
async def observe(msgs: Msg | list[Msg] | None = None) -> None接收外部观察消息并写入上下文。
Cache messages to send ahead of the next reply's own input.
| 名称Name | 类型Type | 说明Description |
|---|---|---|
msgs默认default None | Msg | list[Msg] | None | 消息列表（类型 `Msg | list[Msg] | None`）。The messages to remember. `None` is ignored. |
agentscope/agent/_a2a_agent.py:646–661