对于 AI 代理:可在 https://www.mongodb.com/zh-cn/docs/llms.txt 获取文档索引—通过在任何 URL 路径后添加 .md 可获取所有页面的 Markdown 版本。
See how MongoDB 9.0 delivers up to 2x higher throughput.
MongoDB Branding Shape
Register now >
Docs 菜单

使用代理到代理通信

代理到代理 (A2A) 通信可让一个代理在运行时发现并调用同一项目中的其他代理。编排引擎 (OE) 通过执行以下操作来代理每次调用:

  • 验证访问权限

  • 创建子执行

  • 将请求分派给目标代理

  • 返回结果

人机交互 (HITL)ACID 一致性保证端到端应用。暂停以供人工查看的目标代理会暂停调用,而不是返回失败信息。

2A 在启用时遵循以下顺序:

  1. 您将 a2a:区块添加到 agent.yaml文件中。首次执行时,平台会向 OE 注册代理的配置,使同一项目中的其他代理可以发现和调用该代理。

  2. 当操作环境向代理分派执行时,它会将短期令牌注入执行上下文。在调用其他代理时,您的代理会透明地使用该令牌。

  3. 在代理代码中,您可以使用 app.a2a_tools() 将 2A 作为 LLM 工具公开(推荐),或者直接通过 app._runtime.a2a 调用 2A客户端以实现确定性路由。

  4. OE 实施访问权限控制,将子执行链接到父执行,分派到目标,然后返回结果。

在 agent.yaml文件中添加 a2a: 部分,使代理可被发现和调用。如果省略此部分或将 a2a.enabled设立为 false,则您的代理对 A2A 调用者仍然不可见,并且与现有部署完全向后兼容。

注意

要在本地测试 2A,请在调用或通过 2A 调用的每个代理的 .env文件中将 A2A_JWT_SECRET 环境变量设立为相同的值。

以下示例显示了完全配置的 a2a: 部分:

name: insurance-agent
entrypoint: insurance_agent.main:app
a2a:
enabled: true
allowed_callers:
- "ws-router-002"
skills:
- name: policy-lookup
description: Look up insurance policy details
example_input: '{"policy_id": "POL-123"}'
example_output: '{"status": "active", "premium": 240}'
- name: get-quote
description: Generate an insurance quote
input_modes:
- "text/plain"
output_modes:
- "text/plain"

下表描述了 agent.yaml文件的 a2a: 部分中的字段:

字段
类型
默认
说明

a2a.enabled

bool

false

使代理可通过 A2A.为 false 时,将忽略所有其他 a2a 字段。

a2a.allowed_callers

list[str]

[]

允许调用此代理的工作区 ID。空列表允许在同一项目中使用任何启用 A2A 的代理。非空列表充当允许列表。

a2a.skills

list[object]

[]

代理公布以进行发现的技能。每个条目都需要一个 name 和一个 description。 example_input 和 example_output 字段是可选的。清晰、具体的描述可帮助其他代理判断该代理是否符合他们的需求。

a2a.input_modes

list[str]

[]

代理接受的多用途互联网邮件扩展 ( MIME ) 类型,示例text/plain 或 application/json。用作发现过滤。

a2a.output_modes

list[str]

[]

代理生成的MIME类型。用作发现过滤。

注意

代理的 name、skills、input_modes、output_modes 和 allowed_callers 会在初创企业时自动推送到 OE。发现结果中显示的自由文本描述和功能标签来自您通过API Gateway 工作区端点配置的工作区 AgentCard,而不是来自 agent.yaml。

您可以通过两种方式从代理代码中调用其他代理:将 A2A 公开为 LLM 工具(推荐),或者直接使用 A2A客户端。两者使用相同的根本的客户端。根据您是希望法学硕士来驱动路由决策还是您是否需要确定性控制来选择方法。

app.a2a_tools() 方法返回两个 LangChain StructuredTool 对象:discover_available_agents 和 invoke_a2a_agent。这些对象允许 LLM 自主发现并调用其他代理。将它们与现有工具绑定,以便模型可以自行路由到其他代理。

当 a2a.enabled 是 agent.yaml文件中的 false 时,app.a2a_tools() 方法会返回一个空列表,因此可以安全地包含在所有代理中。

以下示例将 2A 工具与现有工具绑定:

from agent_engine_sdk_langgraph import App
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph
from langgraph.prebuilt import ToolNode
app = App(app_name="router-agent")
@app.entrypoint
def build_agent():
llm = app.llm(ChatOpenAI(model="gpt-5.4"))
a2a = app.a2a_tools()
llm_with_tools = llm.bind_tools(
app.get_tool_schemas() + a2a
)
tool_node = ToolNode(app.get_tools() + a2a)
graph = StateGraph(...)
return graph.compile(checkpointer=app.checkpointer())

绑定到 LLM 时,该模型会按名称调用这两个工具。这些工具公开以下函数签名:

  • discover_available_agents() 返回对调用代理可见的代理的JSON列表。每个条目包括 agent_id、name、description、skills 和 capabilities。调用代理会从其自身的结果中筛选出来。

  • invoke_a2a_agent(agent_id, message, custom_headers="") 调用目标代理并以JSON形式返回结果。如果提供 custom_headers,请将其作为JSON字符串传递,示例'{"x-tenant": "acme"}'。

当您想使用确定性路由时,可以直接通过 runtime.a2a(公开客户端的属性)访问权限AgentToAgent客户端。

以下示例按技能发现客服人员并调用第一个匹配项:

agents = app._runtime.a2a.discover_agents(
skills=["policy-lookup"],
limit=10,
)
if agents:
resp = app._runtime.a2a.invoke_agent(
agent_id=agents[0].agent_id,
message="Look up policy POL-123",
skill="policy-lookup",
timeout=120,
)
if resp.status == "completed":
print(resp.result)

当 A2A 不可用时,runtime.a2a 属性会返回 None。要学习;了解何时会发生这种情况,请参阅注意事项。在调用客户端方法之前,请务必检查是否存在这种情况,如以下示例所示:

a2a = app._runtime.a2a
if a2a is None:
...

注意

通过 app._runtime.a2a 在代理代码中访问 runtime.a2a。不存在公共 app.a2a 或 app.runtime 访问器。对于大多数使用案例,app.a2a_tools() 是推荐的替代方案,因为它不依赖于私有属性。

在本节中,您可以学习;了解A2A客户端SDK,它提供了从代理代码中以编程方式发现和调用代理的方法。

要使用客户端SDK,请从 agent_engine_runner_shared.a2a 导入 2A 类型,如以下示例所示:

from agent_engine_runner_shared.a2a import (
AgentToAgent,
DiscoveredAgent,
AgentSkill,
AgentResponse,
)

AgentToAgent 是与 OE /a2a/discover 和 /a2a/invoke 端点通信的 HTTP客户端。在代理代码中,使用 runtime.a2a 中预先经过身份验证的实例,而不是直接构造此类。

以下示例显示了构造函数及其参数:

AgentToAgent(oe_url: str, auth_token: str = "")

下表描述了 AgentToAgent 构造函数参数:

Argument
说明

oe_url

OE HTTP基本URL,示例http://localhost:8000。

auth_token

A2作为 Authorization: Bearer <token> 发送的JSON web token (JWT)。使用 runtime.a2a 时自动提供。

discover_agents() 方法返回允许调用者查看的代理。当您在单个过滤参数中传递多个值时,如果代理满足其中任何一个值,则匹配。示例,skills=["a", "b"] 会返回宣传任一技能的座席。当您传递多个过滤参数时,代理必须满足所有这些参数。未启用 2A 并通过 allowed_callers 排除调用者的座席永远不会返回。

以下示例展示了此方法及其参数:

discover_agents(
project_id: str = "",
skills: list[str] | None = None,
capabilities: list[str] | None = None,
input_modes: list[str] | None = None,
limit: int = 50,
) -> list[DiscoveredAgent]

下表描述了 discover_agents() 方法的参数:

Parameter
类型
说明

project_id

str

要搜索的项目。空字符串使用调用者的项目。

skills

list[str]

按技能名称过滤。返回具有所列技能的客服人员。

capabilities

list[str]

按功能标签筛选。返回具有所列功能的代理。

input_modes

list[str]

按接受的MIME类型过滤。返回接受任何列出的MIME类型的代理。

limit

int

要返回的最大结果数。默认为 50。

invoke_agent() 方法调用目标代理并等待结果。操作环境会验证令牌、检查访问权限控制、创建子执行并进行轮询,直到代理完成、暂停以供人工查看或超时。

以下示例展示了此方法及其参数:

invoke_agent(
agent_id: str,
message: str,
skill: str = "",
timeout: float | None = None,
custom_headers: dict[str, str] | None = None,
) -> AgentResponse

下表描述了 invoke_agent() 方法的参数:

Parameter
类型
说明

agent_id

str

目标工作区ID。从 discover_agents() 获取。

message

str

要发送给目标代理的提示或消息。

skill

str

针对宣传许多技能的客服人员的可选技能提示。

timeout

浮动 |无

超时前等待的秒数。 None 使用 OE默认300 秒。 300 秒是最大超时时间。

custom_headers

dict[str, str] | None

转发到目标代理的执行上下文的键值标头。密钥不得使用保留的 a2a- 前缀。要学习;了解更多信息,请参阅转发自定义标头。

下表列出了响应类型,它们都是不可变数据类:

类
字段

DiscoveredAgent

agent_id, name, description, project_id, skills: list[AgentSkill], capabilities: list[str], input_modes: list[str], output_modes: list[str]

AgentSkill

name、description、example_input(可选)、example_output(可选)

AgentResponse

status、result(可选)、error(可选)、execution_id

下表描述了 AgentResponse.status字段的每个可能值:

状态
含义
做什么

completed

目标代理已成功完成。

读取 result。

failed

目标代理返回错误。

读取 error。

input-required

目标代理已暂停以供人工查看。

向用户或流程表明暂停。不要将此视为失败。

网络和HTTP级错误(例如来自 OE 的超时或 4xx 和 5xx 响应)会从 discover_agents() 和 invoke_agent() 方法引发 httpx.HTTPError。 status 值为 failed 的 AgentResponse 表示调用已到达 OE,但目标代理本身失败。要同时处理网络错误和代理故障,请将调用包装在 try/except 区块中并检查 status 值。

以下示例处理所有三个状态值:

resp = app._runtime.a2a.invoke_agent(
agent_id="ws-billing-001",
message="Process the payment",
)
if resp.status == "completed":
handle(resp.result)
elif resp.status == "input-required":
notify_user(
"The billing agent needs human approval before continuing."
)
else:
log.error("A2A call failed: %s", resp.error)

您可以使用 2A 将任意键值标头附加到代理调用,以传播请求范围的上下文,例如租户ID、跟踪标签或委托的 OAuth 令牌。

要将标头附加到被调用的代理,请使用 invoke_agent() 的 custom_headers 参数传递 dict。使用 invoke_a2a_agent LLM 工具时,请将标头作为JSON字符串传递。

以下示例使用直接客户端传递自定义标头:

app._runtime.a2a.invoke_agent(
agent_id="ws-billing-001",
message="Charge the customer",
custom_headers={
"x-tenant": "acme",
"oauth-token": "<delegated-token>",
},
)

以下示例使用 LLM 工具传递自定义标头:

invoke_a2a_agent(
agent_id="ws-billing-001",
message="Charge the customer",
custom_headers='{"x-tenant": "acme"}',
)

标头键不得使用保留的 a2a- 不区分大小写的前缀。如果有任何密钥以 a2a- 开头,则 OE 将拒绝具有 400 Bad Request 响应的请求。 a2a- 前缀保留用于平台路由和身份标头,例如 a2a-token 和 a2a-caller-workspace。

要读取调用代理转发的自定义标头,请使用 agent_engine_runner_shared.context 类中的 get_current_custom_headers() 方法。以下示例调用此方法:

from agent_engine_runner_shared.context import get_current_custom_headers
headers = get_current_custom_headers()
tenant = headers.get("x-tenant")

get_current_custom_headers() 返回 dict[str, str]。该方法在返回之前会剥离所有 a2a- 平台标头,因此您的代理代码永远不会看到 A2A 令牌或路由元元数据。如果调用者没有发送自定义标头,该方法将返回一个空字典。

agent.yaml文件中的 a2a.allowed_callers字段控制哪些代理可以通过以下行为发现您的代理:

  • 空列表(默认):同一项目中任何启用 2A 的代理都可以调用您的代理并在发现结果中看到它。

  • 非空列表:只有列出的工作区 ID 才能调用您的代理。所有其他代理都会收到 403 响应,并且无法在发现结果中看到您的代理。

OE 在调度之前会根据 allowed_callers 列表检查调用座席的身份。无需在代理中更改任何代码。以下示例将调用限制为单个路由器代理:

a2a:
enabled: true
allowed_callers:
- "ws-router-002"

每个 2A 调用都会自动出现在会话的跟踪面板中。无需仪器。目标代理的活动嵌套在同一会话跟踪中,因此您可以在一个位置跟踪完整的跨代理调用树。

每次调用都会呈现为标记为 A2A: <target agent name> 的子代理节点,当目标没有显示名称时,则会回退到目标的工作区ID 。该节点显示呼叫的状态和持续时间。状态开始为 running,成功时更改为 done,失败时更改为 error。选择一个节点会打开一个详细信息面板,其中显示名称、状态、持续时间、调度描述、流式输出和任何错误。

目标代理自己的工具调用、LLM 步骤和内存操作嵌套在 子代理节点下,从而跨代理重建完整的调用树。

runtime.a2a 仅当满足以下所有条件时才返回客户端:

  1. 代理代码在代理沙箱中运行。 A2A 无法从工具沙箱中运行的代码中获得。

  2. 执行上下文中存在 OE URL 。

  3. 传入标头中存在 a2a-token。配置 A2A 后,OE 会在每次调度时自动注入此令牌。

不满足任何条件时,runtime.a2a 会返回 None。使用 app.a2a_tools() 时,这些工具会返回结构化错误JSON,而不是引发异常。

默认下,A2A 令牌会在 5 分钟后过期。对于典型的请求与响应流,这就足够了。运行时间超过令牌生命周期的代理可能会收到针对 2A 调用的 401 响应。自动令牌刷新尚未实现。将长时间运行的调用的 401 响应视为暂时性故障。

discover_available_agents LLM 工具会从结果中筛选出调用代理自己的工作区,以便模型不会调用自身。如果您使用原始 discover_agents()客户端方法,则 OE 可能会在结果中包含您自己的代理。如果需要,请自行过滤掉。

在当前发布中,2发现和调用仅限于同一项目中的代理。跨项目路由尚未实现。

平台会自动记录并关联每个 2A 调用。您无需添加仪器。目标作为通过 parent_execution_id 链接到调用者的子执行运行。共享的 root_session_id 将完整的跨代理调用树分组在原始用户会话下。

以下示例显示了使用 LLM 查找并委托给专家代理的路由器代理。 agent.yaml文件启用 A2A 并声明 route技能:

agent.yaml
name: router-agent
entrypoint: router_agent.main:app
a2a:
enabled: true
skills:
- name: route
description: Route a user request to the best specialist agent

以下代理代码将 A2A 工具与其自己的工具绑定在一起,并让 LLM处理路由:

main.py
from agent_engine_sdk_langgraph import App
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, MessagesState
from langgraph.prebuilt import ToolNode
app = App(app_name="router-agent")
@app.entrypoint
def build_agent():
llm = app.llm(ChatOpenAI(model="gpt-5.4"))
a2a = app.a2a_tools()
llm_with_tools = llm.bind_tools(
app.get_tool_schemas() + a2a
)
def call_model(state: MessagesState):
return {
"messages": [llm_with_tools.invoke(state["messages"])]
}
graph = StateGraph(MessagesState)
graph.add_node("model", call_model)
graph.add_node("tools", ToolNode(app.get_tools() + a2a))
graph.set_entry_point("model")
graph.add_conditional_edges(
"model",
lambda s: (
"tools"
if s["messages"][-1].tool_calls
else "__end__"
),
)
graph.add_edge("tools", "model")
return graph.compile(checkpointer=app.checkpointer())

在运行时,LLM 会发现广告匹配技能的专业代理,并通过 invoke_a2a_agent 进行调用,OE 会在执行访问权限控制并记录调用双方的同时代理该调用。

要确定性路由而不是依赖 LLM,请直接在图表节点内调用客户端:

def delegate(state):
a2a = app._runtime.a2a
if a2a is None:
return {"messages": [("ai", "A2A is unavailable.")]}
resp = a2a.invoke_agent(
agent_id="ws-insurance-001",
message=state["messages"][-1].content,
skill="policy-lookup",
custom_headers={"x-request-source": "router-agent"},
)
text = (
resp.result
if resp.status == "completed"
else f"({resp.status}) {resp.error}"
)
return {"messages": [("ai", text)]}

启用助手到助手的通信后,您可以浏览以下指南以学习;了解有关相关任务的更多信息: