构建符合 A2A 标准的客户端和服务器

包安装

1# 创建一个干净的工作区
2mkdir a2a-echo-example
3cd a2a-echo-example
4
5# 设置虚拟环境
6python3 -m venv venv
7source venv/bin/activate  # 在 Windows 上：venv\Scripts\activate
8
9# 安装核心依赖项
10pip install fastapi uvicorn pydantic requests

1"""
2简单的 A2A 服务器示例 - 第一步：基础 Web 服务器
3"""
4from fastapi import FastAPI
5
6app = FastAPI()
7
8@app.get("/")
9def hello():
10    return {"message": "来自我们未来 A2A 代理的问候！"}
11
12if __name__ == "__main__":
13    import uvicorn
14    print("🚀 在 http://localhost:8000 启动基础 Web 服务器")
15    uvicorn.run(app, host="0.0.0.0", port=8000)

代码示例

1{"message": "来自我们未来 A2A 代理的问候！"}

1# 新增：添加此端点用于 A2A 发现
2@app.get("/.well-known/agent-card.json")
3def agent_card():
4    """A2A 代理发现端点"""
5    return {
6        "name": "回声代理",
7        "description": "一个简单的代理，会将您的消息原样返回",
8        "url": "http://localhost:8000",
9        "version": "1.0.0",
10        "protocolVersion": "0.3.0",
11        "skills": [{
12            "id": "echo",
13            "name": "回声消息",
14            "description": "重复您说的任何内容",
15            "examples": ["你好", "你怎么样？"]
16        }],
17        "capabilities": {"streaming": False},
18        "defaultInputModes": ["text/plain"],
19        "defaultOutputModes": ["text/plain"]
20    }
21
22# 更新：修改 hello 消息
23@app.get("/")
24def hello():
25    return {"message": "来自我们的 A2A 代理的问候！"}

1from pydantic import BaseModel
2from typing import List
3import uuid
4from datetime import datetime
5
6class Part(BaseModel):
7    kind: str
8    text: str = None
9
10class Message(BaseModel):
11    kind: str = "message"
12    messageId: str
13    role: str
14    parts: List[Part]
15
16class MessageParams(BaseModel):
17    message: Message
18
19class JSONRPCRequest(BaseModel):
20    jsonrpc: str = "2.0"
21    id: str
22    method: str
23    params: MessageParams

1@app.post("/")
2def handle_message(request: JSONRPCRequest):
3    """A2A 消息处理器 - 即将推出！"""
4    return {"jsonrpc": "2.0", "id": request.id, "result": "处理器即将上线！"}

代理的 POST 处理器

1@app.post("/")
2def handle_message(request: JSONRPCRequest):
3    """A2A 消息处理器"""
4    # 验证方法
5    if request.method != "message/send":
6        return {
7            "jsonrpc": "2.0", 
8            "id": request.id, 
9            "error": {
10                "code": -32601, 
11                "message": "方法未找到"
12            }
13        }
14    
15    # 提取用户消息
16    user_message = request.params.message
17    user_text = user_message.parts[0].text if user_message.parts else "无文本"
18    
19    # 创建我们的 A2A 响应作为已完成的任务
20    task = {
21        "kind": "task",
22        "id": str(uuid.uuid4()),
23        "contextId": str(uuid.uuid4()),
24        "status": {
25            "state": "completed", 
26            "timestamp": datetime.utcnow().isoformat() + "Z"
27        },
28        "history": [
29            # 包含原始用户消息
30            user_message.dict(),
31            # 添加我们的代理回复
32            {
33                "kind": "message",
34                "messageId": str(uuid.uuid4()),
35                "role": "agent",
36                "parts": [{"kind": "text", "text": f"您说：'{user_text}'"}]
37            }
38        ]
39    }
40    
41    return {"jsonrpc": "2.0", "id": request.id, "result": task}

1"""
2简单的 A2A 服务器示例 - 完整实现
3"""
4from fastapi import FastAPI
5from pydantic import BaseModel
6from typing import List
7import uuid
8from datetime import datetime
9
10app = FastAPI()
11
12# A2A 消息结构模型
13class Part(BaseModel):
14    kind: str
15    text: str = None
16
17class Message(BaseModel):
18    kind: str = "message"
19    messageId: str
20    role: str
21    parts: List[Part]
22
23class MessageParams(BaseModel):
24    message: Message
25
26class JSONRPCRequest(BaseModel):
27    jsonrpc: str = "2.0"
28    id: str
29    method: str
30    params: MessageParams
31
32@app.get("/.well-known/agent-card.json")
33def agent_card():
34    """A2A 代理发现端点"""
35    return {
36        "name": "回声代理",
37        "description": "一个简单的代理，会将您的消息原样返回",
38        "url": "http://localhost:8000",
39        "version": "1.0.0",
40        "protocolVersion": "0.3.0",
41        "skills": [{
42            "id": "echo",
43            "name": "回声消息",
44            "description": "重复您说的任何内容",
45            "examples": ["你好", "你怎么样？"]
46        }],
47        "capabilities": {"streaming": False},
48        "defaultInputModes": ["text/plain"],
49        "defaultOutputModes": ["text/plain"]
50    }
51
52@app.get("/")
53def hello():
54    return {"message": "来自我们的 A2A 代理的问候！访问 /.well-known/agent-card.json 查看我的能力。"}
55
56@app.post("/")
57def handle_message(request: JSONRPCRequest):
58    """A2A 消息处理器"""
59    # 验证方法
60    if request.method != "message/send":
61        return {
62            "jsonrpc": "2.0", 
63            "id": request.id, 
64            "error": {
65                "code": -32601, 
66                "message": "方法未找到"
67            }
68        }
69    
70    # 提取用户消息
71    user_message = request.params.message
72    user_text = user_message.parts[0].text if user_message.parts else "无文本"
73    
74    # 创建我们的 A2A 响应作为已完成的任务
75    task = {
76        "kind": "task",
77        "id": str(uuid.uuid4()),
78        "contextId": str(uuid.uuid4()),
79        "status": {
80            "state": "completed", 
81            "timestamp": datetime.utcnow().isoformat() + "Z"
82        },
83        "history": [
84            # 包含原始用户消息
85            user_message.dict(),
86            # 添加我们的代理回复
87            {
88                "kind": "message",
89                "messageId": str(uuid.uuid4()),
90                "role": "agent",
91                "parts": [{"kind": "text", "text": f"您说：'{user_text}'"}]
92            }
93        ]
94    }
95    
96    return {"jsonrpc": "2.0", "id": request.id, "result": task}
97
98if __name__ == "__main__":
99    import uvicorn
100    print("🤖 在 http://localhost:8000 启动简单的 A2A 回声代理")
101    uvicorn.run(app, host="0.0.0.0", port=8000)

1import requests
2import uuid
3import json
4
5def main():
6    # 步骤 1：发现代理
7    base_url = "http://localhost:8000"
8    print("🔍 发现 A2A 代理...")
9    
10    try:
11        agent_card = requests.get(f"{base_url}/.well-known/agent-card.json").json()
12        print(f"✅ 找到：{agent_card['name']} - {agent_card['description']}")
13    except requests.RequestException as e:
14        print(f"❌ 发现失败：{e}")
15        return
16    
17    # 步骤 2：使用 A2A JSON-RPC 发送消息
18    print("\n💬 发送消息...")
19    
20    message = {
21        "jsonrpc": "2.0",
22        "id": str(uuid.uuid4()),
23        "method": "message/send",
24        "params": {
25            "message": {
26                "kind": "message",
27                "messageId": str(uuid.uuid4()),
28                "role": "user",
29                "parts": [{"kind": "text", "text": "你好 A2A 世界！"}]
30            }
31        }
32    }
33    
34    try:
35        response = requests.post(base_url, json=message).json()
36    except requests.RequestException as e:
37        print(f"❌ 通信失败：{e}")
38        return
39    
40    # 步骤 3：处理响应
41    if "result" in response:
42        task = response["result"] 
43        print(f"📋 任务状态：{task['status']['state']}")
44        
45        # 在历史记录中查找代理回复
46        for msg in task.get("history", []):
47            if msg.get("role") == "agent":
48                agent_reply = msg["parts"][0]["text"]
49                print(f"🤖 代理：{agent_reply}")
50                break
51    else:
52        error = response.get("error", {})
53        print(f"❌ 错误：{error.get('message', '未知错误')}")
54    
55    print("\n✨ A2A 通信完成！")
56
57if __name__ == "__main__":
58    main()

代码示例

1python echo_agent.py

代码示例

1cd app
2python simple_a2a_client.py

代码示例

1🔍 发现 A2A 代理...
2✅ 找到：回声代理 - 一个简单的代理，会将您的消息原样返回
3
4💬 发送消息...
5📋 任务状态：completed
6🤖 代理：您说：'你好 A2A 世界！'
7
8✨ A2A 通信完成！

代码示例

1curl http://0.0.0.0:8000/.well-known/agent-card.json