a2a/demo_with_a2a.py

"""
✅ 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())