主题
Agent Server 中的 A2A 端点
使用 A2A 协议在 LangSmith 中启用带分布式追踪的智能体间通信。
Agent2Agent (A2A) 是 Google 的协议，用于实现对话式 AI 智能体之间的通信。LangSmith 实现了 A2A 支持，让你的智能体可以通过标准化协议与其他 A2A 兼容的智能体通信。
Agent Server 中的 A2A 端点位于 /a2a/{assistant_id}。
协议版本
Agent Server 使用 A2A v1.0 JSON-RPC 绑定，同时也接受 v0.3 的方法名，因此 现有的 v0.3 客户端仍可继续工作。agent card 声明一个接口：
json
"supportedInterfaces": [
{
"url": "https://your-deployment/a2a/{assistant_id}",
"protocolBinding": "JSONRPC",
"protocolVersion": "1.0"
}
]WARNING
你发送的方法名还会决定响应中的枚举大小写形式。v1.0 名称返回 SCREAMING_SNAKE_CASE（TASK_STATE_WORKING、ROLE_AGENT）；v0.3 名称返回小写 （working、agent）。每个客户端请选定一种风格并保持一致。
信封结构因方法而异，而非因风格而异：SendMessage 将任务包裹在 result.task 中，而 GetTask 和所有 v0.3 方法则直接在 result 上返回它。ListTasks 返回 result.tasks。
支持的方法
| v1.0 名称 | v0.3 名称 | 支持 |
|---|---|---|
SendMessage | message/send | 是 |
SendStreamingMessage | message/stream | 是 — Server-Sent Events |
GetTask | tasks/get | 是 |
CancelTask | tasks/cancel | 是 |
ListTasks | — | 是 |
GetExtendedAgentCard | — | 是，仅在 v1.0 名称下 |
SubscribeToTask | — | 尚未支持 — 返回 -32601 |
*TaskPushNotificationConfig | — | 尚未支持 — 返回 -32601 |
恰好接受四个 v0.3 名称：message/send、message/stream、tasks/get 和 tasks/cancel。其他任何名称——包括 agent/getAuthenticatedExtendedCard 和 tasks/resubscribe——都会返回 -32601 Method not found。
仅提供 JSON-RPC 绑定。gRPC 和 HTTP+JSON 尚未实现。
响应中的任务历史
一个上下文包含多个任务。默认情况下，SendMessage、GetTask 和 ListTasks 返回 整个上下文的历史，而不仅仅是你所询问的任务的历史。上下文中的第二个任务会重放 第一个任务的消息，因此渲染每一条历史记录的客户端会再次显示更早的轮次—— 包括更早的工具结果和 A2UI 载荷。
将 historyScope 设置为 task，即可仅返回属于你所询问的任务的消息。 默认值仍为 context，因此现有集成不受影响。
该选项放置的位置取决于方法。SendMessage 从 configuration 中读取它：
json
{
"jsonrpc": "2.0",
"id": "1",
"method": "SendMessage",
"params": {
"message": {
"role": "ROLE_USER",
"parts": [{"text": "Second request"}],
"messageId": "message-2",
"contextId": "8b1f0e5c-9a3d-4f27-b0c8-2e6a5d4c1b7a"
},
"configuration": {"historyScope": "task"}
}
}GetTask 和 ListTasks 直接从 params 中读取它：
json
{"jsonrpc": "2.0", "id": "2", "method": "GetTask",
"params": {"id": "<taskId>", "historyScope": "task"}}如果你的客户端无法向请求体添加字段，请改为发送请求头。请求中的显式值 优先于请求头。
LangGraph-A2A-History-Scope: taskagent card 会在 capabilities.extensions 下声明此能力，因此你可以检测是否支持， 而不必假设它受支持：
json
{
"uri": "https://langchain.com/a2a/extensions/history-scope/v1",
"description": "Choose task-only or full-context response history",
"required": false,
"params": {
"header": "LangGraph-A2A-History-Scope",
"values": ["context", "task"],
"default": "context",
"methods": ["SendMessage", "GetTask", "ListTasks"]
}
}三个值得了解的限制：
- 流式传输会忽略这两个历史选项。
SendStreamingMessage既不读取historyScope也不读取historyLength，并且如果你发送它们也不会返回错误——因此不要通过 SSE 依赖其中任何一个。 historyLength上限为 10。更大的值会返回-32602，错误信息为historyLength cannot exceed 10。- 作用范围在
historyLength之前应用，因此你得到的是该任务的最后 N 条消息。
无法识别的值会返回 -32602，错误信息为 historyScope must be 'context' or 'task'。大小写错误的 键（例如 historyscope）不是错误——它会被忽略，而你会静默地得到全上下文 历史，因此如果过滤似乎不起作用，请检查拼写。
WARNING
不要重新发送已完成任务的 taskId。每一新轮次都会在同一 上下文内启动一个新任务——请仅发送 contextId。指名某个终止状态任务的消息会被拒绝，返回 -32004， 而由其他智能体生成的 taskId 会被拒绝，返回 -32001。
Agent card 发现
每个 assistant 都会自动暴露一个 A2A Agent Card，用于描述其能力并提供其他智能体连接所需的信息。你可以使用以下方式获取任意 assistant 的 agent card：
GET /.well-known/agent-card.json?assistant_id={assistant_id}agent card 包含 assistant 的名称、描述、可用技能、支持的输入/输出模式，以及用于通信的 A2A 端点 URL。
可选能力
这些通过 Assistants API 上的 metadata.a2a 按 assistant 进行配置。langgraph.json 无法设置 assistant 元数据，因此在部署后对 assistant 进行修补。
声明输入和输出模式
json
{
"metadata": {
"a2a": {
"input_modes": ["text/plain", "application/pdf"],
"output_modes": ["text/plain", "application/pdf"]
}
}
}这些值会同时用于 card 的 defaultInputModes/defaultOutputModes 和生成的技能的 模式。它们仅用于声明——未声明的模式仍会被接受。替换是按 字段进行的，因此如果你希望两者都被覆盖，就要同时发送，并注意空列表会被拒绝。
文件部分
FilePart 支持双向工作。入站的文件、图像、音频和视频部分会变成 LangChain 内容块。出站的内容块会映射回 FilePart，无论是在任务历史中还是最终的 流式产物中。MIME 类型、URI 和文件名原样传递；内联数据会 重新编码为标准 base64。
A2UI v0.9
按 assistant 选择性启用：
json
{ "metadata": { "a2a": { "a2ui": true } } }随后 card 会声明该扩展，并将规范的 MIME 类型追加到两个模式列表中：
json
"capabilities": {
"extensions": [
{
"uri": "https://a2ui.org/a2a-extension/a2ui/v0.9",
"description": "Ability to render A2UI v0.9",
"required": false,
"params": {
"v0.9": {
"supportedCatalogIds": [],
"acceptsInlineCatalogs": false
}
}
}
]
}客户端可以通过在 message.extensions 中列出该 URI 来激活它。载荷会在两个 方向上依据 v0.9 schema 进行验证，且 A2UI 部分会在 message/send、tasks/get 和最终的 message/stream 产物中得到保留。响应会带有 metadata.mimeType: "application/a2ui+json"。application/json+a2ui 在 输入时作为别名被接受。
过滤工具结果
默认情况下，每个相关的工具结果都会作为 DataPart 发布。若只想发布部分结果，请在部署上设置 工具名称的允许列表：
bash
A2A_ALLOWED_TOOL_CALL_RESULTS=generative_ui_tool,another_tool未设置表示发布所有工具结果。该过滤器同时适用于任务历史和流式传输。
要求
| 功能 | 最低版本 |
|---|---|
| A2A 端点 | langgraph-api >= 0.4.21 |
入站 FilePart | 0.12.0 |
工具结果 DataPart | 0.12.2 |
A2A_ALLOWED_TOOL_CALL_RESULTS | 0.12.4 |
出站 FilePart、可配置的 card 模式 | 0.13.0 |
| A2UI v0.9 | 0.15.0 |
historyScope | 0.15.0 |
bash
pip install "langgraph-api>=0.13.0"A2UI v0.9 和 historyScope 是在 0.14.0 候选发布版切出之后才落地的，因此它们 出现在 0.15.0 中。该版本尚未作为稳定版本发布——在依赖 historyScope 之前， 请检查 agent card 上的 capabilities.extensions。
你的图的状态必须包含 messages 键，才能接受 A2A 文本和文件部分。如果某个 assistant 的输入 schema 没有 messages 字段，它会被拒绝并返回一条解释性错误。
创建一个 A2A 兼容的智能体
此示例创建一个 A2A 兼容的智能体，它使用 OpenAI 的 API 处理传入消息并维护对话状态。该智能体定义基于消息的状态结构，并处理 A2A 协议的消息格式。
为了与 A2A "text" 部分兼容，智能体的状态中必须有一个 messages 键。
A2A 协议使用两个标识符来维持对话的连续性：
contextId：将消息分组到一个对话线程中（类似于会话 ID）taskId：标识该对话中的每个单独请求
在第一条消息中，省略两者——智能体会生成并返回它们。对于对话中所有后续消息，请回传先前响应中的 contextId 并省略 taskId，这样每一轮都会在同一对话中开启一个新任务。仅当要向仍在运行的任务（例如等待输入的任务）追加内容时，才发送 taskId。
LangSmith 追踪： Langsmith Deployment 的 A2A 端点会自动将 A2A contextId 转换为 thread_id 以进行 LangSmith 追踪，从而将对话中的所有消息归组到单个线程下。
例如：
python
"""LangGraph A2A conversational agent.
Supports the A2A protocol with messages input for conversational interactions.
"""
from __future__ import annotations
import os
from dataclasses import dataclass
from typing import Any, Dict, List, TypedDict
from langgraph.graph import StateGraph
from langgraph.runtime import Runtime
from openai import AsyncOpenAI
class Context(TypedDict):
"""Context parameters for the agent."""
my_configurable_param: str
@dataclass
class State:
"""Input state for the agent.
Defines the initial structure for A2A conversational messages.
"""
messages: List[Dict[str, Any]]
async def call_model(state: State, runtime: Runtime[Context]) -> Dict[str, Any]:
"""Process conversational messages and returns output using OpenAI."""
# Initialize OpenAI client
client = AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY"))
# Process the incoming messages
latest_message = state.messages[-1] if state.messages else {}
user_content = latest_message.get("content", "No message content")
# Create messages for OpenAI API
openai_messages = [
{
"role": "system",
"content": "You are a helpful conversational agent. Keep responses brief and engaging."
},
{
"role": "user",
"content": user_content
}
]
try:
# Make OpenAI API call
response = await client.chat.completions.create(
model="gpt-3.5-turbo",
messages=openai_messages,
max_tokens=100,
temperature=0.7
)
ai_response = response.choices[0].message.content
except Exception as e:
ai_response = f"I received your message but had trouble processing it. Error: {str(e)[:50]}..."
# Create a response message
response_message = {
"role": "assistant",
"content": ai_response
}
return {
"messages": state.messages + [response_message]
}
# Define the graph
graph = (
StateGraph(State, context_schema=Context)
.add_node(call_model)
.add_edge("__start__", "call_model")
.compile()
)智能体间通信
一旦你的智能体通过 langgraph dev 在本地运行，或部署到生产环境，你就可以使用 A2A 协议促进它们之间的通信。
此示例演示两个智能体如何通过向彼此的 A2A 端点发送 JSON-RPC 消息进行通信。该脚本模拟一段多轮对话，其中每个智能体处理对方的响应并继续对话。
python
#!/usr/bin/env python3
"""Agent-to-Agent conversation simulation using the LangGraph A2A endpoint."""
import asyncio
import aiohttp
import os
import uuid
def extract_text(result: dict) -> str:
"""Best-effort extraction of response text from an A2A result."""
if "error" in result:
raise RuntimeError(f"A2A error {result['error']['code']}: {result['error']['message']}")
for art in result.get("result", {}).get("artifacts", []) or []:
for part in art.get("parts", []) or []:
if part.get("kind") == "text" and part.get("text"):
return part["text"]
msg = (result.get("result", {}).get("status", {}) or {}).get("message", {}) or {}
for part in msg.get("parts", []) or []:
if part.get("kind") == "text" and part.get("text"):
return part["text"]
return "(no text found)"
async def send_message(session, port, assistant_id, text, context_id=None):
"""Send an A2A message. Returns (response_text, returned_context_id)."""
url = f"http://127.0.0.1:{port}/a2a/{assistant_id}"
message = {
"role": "user",
"parts": [{"kind": "text", "text": text}],
"messageId": str(uuid.uuid4()),
}
# A2A multi-turn continuity: reuse contextId across turns and agents.
# Do not reuse taskId — each turn starts a new task within the same context.
if context_id:
message["contextId"] = context_id
payload = {
"jsonrpc": "2.0",
"id": str(uuid.uuid4()),
"method": "message/send",
"params": {"message": message},
}
headers = {"Accept": "application/json"}
async with session.post(url, json=payload, headers=headers) as response:
result = await response.json()
text = extract_text(result)
returned_context_id = result.get("result", {}).get("contextId") or context_id
return text, returned_context_id
async def simulate_conversation():
"""Simulate a conversation between two agents."""
#Assistant IDs
agent_a_id = os.getenv("AGENT_A_ID")
agent_b_id = os.getenv("AGENT_B_ID")
if not agent_a_id or not agent_b_id:
print("Set AGENT_A_ID and AGENT_B_ID environment variables")
return
message = "Hello! Let's have a conversation."
context_id = None
async with aiohttp.ClientSession() as session:
for i in range(3):
print(f"--- Round {i + 1} ---")
message, context_id = await send_message(
session, 2024, agent_a_id, message, context_id=context_id
)
print(f"🔵 Agent A: {message}")
message, context_id = await send_message(
session, 2025, agent_b_id, message, context_id=context_id
)
print(f"🔴 Agent B: {message}\n")
if __name__ == "__main__":
asyncio.run(simulate_conversation())完整可运行的示例，请参见：
- 两个 LangGraph 智能体通信 - 两个 LangGraph 智能体使用 A2A 协议的示例
- Google ADK 智能体与 LangChain 智能体 - Google ADK 智能体使用 A2A 协议与 LangChain 智能体交互的示例
分布式追踪
当多个智能体通过 A2A 通信时，LangSmith 可以将它们所有的追踪归组到单个线程中，从而让你获得整个多智能体对话的统一视图。
contextId 如何映射到 thread_id
Agent Server 的 A2A 端点会自动将 A2A contextId 转换为 thread_id 以进行 LangSmith 追踪。这意味着对话中的每条消息，无论来自哪个参与的智能体，都会在 LangSmith 中归组到同一个线程下，无需你做任何额外配置。
流程如下：
- 在第一条消息中，客户端省略
contextId。服务器会生成一个并在响应中返回它。 - 客户端在所有后续消息中传递该
contextId，以维持对话的连续性。 - Agent Server 会在 LangSmith 元数据中将
contextId映射到thread_id，因此所有轮次都会出现在同一个线程中。
WARNING
contextId 会直接用作 LangGraph 的 thread_id，因此它必须是 UUID。请回传 服务器返回的那个，而不是自行生成标识符。诸如 session-42 这样的 contextId 会被拒绝，返回 -32602，并附带消息 Failed to create run: Invalid thread ID。
跨多个智能体追踪
当来自不同框架的智能体通过 A2A 通信时，contextId 是统一它们追踪的关键。在之后对第一个智能体以及对他人的每次请求中，请复用第一个智能体返回的 contextId。
WARNING
Agent Server 不会读取 JSON-RPC 载荷上的顶层 metadata 字段。客户端无法直接设置 LangGraph 的 thread_id——它始终是 contextId。向 Agent Server 部署发送 metadata.thread_id 不会产生任何效果。
以下代码片段演示了关键概念。有关包含两个智能体的完整可运行实现，请参阅 Google ADK + LangChain 示例。
python
import asyncio
import aiohttp
import uuid
async def send_message(session, url, text, context_id=None):
"""Send an A2A message and return (response_text, context_id)."""
# --- 1. Build the message ---
# On follow-up turns, include contextId inside the message object so the server
# associates them with the ongoing conversation. Do not resend taskId: each turn
# opens a new task within that conversation.
message = {
"role": "user",
"parts": [{"kind": "text", "text": text}],
"messageId": str(uuid.uuid4()),
}
if context_id:
message["contextId"] = context_id
# --- 2. Send it ---
# contextId travels inside the message. Agent Server turns it into the
# LangGraph thread_id, so no separate tracing field is needed.
payload = {
"jsonrpc": "2.0",
"id": str(uuid.uuid4()),
"method": "message/send",
"params": {"message": message},
}
async with session.post(url, json=payload, headers={"Accept": "application/json"}) as response:
if response.status != 200:
raise RuntimeError(f"HTTP {response.status}: {await response.text()}")
result = await response.json()
if "error" in result:
raise RuntimeError(result["error"].get("message", "Unknown error"))
result_obj = result.get("result", {})
returned_context_id = result_obj.get("contextId") or context_id
text_out = next(
(
part.get("text", "")
for art in result_obj.get("artifacts", []) or []
for part in art.get("parts", []) or []
if part.get("kind") == "text"
),
"(no text)",
)
return text_out, returned_context_id
async def run_conversation(agent_a_url, agent_b_url):
# --- 3. Share the context across agents ---
# The first response carries a contextId. Pass it to every agent from then
# on, and all their traces land in one LangSmith thread.
context_id = None
message = "Hello! Let's collaborate."
async with aiohttp.ClientSession() as session:
for _ in range(3):
message, context_id = await send_message(
session, agent_a_url, message, context_id=context_id
)
message, context_id = await send_message(
session, agent_b_url, message, context_id=context_id
)
asyncio.run(run_conversation(
"http://localhost:2024/a2a/<agent_a_assistant_id>",
"http://localhost:2025/a2a/<agent_b_assistant_id>",
))1. 构建消息：在后续轮次中，将 contextId 包含在 message 对象内，以便服务器将它们与正在进行的对话关联起来。在第一条消息中省略它，因为服务器会生成 contextId 并在响应中返回它。不要重新发送已结束轮次的 taskId。
2. 发送它：contextId 在 params.message 内传递。Agent Server 将其用作 LangGraph 的 thread_id，因此没有单独的追踪字段需要设置。
3. 在多个智能体间共享上下文：让第一个智能体生成 contextId，然后在对话的剩余部分将该相同的值传递给每个智能体。这正是将它们追踪归组到同一线程的方式。
在非 LangGraph 智能体中接收 thread_id
上一节介绍了客户端一侧——在发送消息时传播 contextId。如果你的某个智能体不是基于 LangGraph 构建的，它还需要在接收端读取该 contextId 并将其附加为线程标识符，以便它的追踪落入同一个 LangSmith 线程。使用 langsmith.integrations.otel.configure() 设置自动追踪，并从传入的 A2A 请求中读取 params.message.contextId。
python
from fastapi import FastAPI, Request
from langsmith.integrations.otel import configure as configure_otel
from opentelemetry import trace
import json
# --- 1. Configure OTel ---
# Set up automatic tracing to LangSmith for your non-LangGraph agent.
configure_otel(project_name="my-a2a-project")
tracer = trace.get_tracer(__name__)
app = FastAPI()
@app.middleware("http")
async def set_thread_id_middleware(request: Request, call_next):
thread_id = None
if request.method == "POST":
body_bytes = await request.body()
if body_bytes:
# --- 2. Extract contextId from the incoming A2A message ---
try:
body = json.loads(body_bytes)
thread_id = body["params"]["message"].get("contextId")
except (ValueError, KeyError, TypeError):
pass
# Re-inject the body so downstream handlers can still read it
async def receive():
return {"type": "http.request", "body": body_bytes}
request._receive = receive
# --- 3. Attach thread_id to the trace ---
# langsmith.metadata.thread_id groups this trace with others in the same thread.
with tracer.start_as_current_span("agent") as span:
if thread_id:
span.set_attribute("langsmith.metadata.thread_id", thread_id)
return await call_next(request)在此中间件之后，将你的智能体路由注册到 app 上。
NOTE
在你的环境中设置 LANGSMITH_API_KEY，并可选地设置 LANGSMITH_PROJECT，以启用追踪。对话中的所有智能体都应使用同一个项目，以便它们的追踪可以一起查看。
在 LangSmith 中查看追踪
运行多智能体对话后，打开 LangSmith UI 并导航到 Threads。来自所有参与智能体的所有轮次都会出现在单个线程下，由共享的 thread_id 标识。
测试你的集成
针对你自己的部署
获取 card，然后发送一条消息：
bash
curl "https://your-deployment/a2a/{assistant_id}/.well-known/agent-card.json"bash
curl -X POST "https://your-deployment/a2a/{assistant_id}" \
-H "Content-Type: application/json" \
-H "Accept: application/json" \
-d '{
"jsonrpc": "2.0",
"id": 1,
"method": "SendMessage",
"params": {
"message": {
"role": "ROLE_USER",
"parts": [{"text": "hello"}],
"messageId": "test-1"
}
}
}'响应包含 result.task.id 和 result.task.contextId。在下一条消息中复用该 contextId， 以继续对话。
对于流式传输，发送 Accept: text/event-stream 并使用 SendStreamingMessage。第一个事件是 Task；随后是状态和产物的更新。
针对官方一致性套件
A2A 在 a2aproject/a2a-tck 发布了技术兼容性套件（TCK）。它按 RFC 2119 级别对实现进行评级，并且适用于任何 A2A 端点，包括你的端点。
bash
./run_tck.py --sut-host https://your-deployment/a2a/{assistant_id} --transport jsonrpcNOTE
TCK 通过 messageId 前缀（例如 tck-input-required）驱动某些场景，这些前缀在其 docs/SUT_REQUIREMENTS.md 中有描述。未实现这些前缀的图会将这些 要求报告为跳过，而不是失败。
Agent Server 目前未通过的项目
Agent Server 在每次 CI 构建时都将 TCK 作为必需检查运行，并以一份已提交的 已知失败清单作为门槛。如果出现新的失败，CI 会失败；如果清单中的某项要求开始通过， CI 也会失败，因此该清单不会偏离服务器的实际行为。
在基于某项能力进行构建之前，请先阅读以下内容：
| 差距 | 你观察到的现象 |
|---|---|
| 响应线格式仍为 v0.3 | 任务、消息和部分携带 kind 和 mimeType，而非 v1.0 的成员存在性判别 |
| 流式事件是扁平的 | SSE 发出带有 final 的 v0.3 对象，而不是 statusUpdate / artifactUpdate 包装 |
tool_results 是 snake_case | v1.0 期望 toolResults。这是有意保留的，因为在用的 A2UI 客户端会读取此键 |
| 时间戳 | 序列化为 +00:00，而非 ISO 8601 的 Z 后缀 |
SubscribeToTask | 返回 -32601，而规范要求 -32001 |
| 推送通知配置 | 返回 -32601，而规范要求 -32003 |
错误不携带 data | 未附加 google.rpc.ErrorInfo 的 reason 或 domain |
A2A-Version 请求头 | 未被读取，因此会处理不受支持的版本，而不是返回 -32009 |
| Agent card 缓存 | 没有 Cache-Control、ETag 或 Last-Modified 响应头 |
GetExtendedAgentCard | 有提供，但从未通过 capabilities.extendedAgentCard 声明 |
禁用 A2A
要禁用 A2A 端点，请在你的 langgraph.json 配置文件中将 disable_a2a 设置为 true：
json
{
"$schema": "https://langgra.ph/schema.json",
"http": {
"disable_a2a": true
}
}