init: A2A protocol demo project

.gitignore

@@ -0,0 +1 @@
+.venv/

.idea/.gitignore

@@ -0,0 +1,5 @@
+# 默认忽略的文件
+/shelf/
+/workspace.xml
+# 基于编辑器的 HTTP 客户端请求
+/httpRequests/

.idea/a2a-demo.iml

@@ -0,0 +1,14 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<module type="PYTHON_MODULE" version="4">
+  <component name="NewModuleRootManager">
+    <content url="file://$MODULE_DIR$">
+      <excludeFolder url="file://$MODULE_DIR$/.venv" />
+    </content>
+    <orderEntry type="jdk" jdkName="uv (a2a-demo)" jdkType="Python SDK" />
+    <orderEntry type="sourceFolder" forTests="false" />
+  </component>
+  <component name="PyDocumentationSettings">
+    <option name="format" value="PLAIN" />
+    <option name="myDocStringFormat" value="Plain" />
+  </component>
+</module>

.idea/inspectionProfiles/Project_Default.xml

@@ -0,0 +1,6 @@
+<component name="InspectionProjectProfileManager">
+  <profile version="1.0">
+    <option name="myName" value="Project Default" />
+    <inspection_tool class="Eslint" enabled="true" level="WARNING" enabled_by_default="true" />
+  </profile>
+</component>

.idea/inspectionProfiles/profiles_settings.xml

@@ -0,0 +1,6 @@
+<component name="InspectionProjectProfileManager">
+  <settings>
+    <option name="USE_PROJECT_PROFILE" value="false" />
+    <version value="1.0" />
+  </settings>
+</component>

.idea/modules.xml

@@ -0,0 +1,8 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="ProjectModuleManager">
+    <modules>
+      <module fileurl="file://$PROJECT_DIR$/.idea/a2a-demo.iml" filepath="$PROJECT_DIR$/.idea/a2a-demo.iml" />
+    </modules>
+  </component>
+</project>

README.md

@@ -0,0 +1,149 @@
+# A2A Agent-to-Agent 演示项目
+
+A2A 协议（Google）的 Python 演示。对比「有 A2A」和「没有 A2A」两种集成方式。
+
+## 项目结构
+
+```
+a2a-demo/
+├── server.py                # 问答 Agent（端口 9999）→ A2A Server
+├── server_translator.py     # 翻译 Agent（端口 9998）→ A2A Server
+├── client.py                # A2A 全流程演示（发现 + 调用 + 跨 Agent 管道）
+├── compare.py               # 交互式对比：直连 DeepSeek vs 走 A2A
+├── demo_no_a2a.py           # ❌ 无 A2A：两个 Agent 不同 API，手工适配
+└── demo_with_a2a.py         # ✅ 有 A2A：同一套协议，零适配代码
+```
+
+## 先决条件
+
+Python 3.10+（推荐 3.11），项目自带 uv 虚拟环境。
+
+```bash
+cd ~/Desktop/miaoa/AI/projects/a2a-demo
+source .venv/bin/activate
+```
+
+如果虚拟环境不存在，重建：
+
+```bash
+cd ~/Desktop/miaoa/AI/projects/a2a-demo
+uv venv --python 3.11
+uv pip install a2a-sdk uvicorn starlette openai httpx sse-starlette httpx-sse
+```
+
+---
+
+## 一、快速体验：A2A vs 无 A2A 对比
+
+需要 3 个终端。
+
+### 终端 1 — 启动问答 Agent（A2A Server）
+
+```bash
+cd ~/Desktop/miaoa/AI/projects/a2a-demo
+source .venv/bin/activate
+python server.py
+```
+
+看到输出：
+```
+🚀 A2A Server 启动 → http://127.0.0.1:9999
+INFO:     Uvicorn running on http://127.0.0.1:9999
+```
+
+### 终端 2 — 启动翻译 Agent（A2A Server）
+
+```bash
+cd ~/Desktop/miaoa/AI/projects/a2a-demo
+source .venv/bin/activate
+python server_translator.py
+```
+
+看到输出：
+```
+🌐 翻译 Agent → http://127.0.0.1:9998
+INFO:     Uvicorn running on http://127.0.0.1:9998
+```
+
+### 终端 3 — 运行对比 Demo
+
+**先看没有 A2A 的情况：**
+
+```bash
+cd ~/Desktop/miaoa/AI/projects/a2a-demo
+source .venv/bin/activate
+python demo_no_a2a.py
+```
+
+**再看有 A2A 的情况：**
+
+```bash
+python demo_with_a2a.py
+```
+
+### 预期效果
+
+```
+❌ 没有 A2A
+   问答 → 手动写: POST /ask  {"question": ...}
+   翻译 → 手动写: POST /v2/translate  {"src": ...}  → 解析 resp.data.result
+   第三个 → 再写一套适配代码
+   → N 个 Agent = N 套适配代码
+
+✅ 有 A2A
+   问答 → call_a2a_agent(url, question)  ← 同一函数
+   翻译 → call_a2a_agent(url, question)  ← 同一函数
+   第三个 → call_a2a_agent(url, question)  ← 直接复用
+   → N 个 Agent = 1 套代码
+```
+
+---
+
+## 二、交互式对比测试
+
+一边是直连 DeepSeek API，一边是经过 A2A 协议转发。
+
+**前提**：问答 Agent（server.py）必须在 9999 端口运行。
+
+```bash
+# 终端 1：启动 Server
+cd ~/Desktop/miaoa/AI/projects/a2a-demo
+source .venv/bin/activate
+python server.py
+
+# 终端 2：运行对比
+cd ~/Desktop/miaoa/AI/projects/a2a-demo
+source .venv/bin/activate
+python compare.py
+```
+
+进入交互模式后，输入问题即可看到两种方式的结果和耗时。
+
+---
+
+## 三、A2A 全流程演示
+
+展示 A2A 协议核心价值：Agent Card 发现 + 统一协议调用 + 跨 Agent 管道。
+
+**前提**：问答 Agent（9999）和翻译 Agent（9998）都必须运行。
+
+```bash
+cd ~/Desktop/miaoa/AI/projects/a2a-demo
+source .venv/bin/activate
+python client.py
+```
+
+输出三段：
+1. **Phase 1** — Client 发现两个 Agent 的 Agent Card
+2. **Phase 2** — 分别调用问答和翻译
+3. **Phase 3** — 跨 Agent 管道：问答结果 → 翻译
+
+---
+
+## 注意事项
+
+| 问题 | 解决 |
+|------|------|
+| 端口被占用 | `lsof -ti:9999 -ti:9998 \| xargs kill -9` |
+| 502 Bad Gateway | AgentCard 序列化异常，检查 server.py 中没有 `url` 字段 |
+| pip install a2a-sdk 失败 | 需要 Python ≥ 3.10，本项目用 uv 管理 |

client.py

@@ -0,0 +1,132 @@
+"""
+A2A Demo Client — 展示 A2A 协议核心价值
+
+流程：
+1. 发现两个 Agent 的能力（Agent Card）
+2. 分别提交任务给不同的 Agent
+3. 展示跨 Agent 协作：问答 Agent 回答 → 翻译 Agent 翻译
+"""
+import asyncio
+import httpx
+from a2a.client import A2ACardResolver, ClientConfig, create_client
+from a2a.helpers import new_text_message, display_agent_card
+from a2a.types import Role, SendMessageRequest
+
+AGENTS = {
+    "问答": "http://127.0.0.1:9999",
+    "翻译": "http://127.0.0.1:9998",
+}
+
+
+def extract_text(task) -> str:
+    """从 Task 的 artifacts 中提取回复文本"""
+    texts = []
+    if task.artifacts:
+        for artifact in task.artifacts:
+            for part in artifact.parts:
+                t = part.text.strip() if part.text else ""
+                if t:
+                    texts.append(t)
+    return "\n".join(texts)
+
+
+async def main():
+    async with httpx.AsyncClient() as http_client:
+        # ═══════════════════════════════════════════════
+        # Phase 1: 发现多个 Agent 的能力
+        # ═══════════════════════════════════════════════
+        print("=" * 70)
+        print("  Phase 1: 🔍 发现 Agent 能力（Agent Card）")
+        print("  A2A 协议让 Client 可以在运行期发现 Agent 能干什么")
+        print("=" * 70)
+
+        cards = {}
+        for name, url in AGENTS.items():
+            print(f"\n  ▶ 发现 {name} Agent → {url}")
+            resolver = A2ACardResolver(httpx_client=http_client, base_url=url)
+            card = await resolver.get_agent_card()
+            cards[name] = card
+            # 只取关键信息展示，不打印全量
+            skills_str = ", ".join(s.name for s in card.skills) if card.skills else "(无)"
+            print(f"     Agent: {card.name}  |  技能: {skills_str}")
+
+        # ═══════════════════════════════════════════════
+        # Phase 2: 分别调用不同的 Agent
+        # ═══════════════════════════════════════════════
+        print("\n" + "=" * 70)
+        print("  Phase 2: 🤝 通过相同协议调用不同 Agent")
+        print("  不管是问答还是翻译，Client 都走同一套 A2A 接口")
+        print("=" * 70)
+
+        for name, card in cards.items():
+            print(f"\n  ── {name} Agent ──")
+            config = ClientConfig(streaming=False)
+            client = await create_client(agent=card, client_config=config)
+
+            # 给每个 Agent 一个适合它的任务
+            if name == "问答":
+                query = "Explain what A2A protocol is in one sentence"
+            else:
+                query = "请把下面这句话翻译成英文：人工智能正在改变世界"
+
+            print(f"  输入: {query}")
+
+            message = new_text_message(query, role=Role.ROLE_USER)
+            request = SendMessageRequest(message=message)
+
+            async for response in client.send_message(request):
+                text = extract_text(response.task)
+                if text:
+                    # 缩进展示，区分输入输出
+                    for line in text.split("\n"):
+                        print(f"  输出: {line}")
+
+            await client.close()
+
+        # ═══════════════════════════════════════════════
+        # Phase 3: 跨 Agent 协作（这才是 A2A 的精髓）
+        # ═══════════════════════════════════════════════
+        print("\n" + "=" * 70)
+        print("  Phase 3: 🔄 跨 Agent 协作管道")
+        print("  问答 Agent 的回答 → 翻译 Agent 翻译成英文")
+        print("  不同 Agent 通过标准协议接力，无需定制适配")
+        print("=" * 70)
+
+        # 3a. 先问问题
+        question = "什么是 A2A 协议？用一句话说"
+        print(f"\n  Step 1: 用户提问 → 问答 Agent")
+        print(f"  输入: {question}")
+
+        qa_card = cards["问答"]
+        qa_client = await create_client(agent=qa_card, client_config=ClientConfig(streaming=False))
+        qa_msg = new_text_message(question, role=Role.ROLE_USER)
+        async for resp in qa_client.send_message(SendMessageRequest(message=qa_msg)):
+            qa_answer = extract_text(resp.task)
+        await qa_client.close()
+        print(f"  问答 Agent 回复: {qa_answer[:100]}...")
+
+        # 3b. 把回答送去翻译
+        print(f"\n  Step 2: 问答结果 → 翻译 Agent")
+        print(f"  输入: (上一步的回答，翻译成英文)")
+
+        trans_card = cards["翻译"]
+        trans_client = await create_client(agent=trans_card, client_config=ClientConfig(streaming=False))
+        trans_msg = new_text_message(f"把下面这句话翻译成英文：{qa_answer}", role=Role.ROLE_USER)
+        async for resp in trans_client.send_message(SendMessageRequest(message=trans_msg)):
+            trans_result = extract_text(resp.task)
+        await trans_client.close()
+        print(f"  翻译 Agent 输出: {trans_result[:200]}...")
+
+        # ═══════════════════════════════════════════════
+        print("\n" + "=" * 70)
+        print("  ✅ A2A Demo 完成")
+        print("  🔑 关键点：")
+        print("  • 两个 Agent 能力完全不同（问答 vs 翻译）")
+        print("  • Client 通过 Agent Card 动态发现它们的能力")
+        print("  • Client 用同一套 A2A 协议与两者通信，零适配代码")
+        print("  • 跨 Agent 管道自动流转，无需中间格式转换")
+        print("=" * 70)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

compare.py

@@ -0,0 +1,131 @@
+"""
+A2A vs 直连 对比测试
+交互式输入问题，同时展示两种调用方式的结果
+
+运行前提：问答 Agent (server.py) 必须在 9999 端口运行
+"""
+import asyncio
+import httpx
+from datetime import datetime
+from openai import AsyncOpenAI
+from a2a.client import A2ACardResolver, ClientConfig, create_client
+from a2a.helpers import new_text_message
+from a2a.types import Role, SendMessageRequest
+
+DEEPSEEK_API_KEY = "你的APIKEY"
+A2A_SERVER_URL = "http://127.0.0.1:9999"
+
+
+def extract_text(task) -> str:
+    texts = []
+    if task.artifacts:
+        for artifact in task.artifacts:
+            for part in artifact.parts:
+                t = part.text.strip() if part.text else ""
+                if t:
+                    texts.append(t)
+    return "\n".join(texts)
+
+
+async def direct_call(question: str) -> str:
+    """方式 A：直连 DeepSeek，不经过 A2A"""
+    client = AsyncOpenAI(
+        api_key=DEEPSEEK_API_KEY,
+        base_url="https://api.deepseek.com/v1",
+    )
+    response = await client.chat.completions.create(
+        model="deepseek-chat",
+        messages=[
+            {"role": "system", "content": "你是一个有用的 AI 助手，请用简洁的中文回答用户的问题。"},
+            {"role": "user", "content": question},
+        ],
+        temperature=0.7,
+        max_tokens=1024,
+    )
+    result = response.choices[0].message.content
+    await client.close()
+    return result
+
+
+async def a2a_call(question: str) -> str:
+    """方式 B：走 A2A 协议 → Server → DeepSeek"""
+    async with httpx.AsyncClient() as http_client:
+        resolver = A2ACardResolver(httpx_client=http_client, base_url=A2A_SERVER_URL)
+        agent_card = await resolver.get_agent_card()
+
+        config = ClientConfig(streaming=False)
+        client = await create_client(agent=agent_card, client_config=config)
+        message = new_text_message(question, role=Role.ROLE_USER)
+        request = SendMessageRequest(message=message)
+
+        result = ""
+        async for response in client.send_message(request):
+            result = extract_text(response.task)
+
+        await client.close()
+        return result
+
+
+async def main():
+    print("=" * 72)
+    print("  A2A vs 直连 — 交互式对比测试")
+    print("  =============================")
+    print("  方式 A：Client → DeepSeek API（直连，无 A2A）")
+    print("  方式 B：Client → A2A Server → DeepSeek API（经过 A2A 协议）")
+    print("=" * 72)
+
+    # 先检查 A2A Server 是否在运行
+    print("\n[*] 检查 A2A Server 状态...", end=" ")
+    try:
+        async with httpx.AsyncClient() as c:
+            r = await c.get(f"{A2A_SERVER_URL}/.well-known/agent-card.json", timeout=3)
+            if r.status_code == 200:
+                print("✅ 运行中")
+            else:
+                print(f"⚠️ 响应异常 ({r.status_code})")
+    except Exception:
+        print("❌ 未运行！请先启动：python server.py")
+        return
+
+    while True:
+        question = input("\n" + "─" * 72 + "\n输入问题（输入 exit 退出）> ").strip()
+        if not question:
+            continue
+        if question.lower() in ("exit", "quit", "q"):
+            print("bye")
+            break
+
+        print()
+
+        # 方式 A：直连
+        t0 = datetime.now()
+        print("  ┌─ [A] 直连 DeepSeek ──────────────────────────────")
+        try:
+            direct_result = await direct_call(question)
+            t1 = datetime.now()
+            for line in direct_result.split("\n"):
+                print(f"  │ {line}")
+            print(f"  └─ ⏱ {(t1 - t0).total_seconds():.2f}s")
+        except Exception as e:
+            print(f"  │ ❌ {e}")
+            print("  └─")
+
+        print()
+
+        # 方式 B：走 A2A
+        t2 = datetime.now()
+        print("  ┌─ [B] 通过 A2A 协议 ──────────────────────────────")
+        print("  │   Client → A2A Server → DeepSeek")
+        try:
+            a2a_result = await a2a_call(question)
+            t3 = datetime.now()
+            for line in a2a_result.split("\n"):
+                print(f"  │ {line}")
+            print(f"  └─ ⏱ {(t3 - t2).total_seconds():.2f}s")
+        except Exception as e:
+            print(f"  │ ❌ {e}")
+            print("  └─")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

demo_no_a2a.py

@@ -0,0 +1,91 @@
+"""
+❌ 非 A2A 模式：两个 Agent 各自有完全不同的 API
+需要手动适配，调用方必须提前知道每个 Agent 的接口细节
+"""
+from openai import AsyncOpenAI
+import asyncio
+
+DEEPSEEK_API_KEY = "你的APIKEY"
+client = AsyncOpenAI(api_key=DEEPSEEK_API_KEY, base_url="https://api.deepseek.com/v1")
+
+
+async def call_llm(system: str, user: str) -> str:
+    r = await client.chat.completions.create(
+        model="deepseek-chat",
+        messages=[{"role": "system", "content": system}, {"role": "user", "content": user}],
+        temperature=0.3, max_tokens=1024,
+    )
+    return r.choices[0].message.content
+
+
+# ─── "Agent" A：问答服务 ─────────────────────────────
+# 假装这是一个外部的问答 API，格式如下：
+#   POST /ask  {"question": "..."} → {"answer": "..."}
+# 调用方必须：知道 URL、知道字段名是 "question"、"answer"
+
+async def agent_qa(question: str) -> str:
+    """模拟调用一个自定义问答 API"""
+    print("  [适配] POST /ask  body={\"question\": \"...\"}")
+    return await call_llm(
+        "你是一个有用的AI助手",
+        question,
+    )
+
+
+# ─── "Agent" B：翻译服务 ─────────────────────────────
+# 假装这是另一个公司的翻译 API，格式完全不一样：
+#   POST /v2/translate  {"src": "...", "from": "zh", "to": "en"}
+#   → {"code": 0, "data": {"result": "..."}}
+# 调用方必须：知道不同 URL、不同字段名、不同响应结构
+
+async def agent_translate(text: str) -> str:
+    """模拟调用一个自定义翻译 API"""
+    print("  [适配] POST /v2/translate  body={\"src\": \"...\", \"from\": \"zh\", \"to\": \"en\"}")
+    print("  [适配] 解析响应: resp.data.result")
+    return await call_llm(
+        "你是一个专业中译英翻译，只输出翻译结果",
+        f"把以下中文翻译成英文：{text}",
+    )
+
+
+# ─── 调用方：没有 A2A，必须手工适配每个 Agent ────
+
+async def main():
+    question = "什么是微服务架构？"
+
+    print("=" * 64)
+    print("  ❌ 没有 A2A 协议")
+    print("  两个 Agent，两种 API，两套适配代码")
+    print("=" * 64)
+
+    print(f"\n  用户: {question}")
+
+    # Step 1: 必须先知道问答 API 的地址和格式
+    print("\n  ── 调用 问答 Agent ──")
+    qa_result = await agent_qa(question)
+    print(f"  回答: {qa_result[:80]}...")
+
+    # Step 2: 想把回答翻译成英文
+    #   → 又得知道翻译 API 的地址和格式（完全不同！）
+    print("\n  ── 调用 翻译 Agent ──")
+    print("  ⚠ 需要写全新的适配代码：不同 URL、不同字段名、不同响应结构")
+    trans_result = await agent_translate(qa_result)
+    print(f"  翻译: {trans_result[:80]}...")
+
+    # Step 3: 如果想再加第三个 Agent（比如总结）
+    #   → 又要第三套适配代码
+    print("\n  ❌ 再加第3个 Agent → 再写第3套适配")
+    print("  N 个 Agent = N 套适配代码（API 格式差异越大越痛苦）")
+    print()
+
+    # ===== 显示对比总结 =====
+    print("-" * 64)
+    print("  对比：")
+    print("  问答 API: POST /ask  {\"question\": ...} -> {\"answer\": ...}")
+    print("  翻译 API: POST /v2/translate  {\"src\": ...} -> {\"code\":0, \"data\":{\"result\":...}}")
+    print("  → 完全不同的 URL、字段名、响应结构")
+    print("  → 每接入一个 Agent 都要看文档写适配")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

demo_with_a2a.py

@@ -0,0 +1,85 @@
+"""
+✅ A2A 模式：两个 Agent 通过统一协议通信
+调用方不需要知道 API 细节，Agent Card 动态发现能力
+"""
+import asyncio
+import httpx
+from a2a.client import A2ACardResolver, ClientConfig, create_client
+from a2a.helpers import new_text_message
+from a2a.types import Role, SendMessageRequest
+
+A2A_QA_URL = "http://127.0.0.1:9999"
+A2A_TRANSLATE_URL = "http://127.0.0.1:9998"
+
+
+def extract_text(task) -> str:
+    texts = []
+    if task.artifacts:
+        for artifact in task.artifacts:
+            for part in artifact.parts:
+                t = part.text.strip() if part.text else ""
+                if t:
+                    texts.append(t)
+    return "\n".join(texts)
+
+
+async def call_a2a_agent(url: str, question: str) -> str:
+    """通过 A2A 协议调用任意 Agent（统一接口！）"""
+    async with httpx.AsyncClient() as http_client:
+        # 1. Agent Card 发现（自动获取能力描述）
+        resolver = A2ACardResolver(httpx_client=http_client, base_url=url)
+        agent_card = await resolver.get_agent_card()
+        print(f"  [发现] {agent_card.name} | 技能: {[s.name for s in agent_card.skills]}")
+
+        # 2. 统一接口调用（不管什么 Agent 都一样）
+        config = ClientConfig(streaming=False)
+        client = await create_client(agent=agent_card, client_config=config)
+        message = new_text_message(question, role=Role.ROLE_USER)
+        request = SendMessageRequest(message=message)
+
+        result = ""
+        async for response in client.send_message(request):
+            result = extract_text(response.task)
+
+        await client.close()
+        return result
+
+
+async def main():
+    question = "什么是微服务架构？"
+
+    print("=" * 64)
+    print("  ✅ 使用 A2A 协议")
+    print("  两个 Agent，同一套协议，零适配代码")
+    print("=" * 64)
+
+    print(f"\n  用户: {question}")
+
+    # Step 1: 调用问答 Agent — 通过 A2A
+    print("\n  ── 调用 问答 Agent ──")
+    qa_result = await call_a2a_agent(A2A_QA_URL, question)
+    print(f"  回答: {qa_result[:80]}...")
+
+    # Step 2: 把回答送去翻译 — 同一个函数，换个 URL 而已
+    print("\n  ── 调用 翻译 Agent ──")
+    print("  ✓ 同一套 send_message()，无需任何适配")
+    trans_result = await call_a2a_agent(A2A_TRANSLATE_URL, f"把下面这句话翻译成英文：{qa_result}")
+    print(f"  翻译: {trans_result[:80]}...")
+
+    # Step 3: 再加第三个 Agent 怎么办？
+    print("\n  ── 再加第 3 个 Agent ──")
+    print("  ✓ call_a2a_agent(url, question) 直接复用")
+    print("  ✓ 换个 URL 和问题就行，代码一行不改")
+
+    # ===== 对比总结 =====
+    print()
+    print("-" * 64)
+    print("  对比：")
+    print("  问答 Agent: A2A 协议 → send_message()")
+    print("  翻译 Agent: A2A 协议 → send_message()")
+    print("  → 两个 Agent 共享同一套接口")
+    print("  → 新增 Agent 只需知道 URL，无需适配代码")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

server.py

@@ -0,0 +1,132 @@
+"""
+A2A Server — 基于 DeepSeek 的问答 Agent
+对外暴露 A2A 协议接口，让其他 Agent 可以发现自己并提交任务。
+"""
+import os
+import uvicorn
+from starlette.applications import Starlette
+
+from a2a.server.request_handlers import DefaultRequestHandler
+from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
+from a2a.server.agent_execution import AgentExecutor, RequestContext
+from a2a.server.events import EventQueue
+from a2a.server.tasks import InMemoryTaskStore, TaskUpdater
+from a2a.types import (
+    AgentCapabilities, AgentCard, AgentInterface, AgentSkill,
+    TaskState,
+)
+from a2a.helpers import (
+    get_message_text, new_task_from_user_message,
+    new_text_message, new_text_part,
+)
+from openai import AsyncOpenAI
+
+# ── DeepSeek Agent ─────────────────────────────────────────
+
+DEEPSEEK_API_KEY = os.getenv("DEEPSEEK_API_KEY", "你的APIKEY")
+
+
+class MyAgent:
+    """用 DeepSeek 回答问题的 Agent。"""
+
+    def __init__(self):
+        self.client = AsyncOpenAI(
+            api_key=DEEPSEEK_API_KEY,
+            base_url="https://api.deepseek.com/v1",
+        )
+
+    async def invoke(self, user_request: str) -> str:
+        response = await self.client.chat.completions.create(
+            model="deepseek-chat",
+            messages=[
+                {"role": "system", "content": "你是一个有用的 AI 助手，请用简洁的中文回答用户的问题。"},
+                {"role": "user", "content": user_request},
+            ],
+            temperature=0.7,
+            max_tokens=1024,
+        )
+        return response.choices[0].message.content
+
+
+# ── Agent Executor ─────────────────────────────────────────
+
+class MyAgentExecutor(AgentExecutor):
+    """A2A Executor：把 A2A 任务转发给 MyAgent 处理。"""
+
+    def __init__(self):
+        self.agent = MyAgent()
+
+    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
+        # 1. 获取或创建 Task
+        if context.current_task:
+            task = context.current_task
+        else:
+            task = new_task_from_user_message(context.message)
+            await event_queue.enqueue_event(task)
+
+        # 2. 更新状态：正在处理
+        updater = TaskUpdater(event_queue=event_queue, task_id=task.id, context_id=task.context_id)
+        await updater.start_work(message=new_text_message("正在处理..."))
+
+        # 3. 提取用户输入，调用 DeepSeek
+        query = get_message_text(context.message)
+        result = await self.agent.invoke(user_request=query) if query else "没有收到有效输入"
+
+        # 4. 把结果作为 Artifact 返回
+        await updater.add_artifact(parts=[new_text_part(text=result, media_type="text/plain")])
+
+        # 5. 标记任务完成
+        await updater.complete(message=new_text_message("处理完成！"))
+        print(f"  ✓ 处理完成: {result[:60]}...")
+
+    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
+        raise NotImplementedError("暂不支持取消")
+
+
+# ── Agent Card ─────────────────────────────────────────────
+
+skill = AgentSkill(
+    id="qa_bot",
+    name="问答机器人",
+    description="一个简单的问答 Agent，接收文本问题并返回 DeepSeek 的回答",
+    input_modes=["text/plain"],
+    output_modes=["text/plain"],
+    tags=["问答", "demo", "deepseek"],
+    examples=["你好", "什么是 A2A 协议？"],
+)
+
+agent_card = AgentCard(
+    name="问答 Agent",
+    description="一个通用问答 Agent，基于 DeepSeek 回答各种问题",
+    version="1.0.0",
+    default_input_modes=["text/plain"],
+    default_output_modes=["text/plain"],
+    capabilities=AgentCapabilities(streaming=True),
+    supported_interfaces=[
+        AgentInterface(
+            protocol_binding="JSONRPC",
+            url="http://127.0.0.1:9999",
+            protocol_version="1.0",
+        ),
+    ],
+    skills=[skill],
+)
+
+
+# ── 启动服务 ───────────────────────────────────────────────
+
+if __name__ == "__main__":
+    request_handler = DefaultRequestHandler(
+        agent_executor=MyAgentExecutor(),
+        task_store=InMemoryTaskStore(),
+        agent_card=agent_card,
+    )
+
+    routes = []
+    routes.extend(create_agent_card_routes(agent_card))
+    routes.extend(create_jsonrpc_routes(request_handler, "/"))
+
+    app = Starlette(routes=routes)
+    print("🚀 A2A Server 启动 → http://127.0.0.1:9999")
+    print(f"   Agent Card: http://127.0.0.1:9999/.well-known/agent-card.json")
+    uvicorn.run(app, host="127.0.0.1", port=9999)

server_translator.py

@@ -0,0 +1,101 @@
+"""
+翻译 Agent — 将中文翻译为英文
+作为独立 A2A Server 运行，暴露翻译技能
+"""
+import uvicorn
+from starlette.applications import Starlette
+
+from a2a.server.request_handlers import DefaultRequestHandler
+from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
+from a2a.server.agent_execution import AgentExecutor, RequestContext
+from a2a.server.events import EventQueue
+from a2a.server.tasks import InMemoryTaskStore, TaskUpdater
+from a2a.types import AgentCapabilities, AgentCard, AgentInterface, AgentSkill
+from a2a.helpers import (
+    get_message_text, new_task_from_user_message,
+    new_text_message, new_text_part,
+)
+from openai import AsyncOpenAI
+
+DEEPSEEK_API_KEY = "你的APIKEY"
+
+
+class TranslatorAgent:
+    def __init__(self):
+        self.client = AsyncOpenAI(
+            api_key=DEEPSEEK_API_KEY,
+            base_url="https://api.deepseek.com/v1",
+        )
+
+    async def translate(self, text: str) -> str:
+        response = await self.client.chat.completions.create(
+            model="deepseek-chat",
+            messages=[
+                {"role": "system", "content": "你是一个专业的中译英翻译。只输出翻译结果，不要添加解释。"},
+                {"role": "user", "content": f"把以下中文翻译成英文：{text}"},
+            ],
+            temperature=0.3,
+            max_tokens=1024,
+        )
+        return response.choices[0].message.content
+
+
+class TranslatorExecutor(AgentExecutor):
+    def __init__(self):
+        self.agent = TranslatorAgent()
+
+    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
+        task = context.current_task or new_task_from_user_message(context.message)
+        if not context.current_task:
+            await event_queue.enqueue_event(task)
+
+        updater = TaskUpdater(event_queue=event_queue, task_id=task.id, context_id=task.context_id)
+        await updater.start_work(message=new_text_message("正在翻译..."))
+
+        query = get_message_text(context.message)
+        result = await self.agent.translate(query) if query else "没有收到输入"
+        await updater.add_artifact(parts=[new_text_part(text=result, media_type="text/plain")])
+        await updater.complete(message=new_text_message("翻译完成！"))
+        print(f"  [翻译] {result[:60]}...")
+
+    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
+        raise NotImplementedError("暂不支持取消")
+
+
+# ── Agent Card ──
+
+skill = AgentSkill(
+    id="translator",
+    name="中译英翻译",
+    description="将中文文本翻译成英文",
+    input_modes=["text/plain"],
+    output_modes=["text/plain"],
+    tags=["翻译", "demo"],
+    examples=["你好吗？", "今天天气真好"],
+)
+
+agent_card = AgentCard(
+    name="翻译 Agent",
+    description="一个中译英翻译 Agent，接收中文返回英文",
+    version="1.0.0",
+    default_input_modes=["text/plain"],
+    default_output_modes=["text/plain"],
+    capabilities=AgentCapabilities(streaming=True),
+    supported_interfaces=[
+        AgentInterface(protocol_binding="JSONRPC", url="http://127.0.0.1:9998", protocol_version="1.0"),
+    ],
+    skills=[skill],
+)
+
+if __name__ == "__main__":
+    handler = DefaultRequestHandler(
+        agent_executor=TranslatorExecutor(),
+        task_store=InMemoryTaskStore(),
+        agent_card=agent_card,
+    )
+    routes = []
+    routes.extend(create_agent_card_routes(agent_card))
+    routes.extend(create_jsonrpc_routes(handler, "/"))
+    app = Starlette(routes=routes)
+    print("🌐 翻译 Agent → http://127.0.0.1:9998")
+    uvicorn.run(app, host="127.0.0.1", port=9998)