把概念、原理、失败模式拼成一个真实的 A2A + MCP 两层栈

# 未在本机验证 —— a2a-sdk 1.1.0（spec v1.0）
from a2a.helpers import get_message_text, new_task_from_user_message, new_data_part, new_text_part
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.tasks import TaskUpdater
from a2a.types.a2a_pb2 import TaskState


class WebResearchExecutor(AgentExecutor):
    """A2A 专家 agent：对上是 A2A server，对下用 MCP 调自己的工具（图 5.1 两层）。"""

    def __init__(self, mcp_tools):
        # mcp_tools 是一个已连上 MCP server 的工具客户端（fetch / search）。
        # 它是「专家 agent ↓ 工具」这条垂直 MCP 边界（D3），不是又一个 A2A agent。
        self.mcp = mcp_tools

    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        # ── A2A 入口：收编排方那条显式 Message（D2 不透明：只看这条，不碰对方 state）──
        if context.current_task:
            task = context.current_task
        else:
            task = new_task_from_user_message(context.message)
            await event_queue.enqueue_event(task)

        updater = TaskUpdater(
            event_queue=event_queue,
            task_id=task.id,
            context_id=task.context_id,
        )
        await updater.update_status(state=TaskState.TASK_STATE_WORKING)

        topic = get_message_text(context.message) or "(空)"

        # ── 垂直 MCP 边界：在自己隔离的上下文里调工具干活（D3 工具 + D2 隔离）──
        # 这几行就是 MCP tools/call —— 给查询/URL、回结构化结果，无状态原语。
        hits = await self.mcp.call("search", {"q": f"{topic} pricing"})       # MCP 工具①
        pages = [await self.mcp.call("fetch", {"url": h["url"]}) for h in hits[:3]]  # MCP 工具②

        # ── 直接函数调用：纯本地归纳，无 IO、无信任边界（D3 第三条路）──
        findings = summarize_pricing(topic, pages)   # -> [{"claim":..., "source_url":...}, ...]

        # ── A2A 出口：把结论包成 Artifact 回传，每条挂来源 URL（§1.3 Artifact = Part[]）──
        await updater.add_artifact(
            parts=[
                new_text_part(text=render_md(findings), media_type="text/markdown"),
                new_data_part(data={"findings": findings}),   # 结构化，编排方可机械解析
            ],
            name="competitor-pricing",
        )
        await updater.update_status(state=TaskState.TASK_STATE_COMPLETED)

    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> None:
        # 本例任务短；真实长任务应在此置 TASK_STATE_CANCELED（见 03 章 §3.2 predict）
        raise NotImplementedError("Cancel is not supported.")

# 未在本机验证 —— a2a-sdk 1.1.0（spec v1.0）
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.tasks import InMemoryTaskStore
from a2a.types import AgentCapabilities, AgentCard, AgentInterface, AgentSkill

from research_executor import WebResearchExecutor
from mcp_tools import connect_mcp_tools   # 连上挂着 fetch/search 的 MCP server

PORT = 9100
BASE_URL = f"http://127.0.0.1:{PORT}"

# ── 一项 skill：对外声明「我能做竞品定价调研」（§1.2 skills）──
skill = AgentSkill(
    id="competitor-pricing-research",
    name="Competitor Pricing Research",
    description="给定竞品名，调研其公开定价并返回带来源的结论。",
    input_modes=["text/plain"],
    output_modes=["text/markdown"],
    tags=["research", "pricing"],
    examples=["调研竞品 X 的定价"],
)

# ── Agent Card：对外的一纸能力声明（§1.2 / §2.2）。D1：本例不开流式，同步交互即可 ──
agent_card = AgentCard(
    name="Web Research Agent",
    description="一个用 MCP 工具做网页调研的 A2A 专家 agent。",
    version="1.0.0",
    default_input_modes=["text/plain"],
    default_output_modes=["text/markdown"],
    capabilities=AgentCapabilities(streaming=False),     # D1 同步：不声明流式
    supported_interfaces=[AgentInterface(protocol_binding="JSONRPC", url=BASE_URL)],
    skills=[skill],
)

handler = DefaultRequestHandler(
    agent_executor=WebResearchExecutor(connect_mcp_tools()),   # 注入 MCP 工具
    task_store=InMemoryTaskStore(),
    agent_card=agent_card,
)

routes = []
routes.extend(create_agent_card_routes(agent_card))   # GET /.well-known/agent-card.json
routes.extend(create_jsonrpc_routes(handler, "/"))    # POST / —— JSON-RPC SendMessage 入口
app = Starlette(routes=routes)

if __name__ == "__main__":
    uvicorn.run(app, host="127.0.0.1", port=PORT)

# 未在本机验证 —— a2a-sdk 1.1.0（spec v1.0）
import asyncio
import httpx

from a2a.client import A2ACardResolver, ClientConfig, ClientFactory
from a2a.helpers import get_stream_response_text, new_text_message
from a2a.types.a2a_pb2 import Role, SendMessageRequest

SPECIALIST_URL = "http://127.0.0.1:9100"
PINNED_KEYS = {"127.0.0.1:9100": {"<已登记的专家 agent 签发公钥>"}}


async def delegate_research(topic: str) -> None:
    async with httpx.AsyncClient() as httpx_client:
        # 1. 发现：GET 专家 agent 的 Card（§1.2 / §2.2）
        resolver = A2ACardResolver(httpx_client=httpx_client, base_url=SPECIALIST_URL)
        agent_card = await resolver.get_agent_card()

        # 2. D4 §4.2：先验 Card 签名，公钥不在信任清单就拒——不验就委托是裸奔
        verify_card_signature(agent_card, PINNED_KEYS)   # 缺签名/不符即抛错

        # 3. D2 不透明：只发一条显式意图 Message，不把编排方内部 state 推过去
        config = ClientConfig(streaming=False, httpx_client=httpx_client)   # D1 同步
        client = ClientFactory(config).create(agent_card)
        message = new_text_message(f"调研竞品 {topic} 的定价，给出带来源的结论。",
                                   role=Role.ROLE_USER)
        request = SendMessageRequest(message=message)

        # 4. 同步收回 Artifact（D1：一次 send_message 拿回结果）
        async for chunk in client.send_message(request):
            artifact_text = get_stream_response_text(chunk)
            # 5. D4 §4.3：对端输出当「数据」，绝不原样当指令喂回自己的 LLM
            store_as_quoted_data(artifact_text)          # 进数据通道，不进 system 指令通道

        await client.close()


if __name__ == "__main__":
    asyncio.run(delegate_research("X"))