第一章：通过 A2A 暴露 Deep Agent

A2A 是 AgenttoAgent 协议：调用方先读取远端的 Agent Card，了解它的接口和 Skill 元数据，再发送任务并接收 Message 或 Task。它不是 Deep Agent 的 task 工具，也不是 MCP；本例把一个真实 Deep Agent 包在 A2A Server 中，让另一个 A2

全部笔记查看源文档

A2A 是 Agent-to-Agent 协议：调用方先读取远端的 Agent Card，了解它的接口和 Skill 元数据，再发送任务并接收 Message 或 Task。它不是 Deep Agent 的 task 工具，也不是 MCP；本例把一个真实 Deep Agent 包在 A2A Server 中，让另一个 A2A Client 通过 JSON-RPC 调用它。

最小代码

代码在 deepagent_src/a2a_teach/01_deep_agent_server.py。

RemoteDeepAgentExecutor 是 A2A 与 Deep Agent 的适配层：读取 RequestContext 中的 A2A 用户文本，调用 agent.ainvoke，并把最终文本作为 Task artifact 发布。

state = await self.agent.ainvoke(
    {"messages": [HumanMessage(content=context.get_user_input())]}
)
await updater.add_artifact([new_text_part(state["messages"][-1].text)])
await updater.complete()

Executor 必须先 enqueue 一个初始 Task，之后才能发送 working、artifact 或 completed 更新。A2A SDK 用这个 Task 保存状态、历史与 artifact；直接先发 status 会被协议拒绝。

本地真实调用

uv run python -m deepagent_src.a2a_teach.01_deep_agent_server

这会发起一次真实 gpt-5.5 调用，但 A2A HTTP 通路使用 httpx.ASGITransport 在进程内执行：没有开放端口，也没有调用外部 A2A 服务。

客户端仍执行完整协议流程：

从 /.well-known/agent-card.json 读取 Agent Card。
根据 Card 选择 JSON-RPC transport。
发送 SendMessageRequest。
Server 创建 Task，调用 Deep Agent，并发布 artifact 与 completed 状态。
Client 收到最终 Task，读取 artifact。

预期现象

task_state: TASK_STATE_COMPLETED
artifact: A2A_REMOTE_OK: inv-1001
a2a deep agent real call ok

常见误区

AgentSkill 只是在 Agent Card 中声明远端能力，不能自动把本地 Skill 文件、工具、MCP 连接或权限同步给对方。远端服务仍要独立进行认证、授权、输入验证、超时与审计。

这个示例禁用 A2A streaming，目的是聚焦最小 Task 生命周期。长任务要在后续章节使用 streaming status/artifact updates、任务取消、input_required / auth_required 和持久化 TaskStore。

官方依据：A2A SDK 的 AgentExecutor 要求 Executor 发布 Message 或 Task，并在长任务中持续发布状态和 artifact 事件；ClientFactory.create_from_url 会先解析 Agent Card，再选择兼容 transport。

相关资源

查看示例代码：deepagent_src/a2a_teach/01_deep_agent_server.py

from __future__ import annotations

import asyncio
import os

import httpx
from a2a.client import ClientConfig, create_client
from a2a.helpers.proto_helpers import (
    get_text_parts,
    new_text_message,
    new_text_part,
)
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.routes.agent_card_routes import create_agent_card_routes
from a2a.server.routes.fastapi_routes import add_a2a_routes_to_fastapi
from a2a.server.routes.jsonrpc_routes import create_jsonrpc_routes
from a2a.server.tasks import InMemoryTaskStore, TaskUpdater
from a2a.types import (
    AgentCapabilities,
    AgentCard,
    AgentInterface,
    AgentSkill,
    Role,
    SendMessageConfiguration,
    SendMessageRequest,
    Task,
    TaskState,
    TaskStatus,
)
from a2a.utils.constants import TransportProtocol
from deepagents import create_deep_agent
from fastapi import FastAPI
from langchain.messages import HumanMessage

from deepagent_src.llms import get_gpt_model

AGENT_URL = "http://a2a.local"


class RemoteDeepAgentExecutor(AgentExecutor):
    def __init__(self) -> None:
        self.agent = create_deep_agent(
            model=get_gpt_model(disable_tool_streaming=True),
            subagents=[],
            system_prompt=(
                "你是通过 A2A 暴露的远程 Deep Agent。"
                "用户会提供一个发票 ID。只回复 `A2A_REMOTE_OK: <发票ID>`。"
            ),
        )

    async def execute(self, context: RequestContext, event_queue) -> None:
        await event_queue.enqueue_event(
            Task(
                id=context.task_id,
                context_id=context.context_id,
                status=TaskStatus(state=TaskState.TASK_STATE_SUBMITTED),
            )
        )
        updater = TaskUpdater(event_queue, context.task_id, context.context_id)
        await updater.start_work()

        state = await self.agent.ainvoke(
            {"messages": [HumanMessage(content=context.get_user_input())]}
        )
        answer = state["messages"][-1].text
        await updater.add_artifact([new_text_part(answer)], name="deep-agent-answer")
        await updater.complete()

    async def cancel(self, context: RequestContext, event_queue) -> None:
        updater = TaskUpdater(event_queue, context.task_id, context.context_id)
        await updater.cancel()


def build_a2a_app() -> FastAPI:
    card = AgentCard(
        name="invoice-a2a-deep-agent",
        version="0.1.0",
        description="A Deep Agent exposed through the A2A protocol.",
        supported_interfaces=[
            AgentInterface(
                protocol_binding=TransportProtocol.JSONRPC,
                url=f"{AGENT_URL}/",
            )
        ],
        capabilities=AgentCapabilities(streaming=False),
        default_input_modes=["text/plain"],
        default_output_modes=["text/plain"],
        skills=[
            AgentSkill(
                id="invoice-id-echo",
                name="Invoice ID response",
                description="Returns the invoice ID supplied by the caller.",
                tags=["invoice", "teaching"],
            )
        ],
    )
    handler = DefaultRequestHandler(
        RemoteDeepAgentExecutor(),
        InMemoryTaskStore(),
        card,
    )
    app = FastAPI()
    add_a2a_routes_to_fastapi(
        app,
        agent_card_routes=create_agent_card_routes(card),
        jsonrpc_routes=create_jsonrpc_routes(handler, rpc_url="/"),
    )
    return app


async def main() -> None:
    os.environ["LANGSMITH_TRACING"] = "false"
    os.environ["LANGCHAIN_TRACING_V2"] = "false"

    app = build_a2a_app()
    transport = httpx.ASGITransport(app=app)
    async with httpx.AsyncClient(
        transport=transport,
        base_url=AGENT_URL,
    ) as http_client:
        client = await create_client(
            AGENT_URL,
            ClientConfig(streaming=False, httpx_client=http_client),
        )
        request = SendMessageRequest(
            message=new_text_message(
                "请处理发票 ID：inv-1001。",
                role=Role.ROLE_USER,
            ),
            configuration=SendMessageConfiguration(),
        )
        responses = [response async for response in client.send_message(request)]
        await client.close()

    task = responses[-1].task
    answer = "\n".join(get_text_parts(task.artifacts[-1].parts))
    print("task_state:", TaskState.Name(task.status.state))
    print("artifact:", answer)
    assert task.status.state == TaskState.TASK_STATE_COMPLETED
    assert answer == "A2A_REMOTE_OK: inv-1001", answer
    print("a2a deep agent real call ok")


if __name__ == "__main__":
    asyncio.run(main())