智能体间通信 (A2A) 是 Google 推出的用于实现对话式 AI 智能体之间通信的协议。 LangSmith 实现了 A2A 支持，允许您的智能体通过标准化协议与其他 A2A 兼容的智能体进行通信。
A2A 端点在以下位置可用 Agent Server at /a2a/{assistant_id}.
支持的方法
Agent Server 支持以下 A2A RPC 方法：
- message/send：向助手发送消息并接收完整响应
- message/stream：使用服务器发送事件 (SSE) 发送消息并实时流式传输响应
- tasks/get：获取之前创建任务的状态和结果
智能体卡片发现
每个助手自动暴露一个 A2A 智能体卡片，用于描述其功能并提供其他智能体连接所需的信息。您可以使用以下方式获取任何助手的智能体卡片：
GET /.well-known/agent-card.json?assistant_id={assistant_id}The agent card includes the assistant's name, description, available skills, supported input/output modes, and the A2A endpoint URL for communication.
需求
要使用 A2A，请确保已安装以下依赖项：
- *
langgraph-api >= 0.4.21
安装方式：
pip install "langgraph-api>=0.4.21"用法概述
启用 A2A：
- * 升级到使用 langgraph-api>=0.4.21。
- * 使用基于消息的状态结构部署您的智能体。
- * 使用该端点与其他 A2A 兼容的智能体建立连接。
创建 A2A 兼容的智能体
此示例创建一个 A2A 兼容的智能体，使用 OpenAI 的 API 处理传入消息并维护对话状态。智能体定义了基于消息的状态结构并处理 A2A 协议的消息格式。
要兼容 A2A "文本"部分，智能体必须在状态中包含 messages 键。
A2A 协议使用两个标识符来维持对话连续性： * contextId：将消息分组到对话线程中（类似于会话 ID） * taskId：标识该对话中的每个单独请求
在第一条消息中，省略 contextId 和 taskId - 智能体会生成并返回它们。对于对话中的所有后续消息，请包含 contextId 和 taskId 从之前的响应中获取以维持线程连续性。
LangSmith 追踪： LangSmith 部署的 A2A 端点自动转换 A2A contextId to thread_id 用于 LangSmith 追踪，将对话中的所有消息分组到单个线程下。
例如：
"""LangGraph A2A conversational agent.
Supports the A2A protocol with messages input for conversational interactions.
"""
from __future__ import annotations
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 or 在本地运行或部署到生产环境，您可以使用 A2A 协议促进它们之间的通信。
此示例演示了两个代理如何通过向彼此的 A2A 端点发送 JSON-RPC 消息进行通信。该脚本模拟了一个多轮对话，其中每个代理处理对方的响应并继续对话。
#!/usr/bin/env python3
"""Agent-to-Agent conversation simulation using the LangGraph A2A endpoint."""
def extract_text(result: dict) -> str:
"""Best-effort extraction of response text from an A2A result."""
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, task_id=None):
"""Send an A2A message. Returns (response_text, returned_context_id, returned_task_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 and taskId across turns/agents
if context_id:
message["contextId"] = context_id
if task_id:
message["taskId"] = task_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()
returned_context_id = result.get("result", {}).get("contextId") or context_id
returned_task_id = result.get("result", {}).get("id")
return extract_text(result), returned_context_id, returned_task_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
task_id = None
async with aiohttp.ClientSession() as session:
for i in range(3):
print(f"--- Round {i + 1} ---")
message, context_id, task_id = await send_message(
session, 2024, agent_a_id, message,
context_id=context_id,
task_id=task_id,
)
print(f"🔵 Agent A: {message}")
message, context_id, task_id = await send_message(
session, 2025, agent_b_id, message,
context_id=context_id,
task_id=task_id,
)
print(f"🔴 Agent B: {message}\n")
if __name__ == "__main__":
asyncio.run(simulate_conversation())完整的工作示例请参见： - 两个 LangGraph 代理通信 - 两个使用 A2A 协议的 LangGraph 代理示例 - Google ADK 代理与 LangChain 代理 - Google ADK 代理与使用 A2A 协议的 LangChain 代理交互示例
分布式追踪
当多个代理通过 A2A 通信时，LangSmith 可以将它们的 追踪记录 分组到单个 线程中，这使您可以统一查看整个多代理对话。
contextId 如何映射到线程_id
代理服务器 A2A 端点自动将 A2A contextId to thread_id 用于 LangSmith 追踪。这意味着对话中的每条消息（跨所有参与的代理）都被分组到 LangSmith 的同一线程中，无需您进行任何额外配置。
流程如下：
- 在第一条消息时，客户端省略
contextId。服务器生成一个并在响应中返回。 - 客户端在所有后续消息中传递
contextId以维持对话连续性。 - 代理服务器将
contextIdtothread_id映射到 LangSmith 元数据中，因此所有轮次都出现在同一线程中。
跨多个代理的追踪
当来自不同框架的代理通过 A2A 通信时，您可以通过在所有代理之间共享相同的 thread_id 来统一它们在 LangSmith 中的追踪。使用 contextId 第一个代理返回的 thread_id 作为所有后续请求的
以下代码片段演示了关键概念。对于两个代理的完整可运行实现，请参阅 Google ADK + LangChain 示例.
async def send_message(session, url, text, context_id=None, task_id=None, thread_id=None):
"""Send an A2A message and return (response_text, context_id, task_id)."""
# --- 1. Build the message ---
# On follow-up turns, include contextId and taskId inside the message object
# so the server associates them with the ongoing conversation.
message = {
"role": "user",
"parts": [{"kind": "text", "text": text}],
"messageId": str(uuid.uuid4()),
}
if context_id:
message["contextId"] = context_id
if task_id:
message["taskId"] = task_id
# --- 2. Set thread_id in metadata ---
# thread_id goes at the top level of the JSON-RPC payload, not inside params.
payload = {
"jsonrpc": "2.0",
"id": str(uuid.uuid4()),
"method": "message/send",
"params": {"message": message},
"metadata": {"thread_id": thread_id},
}
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
returned_task_id = result_obj.get("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, returned_task_id
async def run_conversation(agent_a_url, agent_b_url):
# --- 3. Share thread_id across agents ---
# Generate a shared thread_id upfront. Once the server returns a contextId,
# use that instead — this keeps the A2A context and LangSmith thread in sync.
thread_id = str(uuid.uuid4())
context_id = None
task_id = None
message = "Hello! Let's collaborate."
async with aiohttp.ClientSession() as session:
for _ in range(3):
message, context_id, task_id = await send_message(
session, agent_a_url, message,
context_id=context_id, task_id=task_id,
thread_id=context_id or thread_id,
)
# Passing the same thread_id to every agent groups all traces in LangSmith
message, context_id, task_id = await send_message(
session, agent_b_url, message,
context_id=context_id, task_id=task_id,
thread_id=context_id or thread_id,
)
asyncio.run(run_conversation(
"http://localhost:2024/a2a/<agent_a_assistant_id>",
"http://localhost:2025/a2a/<agent_b_assistant_id>",
))1. 构建消息：在后续轮次的消息中包含 contextId 和 taskId 在 message 对象中，以便服务器可以将它们与正在进行的对话关联起来。在第一条消息时省略它们，因为服务器会生成一个 contextId 并在响应中返回。
2. 在元数据中设置线程_id：在 JSON-RPC 负载的顶层 thread_id 字段中传递 metadata ，而不是放在 params.
3. 在代理之间共享线程_id：在第一条消息之前生成一个随机 thread_id 。一旦服务器返回一个 contextId，就使用它作为 thread_id 用于所有后续请求，使 A2A 对话上下文和 LangSmith 线程保持同步。传递相同的 thread_id 给每个代理，以便所有追踪被分组到一个线程中。
在非 LangGraph 代理中接收线程_ID
前 上一节 涵盖了客户端——发送消息时传播 thread_id 。如果您的一个代理不是基于 LangGraph 构建的，它还需要在接收端提取并附加 thread_id ，以便其追踪落在同一个 LangSmith 线程中。使用 langsmith.integrations.otel.configure() 设置自动追踪，并从 thread_id 中提取传入的 A2A 请求元数据，以将追踪分组到同一线程中。
from fastapi import FastAPI, Request
from langsmith.integrations.otel import configure as configure_otel
from opentelemetry import trace
# --- 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 thread_id from incoming A2A metadata ---
try:
body = json.loads(body_bytes)
thread_id = body.get("metadata", {}).get("thread_id")
except Exception:
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 。
在 LangSmith 中查看追踪
运行多代理对话后，打开 LangSmith UI 并导航到 **线程**。所有参与代理的所有轮次将显示在单个线程下，由共享的 thread_id.
禁用 A2A
要禁用 A2A 端点，请在 disable_a2a to true 中设置 langgraph.json ：
{
"$schema": "https://langgra.ph/schema.json",
"http": {
"disable_a2a": true
}
}