C3-11 A2A协议

{
  "name": "代码智能体",
  "description": "编写代码的智能体",
  "url": "http://localhost:9999/",
  "capabilities": {
    "streaming": true,
    "pushNotifications": false
  },
  "skills": [
    {
      "id": "88",
      "name": "coding agent",
      "description": "编代码的智能体",
      "tags": ["编码", "代码", "coding"],
      "examples": ["编写一个 hello world 程序", "编写一个快速排序的 python 程序"]
    }
  ],
  "version": "1.0.0"
}

from a2a.types import Message, TextPart, MessageSendParams

# 构建一条发送给其他 Agent 的消息
message = Message(
    role="user",          # user = 发送方,agent = 接收方
    parts=[TextPart(text="帮我写一个 Python 的两数求和函数")],
    message_id="unique-id-123",
)

# 包装成请求
request = SendMessageRequest(
    id="req-001",
    params=MessageSendParams(message=message)
)

Task 状态流转:
  submitted(已提交)
      ↓
  working(工作中)
      ↓
  completed(完成)  或  failed(失败)  或  canceled(取消)

from a2a.types import TaskState
from a2a.server.tasks import TaskUpdater

# 在 Server 端更新任务状态
updater = TaskUpdater(event_queue, task.id, task.context_id)

# 工作中
await updater.update_status(TaskState.working, ...)

# 完成
await updater.add_artifact(parts=[...])
await updater.complete()

用户
  ↓ 提问
Entry Agent(主控/规划)
  ↓ 通过 A2A 协议发送消息
Coding Agent(专家执行者,运行在 A2A Server 上)
  ↓ 流式返回结果
Entry Agent 收到结果,展示给用户

# coding_agent.py
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.types import AgentCard, AgentSkill, AgentCapabilities
from a2a.server.tasks import InMemoryTaskStore
from openai import AsyncOpenAI
import os, uvicorn

# 1. 定义 Agent 技能
skill = AgentSkill(
    id="88",
    name="coding agent",
    description="编代码的智能体",
    tags=["编码", "代码", "coding"],

    examples=["编写一个 hello world 程序", "编写一个快速排序的 python 程序"],
)

# 2. 构建 Agent Card(名片)
coding_agent_card = AgentCard(
    name="代码智能体",
    description="编写代码的智能体",
    url="http://localhost:9999/",
    capabilities=AgentCapabilities(streaming=True),
    skills=[skill],
    defaultInputModes=["text"],
    defaultOutputModes=["text"],
    version="1.0.0",
)

# 3. 实现 Agent 执行逻辑
class CodingAgent:
    def __init__(self, api_key: str, base_url: str):
        self.client = AsyncOpenAI(api_key=api_key, base_url=base_url)

    async def stream(self, messages: list):
        response = await self.client.chat.completions.create(
            model="qwen-plus",
            messages=messages,
            stream=True,
        )
        async for chunk in response:
            if chunk.choices[0].delta.content:
                yield {
                    "is_final_answer": False,
                    "content": chunk.choices[0].delta.content,
                }
        yield {"is_final_answer": True, "content": "..."}

class CodingAgentExecutor:
    """A2A 要求的执行器接口"""
    async def execute(self, context, event_queue):
        task = context.current_task or new_task(context.message)
        updater = TaskUpdater(event_queue, task.id, task.context_id)

        # 调用 LLM 生成代码(流式)
        async for chunk in self.agent.stream(context.get_user_input()):
            if not chunk["is_final_answer"]:
                await updater.update_status(
                    TaskState.working,
                    new_agent_text_message(chunk["content"], ...)
                )
            else:
                await updater.add_artifact(...)
                await updater.complete()

# 4. 启动 Server
if __name__ == "__main__":
    request_handler = DefaultRequestHandler(
        agent_executor=CodingAgentExecutor(
            agent=CodingAgent(
                api_key=os.getenv("OPENAI_API_KEY"),
                base_url=os.getenv("OPENAI_BASE_URL")
            )
        ),
        task_store=InMemoryTaskStore(),
    )
    server = A2AStarletteApplication(
        agent_card=coding_agent_card,
        http_handler=request_handler,
    )
    uvicorn.run(server.build(), host="0.0.0.0", port=9999)

# entry_agent.py
import httpx
from a2a.client import A2AClient, A2ACardResolver
from a2a.types import Message, TextPart, SendStreamingMessageRequest
from uuid import uuid4

class EntryAgent:
    async def invoke(self, base_url: str, prompt: str):
        """通过 A2A 协议调用远程 Agent"""
        async with httpx.AsyncClient() as httpx_client:
            # 1. 获取对方的 Agent Card(自动发现能力)
            resolver = A2ACardResolver(
                httpx_client=httpx_client,
                base_url=base_url
            )
            agent_card = await resolver.get_agent_card()

            # 2. 初始化 A2A Client
            client = A2AClient(
                httpx_client=httpx_client,
                agent_card=agent_card
            )

            # 3. 构建消息并发送(流式)
            message = Message(
                role="user",
                parts=[TextPart(text=prompt)],
                message_id=uuid4().hex,
            )
            streaming_request = SendStreamingMessageRequest(
                id=str(uuid4()),
                params=MessageSendParams(message=message)
            )

            # 4. 处理流式响应
            async for chunk in client.send_message_streaming(streaming_request):
                yield chunk

# 运行
# asyncio.run(EntryAgent().invoke("http://localhost:9999", "帮我写一个快速排序"))

[用户]  "写一个 Python 快速排序"
    ↓
[Entry Agent]  ① 通过 A2A Client 发现 Coding Agent 的 Agent Card
    ↓          ② 构建 Message,调用 send_message_streaming()
[Coding Agent Server]
    ↓  ③ 收到请求,启动 LLM 生成代码(流式)
    ↓  ④ 通过 A2A 流式接口逐步返回生成内容
[Entry Agent]  ⑤ 收到流式 chunk,实时展示给用户
    ↓
[用户]  看到代码逐步生成...