代理到代理（A2A）协议支持代理之间的标准化通信，允许使用不同框架和技术构建的代理无缝通信。
什么是 A2A？
A2A 是一个标准化协议，支持：
- 通过代理卡进行代理发现
- 代理之间的基于消息的通信
- 通过任务长时间运行的代理进程
- 不同代理框架之间的跨平台互作性
有关详细信息，请参阅 A2A 协议规范。
该 Microsoft.Agents.AI.Hosting.A2A.AspNetCore 库提供 ASP.NET Core 集成，用于通过 A2A 协议公开代理。
NuGet 包：
Example
此最小示例演示如何通过 A2A 暴露代理。 此示例包括 OpenAPI 和 Swagger 依赖项，以简化测试。
1.创建 ASP.NET 核心 Web API 项目
创建新的 ASP.NET Core Web API 项目或使用现有项目。
2.安装所需的依赖项
安装以下包：
在项目目录中运行以下命令以安装所需的 NuGet 包：
# Hosting.A2A.AspNetCore for A2A protocol integration
dotnet add package Microsoft.Agents.AI.Hosting.A2A.AspNetCore --prerelease
# Libraries to connect to Azure OpenAI
dotnet add package Azure.AI.OpenAI --prerelease
dotnet add package Azure.Identity
dotnet add package Microsoft.Extensions.AI
dotnet add package Microsoft.Extensions.AI.OpenAI --prerelease
# Swagger to test app
dotnet add package Microsoft.AspNetCore.OpenApi
dotnet add package Swashbuckle.AspNetCore
3.配置 Azure OpenAI 连接
应用程序需要 Azure OpenAI 连接。 使用 dotnet user-secrets 或环境变量配置终结点和部署名称。
您还可以简单地编辑appsettings.json，但不建议对生产环境中部署的应用程序这样操作，因为某些数据可以被视为机密。
dotnet user-secrets set "AZURE_OPENAI_ENDPOINT" "https://<your-openai-resource>.openai.azure.com/"
dotnet user-secrets set "AZURE_OPENAI_DEPLOYMENT_NAME" "gpt-4o-mini"
4.将代码添加到Program.cs
将 Program.cs 的内容替换为以下代码，然后运行应用程序：
using A2A.AspNetCore;
using Azure.AI.OpenAI;
using Azure.Identity;
using Microsoft.Agents.AI.Hosting;
using Microsoft.Extensions.AI;
var builder = WebApplication.CreateBuilder(args);
builder.Services.AddOpenApi();
builder.Services.AddSwaggerGen();
string endpoint = builder.Configuration["AZURE_OPENAI_ENDPOINT"]
?? throw new InvalidOperationException("AZURE_OPENAI_ENDPOINT is not set.");
string deploymentName = builder.Configuration["AZURE_OPENAI_DEPLOYMENT_NAME"]
?? throw new InvalidOperationException("AZURE_OPENAI_DEPLOYMENT_NAME is not set.");
// Register the chat client
IChatClient chatClient = new AzureOpenAIClient(
new Uri(endpoint),
new DefaultAzureCredential())
.GetChatClient(deploymentName)
.AsIChatClient();
builder.Services.AddSingleton(chatClient);
// Register an agent
var pirateAgent = builder.AddAIAgent("pirate", instructions: "You are a pirate. Speak like a pirate.");
var app = builder.Build();
app.MapOpenApi();
app.UseSwagger();
app.UseSwaggerUI();
// Expose the agent via A2A protocol. You can also customize the agentCard
app.MapA2A(pirateAgent, path: "/a2a/pirate", agentCard: new()
{
Name = "Pirate Agent",
Description = "An agent that speaks like a pirate.",
Version = "1.0"
});
app.Run();
测试代理
应用程序运行后，可以使用以下 .http 文件或通过 Swagger UI 测试 A2A 代理。
输入格式符合 A2A 规范。 可以为以下项提供值：
-
messageId- 此特定消息的唯一标识符。 可以创建自己的 ID（例如 GUID），也可以将其null设置为允许代理自动生成一个 ID。 -
contextId- 会话标识符。 提供您自己的 ID 以启动新对话，或者通过重用之前的contextId来继续现有对话。 代理将保留同一contextId会话历史记录。 如果未提供，代理将为你生成一个。
# Send A2A request to the pirate agent
POST {{baseAddress}}/a2a/pirate/v1/message:stream
Content-Type: application/json
{
"message": {
"kind": "message",
"role": "user",
"parts": [
{
"kind": "text",
"text": "Hey pirate! Tell me where have you been",
"metadata": {}
}
],
"messageId": null,
"contextId": "foo"
}
}
注意：将 {{baseAddress}} 替换为您的服务器终结点。
此请求返回以下 JSON 响应：
{
"kind": "message",
"role": "agent",
"parts": [
{
"kind": "text",
"text": "Arrr, ye scallywag! Ye’ll have to tell me what yer after, or be I walkin’ the plank? 🏴☠️"
}
],
"messageId": "chatcmpl-CXtJbisgIJCg36Z44U16etngjAKRk",
"contextId": "foo"
}
响应包括 contextId （对话标识符）、 messageId （消息标识符）和海盗代理的实际内容。
AgentCard 配置
`AgentCard 提供关于您的代理的元数据，以便于发现和集成。`
app.MapA2A(agent, "/a2a/my-agent", agentCard: new()
{
Name = "My Agent",
Description = "A helpful agent that assists with tasks.",
Version = "1.0",
});
可以通过发送此请求来访问代理卡：
# Send A2A request to the pirate agent
GET {{baseAddress}}/a2a/pirate/v1/card
注意：将 {{baseAddress}} 替换为您的服务器终结点。
AgentCard 属性
- 名称：代理的显示名称
- 说明：代理的简要说明
- 版本：代理的版本字符串
- URL：终结点 URL（如果未指定，将自动分配）
- 功能：有关流式处理、推送通知和其他功能的可选元数据
暴露多个代理
只要单个应用程序中的终结点不冲突，就可以在单个应用程序中公开多个代理。 下面是一个示例：
var mathAgent = builder.AddAIAgent("math", instructions: "You are a math expert.");
var scienceAgent = builder.AddAIAgent("science", instructions: "You are a science expert.");
app.MapA2A(mathAgent, "/a2a/math");
app.MapA2A(scienceAgent, "/a2a/science");
通过此 agent-framework-a2a 包，可以连接到外部符合 A2A 的代理并与之通信。
pip install agent-framework-a2a --pre
连接到 A2A 代理
使用 A2AAgent 包装任何远程 A2A 终结点。 代理通过其 AgentCard 解析远程代理的功能，并处理所有协议详细信息。
import asyncio
import httpx
from a2a.client import A2ACardResolver
from agent_framework.a2a import A2AAgent
async def main():
a2a_host = "https://your-a2a-agent.example.com"
# 1. Discover the remote agent's capabilities
async with httpx.AsyncClient(timeout=60.0) as http_client:
resolver = A2ACardResolver(httpx_client=http_client, base_url=a2a_host)
agent_card = await resolver.get_agent_card()
print(f"Found agent: {agent_card.name}")
# 2. Create an A2AAgent and send a message
async with A2AAgent(
name=agent_card.name,
agent_card=agent_card,
url=a2a_host,
) as agent:
response = await agent.run("What are your capabilities?")
for message in response.messages:
print(message.text)
asyncio.run(main())
流式处理响应
A2A 自然支持通过 Server-Sent 事件来实现流式传输，更新在远程代理工作时实时到达：
async with A2AAgent(name="remote", url="https://a2a-agent.example.com") as agent:
async with agent.run("Tell me about yourself", stream=True) as stream:
async for update in stream:
for content in update.contents:
if content.text:
print(content.text, end="", flush=True)
final = await stream.get_final_response()
print(f"\n({len(final.messages)} message(s))")
长时间运行任务
默认情况下， A2AAgent 等待远程代理在返回之前完成。 对于长时间运行的任务，请配置 background=True 来获取可用于稍后轮询或重新订阅的延续令牌：
async with A2AAgent(name="worker", url="https://a2a-agent.example.com") as agent:
# Start a long-running task
response = await agent.run("Process this large dataset", background=True)
if response.continuation_token:
# Poll for completion later
result = await agent.poll_task(response.continuation_token)
print(result)
Authentication
请使用AuthInterceptor来保护 A2A 终结点：
from a2a.client.auth.interceptor import AuthInterceptor
class BearerAuth(AuthInterceptor):
def __init__(self, token: str):
self.token = token
async def intercept(self, request):
request.headers["Authorization"] = f"Bearer {self.token}"
return request
async with A2AAgent(
name="secure-agent",
url="https://secure-a2a-agent.example.com",
auth_interceptor=BearerAuth("your-token"),
) as agent:
response = await agent.run("Hello!")