跳转至

tRPC-Agent-Go A2A 集成指南

本文介绍面向 A2A v0.2.x 的旧版 server/a2a 与 agent/a2aagent 包。新接入和迁移说明请参考 A2A v1.0 接入与迁移指南。

概述

tRPC-Agent-Go 提供了完整的 A2A (Agent-to-Agent) 解决方案，包含两个核心组件：

  • A2A Server: 将本地 Agent 暴露为 A2A 服务，供其他 Agent 调用
  • A2A Agent: 调用远程 A2A 服务的客户端代理，像使用本地 Agent 一样使用远程 Agent

核心能力

  • 零协议感知: 开发者只需关注 Agent 的业务逻辑，无需了解 A2A 协议细节
  • 自动适配: 框架自动将 Agent 信息转换为 A2A AgentCard
  • 消息转换: 自动处理 A2A 协议消息与 Agent 消息格式的转换

A2A Server：暴露 Agent 为服务

概念介绍

A2A Server 是 tRPC-Agent-Go 提供的服务端组件，用于将任何本地 Agent 快速转换为符合 A2A 协议的网络服务。

核心特性

  • 一键转换: 通过简单配置将 Agent 暴露为 A2A 服务
  • 自动协议适配: 自动处理 A2A 协议与 Agent 接口的转换
  • AgentCard 生成: 自动生成服务发现所需的 AgentCard
  • 流式支持: 支持流式和非流式两种响应模式

Agent 到 A2A 的自动转换

tRPC-Agent-Go 通过 server/a2a 包实现了从 Agent 到 A2A 服务的无缝转换：

1
func New(opts ...Option) (*a2a.A2AServer, error) {}

AgentCard 自动生成

框架会自动提取 Agent 的元数据（名称、描述、工具等），生成符合 A2A 协议的 AgentCard，包括： - Agent 基本信息（名称、描述、URL） - 能力声明（是否支持流式） - 技能列表（基于 Agent 的工具自动生成）

这里要特别区分两层语义：

  • WithAgent(agent, streaming) 里的 streaming，本质上是用来声明生成出来的 AgentCard.Capabilities.Streaming。
  • 它不是 runner 的执行开关，也不是服务端内部消息处理链路的总开关。
  • 如果你走的是 WithRunner(...) + WithAgentCard(...) 模式，那么是否支持流式应该直接写在 AgentCard 上，例如通过 NewAgentCard(..., streaming) 来设置。
  • WithRunner(...) + WithAgentCard(...) 模式下，skills 由调用方自己维护；NewAgentCard(...) 只帮你生成默认结构和默认 skill，不会自动从自定义 runner 里推导完整工具列表。

消息协议转换

框架内置 messageProcessor 实现 A2A 协议消息与 Agent 消息格式的双向转换，用户无需关心消息格式转换的细节。

A2A Server 快速开始

使用 A2A Server 暴露 Agent 服务

只需几行代码，就可以将任意 Agent 转换为 A2A 服务：

基础示例：创建 A2A Server

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
package main

import (
    "trpc.group/trpc-go/trpc-agent-go/agent/llmagent"
    "trpc.group/trpc-go/trpc-agent-go/model/openai"
    a2aserver "trpc.group/trpc-go/trpc-agent-go/server/a2a"
)

func main() {
    // 1. 创建一个普通的 Agent
    model := openai.New("gpt-4o-mini")
    agent := llmagent.New("MyAgent",
        llmagent.WithModel(model),
        llmagent.WithDescription("一个智能助手"),
    )

    // 2. 一键转换为 A2A 服务
    server, _ := a2aserver.New(
        a2aserver.WithHost("localhost:8080"),
        a2aserver.WithAgent(agent, true), // 开启 streaming
    )

    // 3. 启动服务，即可接受 A2A 请求
    server.Start(":8080")
}

流式输出事件类型（Message vs Artifact）

当开启 streaming 时，A2A 允许服务端用不同的方式下发增量输出：

  • TaskArtifactUpdateEvent（默认）：ADK 风格。把增量内容作为任务的 artifact 更新事件（artifact-update）下发。
  • Message：轻量模式。把增量内容作为 message 下发，客户端可以直接渲染 Message.parts，无需把输出当作“可持久化 artifact”来处理。

如果你的业务更希望把流式内容直接当作 message 来消费，可以这样配置：

1
2
3
4
5
6
7
server, _ := a2aserver.New(
    a2aserver.WithHost("localhost:8080"),
    a2aserver.WithAgent(agent, true),
    a2aserver.WithStreamingEventType(
        a2aserver.StreamingEventTypeMessage,
    ),
)

任务状态更新（submitted、completed）仍然会以 TaskStatusUpdateEvent 的形式 下发。如果开启 WithStructuredTaskErrors(true)，终态失败也会通过 failed TaskStatusUpdateEvent 下发：机器可读字段优先位于外层 metadata，并为了 0.1 兼容继续镜像到 status.message.metadata，展示文本位于 status.message.parts。

直接使用 A2A 协议客户端调用

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
import (
    "context"
    "net/http"
    "net/http/cookiejar"

    "trpc.group/trpc-go/trpc-agent-go/agent/a2aagent"
    "trpc.group/trpc-go/trpc-a2a-go/client"
    "trpc.group/trpc-go/trpc-a2a-go/protocol"
)

func main() {
    jar, _ := cookiejar.New(nil)
    httpClient := &http.Client{Jar: jar}
    // 该 helper 会持久化匿名 Cookie，并在 principal 建立前串行化同一 client 的首次并发请求。
    a2aClient, _ := a2aagent.NewAnonymousA2AClient(
        "http://localhost:8080/",
        client.WithHTTPClient(httpClient),
    )

    // 发送消息给 Agent
    message := protocol.NewMessage(
        protocol.MessageRoleUser,
        []protocol.Part{protocol.NewTextPart("你好，请帮我分析这段代码")},
    )

    // Agent 会自动处理并返回结果
    response, _ := a2aClient.SendMessage(context.Background(),
        protocol.SendMessageParams{Message: message})
}

匿名直连 client 应复用同一个 client 和 Cookie Jar。NewAnonymousA2AClient 会在服务端建立匿名 principal 期间，串行化该 client 发出的首次请求。 这个保证只覆盖单个 client 实例；不同 client 需要自行协调。匿名直连 client 会在 Cookie Jar 中遵循 anonymous Set-Cookie 的 Path、Domain、过期和删除指令。 它不会接入 SessionService，也不负责持久化 session state。 应先完成一次初始请求，再开始并发匿名消息发送，或者提供可信用户身份。

NewAnonymousA2AClient 仅在请求期间从响应或自定义 HTTP handler 观察到有效匿名 Cookie，且配置的 Cookie Jar 接受同一值并可用于 agent URL 后，才解除初始化 gate。预加载 Cookie 会随请求发送，但本身不能确认初始化完成；自定义服务端必须通过 Set-Cookie 回写已接受的 Cookie，或返回替代值，后续请求才能并发执行。内置 A2A Server 已具备该行为。

浏览器匿名身份连续性要求同站部署。服务端 Cookie 使用 SameSite=Lax，因此浏览器不会在跨站 JSON-RPC POST 请求中携带该 Cookie。当前端页面与 A2A 接口同站但跨源时，Fetch 请求必须设置 credentials: "include"，部署层还必须为该前端的明确 Origin 返回 Access-Control-Allow-Origin，并设置 Access-Control-Allow-Credentials: true。跨站部署应改为提供可信用户身份；当前不支持跨站匿名 Cookie 连续性。

匿名 Principal 行为

当请求没有通过配置的 UserID header（默认是 X-User-ID）提供非空身份时， 内置 A2A Server 认证会生成带有 A2A_ANONYMOUS_ 前缀的随机 principal。 服务端通过 HTTP-only Cookie trpc_agent_a2a_anon 返回该 principal，并利用 这个 Cookie 维持后续请求的匿名身份连续性。A2A contextID 仍然用于标识 session，不再作为 principal 的来源。

需要保持连续性的客户端必须复用 Cookie Jar；不保留 Cookie 的客户端或独立请求， 每次都可能获得新的匿名 principal。请求携带非空 UserID header 时，服务端使用 header 中的身份，不走匿名 Cookie 流程。

高级配置

自定义 Runner（WithRunner）

默认情况下，A2A Server 会自动为你创建一个 Runner。如果你需要更精细地控制，例如注入 MemoryService、自定义 SessionService，可以使用 WithRunner。

注意：WithRunner 与 WithAgent 互斥。使用 WithRunner 时，需要显式通过 WithAgentCard 提供对外暴露的 Agent 身份：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
import (
    "trpc.group/trpc-go/trpc-agent-go/memory/inmemory"
    "trpc.group/trpc-go/trpc-agent-go/runner"
    sessionmemory "trpc.group/trpc-go/trpc-agent-go/session/inmemory"
    "trpc.group/trpc-go/trpc-agent-go/server/a2a"
)

memoryService := inmemory.NewMemoryService()
sessionService := sessionmemory.NewSessionService()
streaming := true

r := runner.NewRunner(
    agent.Info().Name,
    agent,
    runner.WithSessionService(sessionService),
    runner.WithMemoryService(memoryService),
)

card, _ := a2a.NewAgentCard(agent.Info().Name, agent.Info().Description, "localhost:8080", streaming)

server, _ := a2a.New(
    a2a.WithRunner(r),
    a2a.WithAgentCard(card),
)

在这种 runner-only 模式下，流式能力不再通过 WithAgent(...) 传入，而是直接由 WithAgentCard(...) 提供的 AgentCard.Capabilities.Streaming 决定。

examples/a2aagent 里也提供了显式示例。使用 -server-mode runner-card 可以切换到 WithRunner(...) + WithAgentCard(...) 的建服方式：

1
2
cd examples/a2aagent
go run . -server-mode runner-card

如果你只想快速得到一张符合默认约定的基础卡片，推荐直接使用 NewAgentCard(...)，而不是手写 Name、Description、Capabilities 和默认 Skill。如果你的 runner 需要暴露更准确的 skills，请在业务侧自己补齐和维护。

动态更新 AgentCard

如果你希望在运行时热更新对外暴露的 AgentCard，可以直接通过 WithExtraA2AOptions(...) 接入底层 a2a.WithAgentCardHandler(...)，并配合 NewAgentCardHandler(...) 输出当前快照：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
import (
    "sync"

    a2aprotocolserver "trpc.group/trpc-go/trpc-a2a-go/server"
    "trpc.group/trpc-go/trpc-agent-go/runner"
    "trpc.group/trpc-go/trpc-agent-go/server/a2a"
)

card, _ := a2a.NewAgentCard(agent.Info().Name, agent.Info().Description, "localhost:8080", true)
var (
    cardMu      sync.RWMutex
    currentCard = card
)

server, _ := a2a.New(
    a2a.WithRunner(runner.NewRunner(agent.Info().Name, agent)),
    a2a.WithAgentCard(currentCard),
    a2a.WithExtraA2AOptions(
        a2aprotocolserver.WithAgentCardHandler(
            a2a.NewAgentCardHandler(func() a2aprotocolserver.AgentCard {
                cardMu.RLock()
                defer cardMu.RUnlock()
                return currentCard
            }),
        ),
    ),
)

cardMu.Lock()
updated := currentCard
updated.Description = "new description"
currentCard = updated
cardMu.Unlock()

这种方式只会更新对外暴露的 metadata，不会影响底层 runner、taskManager 和消息处理链路。至于 currentCard 存在内存、配置中心还是数据库里，由业务自己决定。

同时建议把 Name、URL 这类启动期就参与路由、身份或发现语义的字段视为不可变字段；如果确实要改，优先重建 server，而不是只热更新 card endpoint。

服务端消息处理 Hook（WithProcessMessageHook）

WithProcessMessageHook 允许你在 A2A Server 处理消息之前/之后插入自定义逻辑。它采用中间件模式，包装底层的 MessageProcessor：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
import "trpc.group/trpc-go/trpc-a2a-go/taskmanager"

// 自定义 Hook 处理器
type hookProcessor struct {
    next taskmanager.MessageProcessor
}

func (h *hookProcessor) ProcessMessage(
    ctx context.Context,
    message protocol.Message,
    options taskmanager.ProcessOptions,
    handler taskmanager.TaskHandler,
) (*taskmanager.MessageProcessingResult, error) {
    // 在处理之前：读取客户端注入的自定义 metadata
    if traceID, ok := message.Metadata["trace_id"]; ok {
        fmt.Printf("received trace_id: %v\n", traceID)
    }
    // 委托给下一个处理器
    return h.next.ProcessMessage(ctx, message, options, handler)
}

server, _ := a2a.New(
    a2a.WithHost("localhost:8080"),
    a2a.WithAgent(agent, true),
    a2a.WithProcessMessageHook(
        func(next taskmanager.MessageProcessor) taskmanager.MessageProcessor {
            return &hookProcessor{next: next}
        },
    ),
)

典型使用场景： - 读取客户端通过 BuildMessageHook 注入的自定义 metadata - 在消息处理前后添加日志、监控、审计 - 修改或验证入站消息

客户端消息构建 Hook（WithBuildMessageHook）

WithBuildMessageHook 是 A2AAgent（客户端）侧的 Hook，允许在将消息发送到远程 A2A Server 之前注入自定义数据。它同样采用中间件模式：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
import "trpc.group/trpc-go/trpc-agent-go/agent/a2aagent"

a2aAgent, _ := a2aagent.New(
    a2aagent.WithAgentCardURL("http://remote-agent:8888"),
    a2aagent.WithBuildMessageHook(
        func(next a2aagent.ConvertToA2AMessageFunc) a2aagent.ConvertToA2AMessageFunc {
            return func(isStream bool, agentName string, inv *agent.Invocation) (*protocol.Message, error) {
                // 调用默认转换器
                msg, err := next(isStream, agentName, inv)
                if err != nil {
                    return nil, err
                }
                // 注入自定义 metadata
                if msg.Metadata == nil {
                    msg.Metadata = make(map[string]any)
                }
                msg.Metadata["trace_id"] = "my-trace-123"
                msg.Metadata["business_tag"] = "order-service"
                return msg, nil
            }
        },
    ),
)

BuildMessageHook + ProcessMessageHook 联动：

1
2
3
4
5
6
┌──────────────────┐                    ┌───────────────────┐
│    A2AAgent      │   A2A protocol     │    A2A Server     │
│                  │                    │                   │
│ BuildMessageHook │── metadata ──────→ │ProcessMessageHook │
│ (inject data)    │                    │ (read data)       │
└──────────────────┘                    └───────────────────┘

客户端通过 BuildMessageHook 将自定义数据（如 trace_id、业务标签）注入到 A2A 消息的 metadata 字段中，服务端通过 ProcessMessageHook 读取并处理这些数据。

追加 RunOption（WithRunOptions）

WithRunOptions 允许为 A2A Server 的每次 Agent 调用追加额外的 RunOption：

1
2
3
4
5
6
7
server, _ := a2a.New(
    a2a.WithHost("localhost:8080"),
    a2a.WithAgent(agent, true),
    a2a.WithRunOptions(
        agent.WithRequestID("custom-req-id"),
    ),
)

Graph 内部事件透传

从默认行为看，A2A Server 会过滤大部分 graph.* 运行时内部事件（例如 graph.node.start、graph.node.complete、graph.pregel.*、graph.checkpoint.*）， 避免把执行细节全部暴露给下游。

默认仍会保留终态 graph.execution（以及普通消息/错误事件），用于恢复最终状态与 state_delta。

如果你需要做链路调试或节点级 trace，可以扩展 graph 事件白名单：

1
2
3
4
5
6
7
8
server, _ := a2aserver.New(
    a2aserver.WithHost("localhost:8080"),
    a2aserver.WithAgent(agent, true),
    a2aserver.WithGraphEventObjectAllowlist(
        "graph.execution", // 保留终态事件
        "graph.node.*",    // 透传节点生命周期事件
    ),
)

说明：

  • 不设置该选项时，默认白名单是 ["graph.execution"]。
  • 如果显式调用 WithGraphEventObjectAllowlist() 且不传参数， 那么所有 graph.* 事件都会被过滤（包括 graph.execution）。

建议仅在调试场景开启，生产环境默认关闭可以减少噪音与带宽开销。

改写出站 A2A 响应

WithResponseRewriter 允许服务端在 A2A 结果返回给调用方、或发送给流式订阅者之前做最后一次改写。 典型用途包括隐藏内部 metadata、删除调试字段，或丢弃不希望暴露到公开 A2A 协议面的服务端事件。

对于 unary 响应，rewriter 看到的是服务端聚合后的最终结果。对于 streaming 响应，rewriter 会看到每一个即将发送的出站结果，包括转换后的 agent 事件、task 状态更新、final artifact、结构化 task error，以及 ErrorHandler 返回的消息。每次 rewrite 都会收到本次请求的 context，方便读取请求级日志字段。

rewriter 返回 nil 表示丢弃该出站结果。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
import (
    "context"

    "trpc.group/trpc-go/trpc-a2a-go/protocol"
    a2aserver "trpc.group/trpc-go/trpc-agent-go/server/a2a"
)

server, _ := a2aserver.New(
    a2aserver.WithHost("localhost:8080"),
    a2aserver.WithAgent(agent, true),
    a2aserver.WithResponseRewriter(a2aserver.ResponseRewriterFuncs{
        Unary: func(ctx context.Context, result protocol.UnaryMessageResult) protocol.UnaryMessageResult {
            if msg, ok := result.(*protocol.Message); ok {
                delete(msg.Metadata, "debug_trace")
            }
            return result
        },
        Streaming: func(ctx context.Context, result protocol.StreamingMessageResult) protocol.StreamingMessageResult {
            if msg, ok := result.(*protocol.Message); ok {
                delete(msg.Metadata, "debug_trace")
            }
            return result
        },
    }),
)

在一个端口上暴露多个 A2A Agent（base path）

有时你希望 一个服务（一个端口） 同时对外提供多个 A2A Agent。 在 A2A 的惯用做法里，每个 Agent 都应该有自己的 base URL，客户端通过选择 URL 来选择要调用的 Agent，而不是额外传一个 agent_name 参数做路由。

在 tRPC-Agent-Go 中，a2a.WithHost(...) 支持带 path 的 URL。 当 host URL 包含 path（例如 http://localhost:8888/agents/math）时， A2A server 会自动把这段 path 作为自己的 base path 用于路由。

核心思路：

  • 每个 Agent 创建一个独立的 A2A server（每个 server 使用不同的 base path）
  • 不要对每个 server 都调用 Start（否则会争抢同一个端口）
  • 通过 server.Handler() 把多个 A2A server 挂到一个共享的 http.Server 上

示例：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
mathServer, err := a2a.New(
    a2a.WithHost("http://localhost:8888/agents/math"),
    a2a.WithAgent(mathAgent, false),
)
if err != nil {
    panic(err)
}

weatherServer, err := a2a.New(
    a2a.WithHost("http://localhost:8888/agents/weather"),
    a2a.WithAgent(weatherAgent, false),
)
if err != nil {
    panic(err)
}

mux := http.NewServeMux()
mux.Handle("/agents/math/", mathServer.Handler())
mux.Handle("/agents/weather/", weatherServer.Handler())

if err := http.ListenAndServe(":8888", mux); err != nil {
    panic(err)
}

服务启动后，每个 Agent 都有自己独立的 AgentCard：

  • http://localhost:8888/agents/math/.well-known/agent-card.json
  • http://localhost:8888/agents/weather/.well-known/agent-card.json

完整可运行示例：examples/a2amultipath。

A2AAgent：调用远程 A2A 服务

与 A2A Server 相对应，tRPC-Agent-Go 还提供了 A2AAgent，用于调用远程的 A2A 服务，实现 Agent 间的通信。

概念介绍

A2AAgent 是一个特殊的 Agent 实现，它不直接处理用户请求，而是将请求转发给远程的 A2A 服务。从使用者角度看，A2AAgent 就像一个普通的 Agent，但实际上它是远程 Agent 的本地代理。

简单理解： - A2A Server: 我有一个 Agent，想让别人调用 → 暴露为 A2A 服务 - A2AAgent: 我想调用别人的 Agent → 通过 A2AAgent 代理调用

核心特性

  • 透明代理: 像使用本地 Agent 一样使用远程 Agent
  • 自动发现: 通过 AgentCard 自动发现远程 Agent 的能力
  • 协议转换: 自动处理本地消息格式与 A2A 协议的转换
  • 流式支持: 支持流式和非流式两种通信模式
  • 状态传递: 支持将本地状态传递给远程 Agent
  • 错误处理: 完善的错误处理和重试机制

如果你想看 A2A server 和 A2AAgent 之间推荐的结构化 task error 约定，见 Error Handling。

使用场景

  1. 分布式 Agent 系统: 在微服务架构中调用其他服务的 Agent
  2. Agent 编排: 将多个专业 Agent 组合成复杂的工作流
  3. 跨团队协作: 调用其他团队提供的 Agent 服务

A2AAgent 快速开始

基本用法

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
package main

import (
    "context"
    "fmt"

    "trpc.group/trpc-go/trpc-agent-go/agent/a2aagent"
    "trpc.group/trpc-go/trpc-agent-go/runner"
    "trpc.group/trpc-go/trpc-agent-go/model"
    "trpc.group/trpc-go/trpc-agent-go/session/inmemory"
)

func main() {
    // 1. 创建 A2AAgent，指向远程 A2A 服务
    a2aAgent, err := a2aagent.New(
        a2aagent.WithAgentCardURL("http://localhost:8888"),
    )
    if err != nil {
        panic(err)
    }

    // 2. 像使用普通 Agent 一样使用
    sessionService := inmemory.NewSessionService()
    runner := runner.NewRunner("test", a2aAgent, 
        runner.WithSessionService(sessionService))

    // 3. 发送消息
    events, err := runner.Run(
        context.Background(),
        "user1",
        "session1", 
        model.NewUserMessage("请帮我讲个笑话"),
    )
    if err != nil {
        panic(err)
    }

    // 4. 处理响应
    for event := range events {
        if event.Response != nil && len(event.Response.Choices) > 0 {
            fmt.Print(event.Response.Choices[0].Message.Content)
        }
    }
}

在多 Agent 系统中，A2AAgent 通常作为本地协调者 Agent （例如 LLMAgent）的 SubAgent 使用。你可以将 A2AAgent 与 LLMAgent.SetSubAgents 结合，从注册中心动态加载并更新 远程 SubAgent，而无需重建协调者实例。

高级配置

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
// 创建带有高级配置的 A2AAgent
a2aAgent, err := a2aagent.New(
    // 指定远程服务地址
    a2aagent.WithAgentCardURL("http://remote-agent:8888"),

    // 设置流式缓冲区大小
    a2aagent.WithStreamingChannelBufSize(2048),

    // 自定义协议转换
    a2aagent.WithCustomEventConverter(customEventConverter),
    a2aagent.WithCustomA2AConverter(customA2AConverter),

    // 显式控制流式模式（覆盖 AgentCard 中的 capability 声明）
    a2aagent.WithEnableStreaming(true),
)

客户端是否发起流式请求，遵循下面的优先级：

  1. 单次调用显式指定的 agent.WithStream(...)
  2. a2aagent.WithEnableStreaming(...)
  3. 远端 AgentCard.Capabilities.Streaming
  4. 默认关闭

也就是说，服务端这边的 streaming 声明主要是告诉客户端“我是否支持流式 A2A 请求”；客户端再基于这份 capability 决定默认发送流式还是非流式请求。
如果客户端显式指定了 agent.WithStream(...) 或 a2aagent.WithEnableStreaming(...)，就会覆盖 AgentCard 中的声明。

完整示例：A2A Server + A2AAgent 综合使用

以下是一个完整的示例，展示了如何在同一个程序中同时运行 A2A Server（暴露本地 Agent）和 A2AAgent（调用远程服务）：

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
package main

import (
    "context"
    "fmt"
    "time"

    "trpc.group/trpc-go/trpc-agent-go/agent/a2aagent"
    "trpc.group/trpc-go/trpc-agent-go/agent/llmagent"
    "trpc.group/trpc-go/trpc-agent-go/model/openai"
    "trpc.group/trpc-go/trpc-agent-go/runner"
    "trpc.group/trpc-go/trpc-agent-go/server/a2a"
    "trpc.group/trpc-go/trpc-agent-go/session/inmemory"
)

func main() {
    // 1. 创建并启动远程 Agent 服务
    remoteAgent := createRemoteAgent()
    startA2AServer(remoteAgent, "localhost:8888")

    time.Sleep(1 * time.Second) // 等待服务启动

    // 2. 创建 A2AAgent 连接到远程服务
    a2aAgent, err := a2aagent.New(
        a2aagent.WithAgentCardURL("http://localhost:8888"),
        a2aagent.WithTransferStateKey("user_context"),
    )
    if err != nil {
        panic(err)
    }

    // 3. 创建本地 Agent
    localAgent := createLocalAgent()

    // 4. 对比本地和远程 Agent 的响应
    compareAgents(localAgent, a2aAgent)
}

func createRemoteAgent() agent.Agent {
    model := openai.New("gpt-4o-mini")
    return llmagent.New("JokeAgent",
        llmagent.WithModel(model),
        llmagent.WithDescription("I am a joke-telling agent"),
        llmagent.WithInstruction("Always respond with a funny joke"),
    )
}

func createLocalAgent() agent.Agent {
    model := openai.New("gpt-4o-mini") 
    return llmagent.New("LocalAgent",
        llmagent.WithModel(model),
        llmagent.WithDescription("I am a local assistant"),
    )
}

func startA2AServer(agent agent.Agent, host string) {
    server, err := a2a.New(
        a2a.WithHost(host),
        a2a.WithAgent(agent, true), // 启用流式
    )
    if err != nil {
        panic(err)
    }

    go func() {
        server.Start(host)
    }()
}

func compareAgents(localAgent, remoteAgent agent.Agent) {
    sessionService := inmemory.NewSessionService()

    localRunner := runner.NewRunner("local", localAgent,
        runner.WithSessionService(sessionService))
    remoteRunner := runner.NewRunner("remote", remoteAgent,
        runner.WithSessionService(sessionService))

    userMessage := "请帮我讲个笑话"

    // 调用本地 Agent
    fmt.Println("=== Local Agent Response ===")
    processAgent(localRunner, userMessage)

    // 调用远程 Agent (通过 A2AAgent)
    fmt.Println("\n=== Remote Agent Response (via A2AAgent) ===")
    processAgent(remoteRunner, userMessage)
}

func processAgent(runner runner.Runner, message string) {
    events, err := runner.Run(
        context.Background(),
        "user1",
        "session1",
        model.NewUserMessage(message),
        agent.WithRuntimeState(map[string]any{
            "user_context": "test_context",
        }),
    )
    if err != nil {
        fmt.Printf("Error: %v\n", err)
        return
    }

    for event := range events {
        if event.Response != nil && len(event.Response.Choices) > 0 {
            content := event.Response.Choices[0].Message.Content
            if content == "" {
                content = event.Response.Choices[0].Delta.Content
            }
            if content != "" {
                fmt.Print(content)
            }
        }
    }
    fmt.Println()
}

AgentCard 自动发现

A2AAgent 支持通过标准的 AgentCard 发现机制自动获取远程 Agent 的信息：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
// A2AAgent 会自动从以下路径获取 AgentCard
// http://remote-agent:8888/.well-known/agent.json

type AgentCard struct {
    Name         string                 `json:"name"`
    Description  string                 `json:"description"`
    URL          string                 `json:"url"`
    Capabilities AgentCardCapabilities  `json:"capabilities"`
}

type AgentCardCapabilities struct {
    Streaming *bool `json:"streaming,omitempty"`
}

状态传递

A2AAgent 支持将本地运行时状态传递给远程 Agent：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
a2aAgent, _ := a2aagent.New(
    a2aagent.WithAgentCardURL("http://remote-agent:8888"),
    // 指定要传递的状态键
    a2aagent.WithTransferStateKey("user_id", "session_context", "preferences"),
)

// 运行时状态会通过 A2A 协议的 metadata 字段传递给远程 Agent
events, _ := runner.Run(ctx, userID, sessionID, message,
    agent.WithRuntimeState(map[string]any{
        "user_id":         "12345",
        "session_context": "shopping_cart",
        "preferences":     map[string]string{"language": "zh"},
    }),
)

自定义 HTTP Headers

你可以使用 WithA2ARequestOptions 为每次请求传递自定义的 HTTP headers：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
import "trpc.group/trpc-go/trpc-a2a-go/client"

events, err := runner.Run(
    context.Background(),
    userID,
    sessionID,
    model.NewUserMessage("你的问题"),
    // 为本次请求传递自定义 HTTP headers
    agent.WithA2ARequestOptions(
        client.WithRequestHeader("X-Custom-Header", "custom-value"),
        client.WithRequestHeader("X-Request-ID", fmt.Sprintf("req-%d", time.Now().UnixNano())),
        client.WithRequestHeader("Authorization", "Bearer your-token"),
    ),
)

常见使用场景：

  1. 身份认证：传递认证 token 

    1
2
3
    agent.WithA2ARequestOptions(
    client.WithRequestHeader("Authorization", "Bearer "+token),
)

  2. 分布式追踪：添加请求 ID（OpenTelemetry trace context 会自动通过 HTTP header 传播） 

    1
2
3
    agent.WithA2ARequestOptions(
    client.WithRequestHeader("X-Request-ID", requestID),
)

配置 UserID Header：

客户端和服务端都支持配置使用哪个 HTTP header 来传递 UserID，默认使用 X-User-ID：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
// 客户端：配置通过哪个 header 发送 UserID
a2aAgent, _ := a2aagent.New(
    a2aagent.WithAgentCardURL("http://remote-agent:8888"),
    // 默认是 "X-User-ID"，可以自定义
    a2aagent.WithUserIDHeader("X-Custom-User-ID"),
)

// 服务端：配置从哪个 header 读取 UserID
server, _ := a2a.New(
    a2a.WithHost("localhost:8888"),
    a2a.WithAgent(agent, true),
    // 默认是 "X-User-ID"，可以自定义
    a2a.WithUserIDHeader("X-Custom-User-ID"),
)

来自 invocation.Session.UserID 的 UserID 会自动通过配置的 header 发送给 A2A server。

ADK 兼容模式

如果需要与 Google ADK (Agent Development Kit) Python 客户端互通，可以启用 ADK 兼容模式。 启用后，Server 会在 metadata 中额外写入 adk_ 前缀的 key（如 adk_type、adk_thought）， 以兼容 ADK 的 part converter 解析逻辑：

1
2
3
4
5
server, _ := a2a.New(
    a2a.WithHost("localhost:8888"),
    a2a.WithAgent(agent, true),
    a2a.WithADKCompatibility(true), // 默认开启
)

自定义转换器

对于特殊需求，可以自定义消息和事件转换器：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
// 自定义 A2A 消息转换器（Invocation → A2A Message）
// 实现 a2aagent.InvocationA2AConverter 接口
type CustomA2AConverter struct{}

func (c *CustomA2AConverter) ConvertToA2AMessage(
    isStream bool, 
    agentName string, 
    invocation *agent.Invocation,
) (*protocol.Message, error) {
    // 自定义消息转换逻辑
    msg := protocol.NewMessage(protocol.MessageRoleUser, []protocol.Part{
        protocol.NewTextPart(invocation.Message.Content),
    })
    return &msg, nil
}

// 自定义事件转换器（A2A Response → Event）
// 实现 a2aagent.A2AEventConverter 接口
type CustomEventConverter struct{}

func (c *CustomEventConverter) ConvertToEvents(
    result protocol.MessageResult,
    agentName string,
    invocation *agent.Invocation,
) ([]*event.Event, error) {
    // 自定义非流式事件转换逻辑
    return []*event.Event{event.New(invocation.InvocationID, agentName)}, nil
}

func (c *CustomEventConverter) ConvertStreamingToEvents(
    result protocol.StreamingMessageEvent,
    agentName string,
    invocation *agent.Invocation,
) ([]*event.Event, error) {
    // 自定义流式事件转换逻辑
    return []*event.Event{event.New(invocation.InvocationID, agentName)}, nil
}

// 使用自定义转换器
a2aAgent, _ := a2aagent.New(
    a2aagent.WithAgentCardURL("http://remote-agent:8888"),
    a2aagent.WithCustomA2AConverter(&CustomA2AConverter{}),
    a2aagent.WithCustomEventConverter(&CustomEventConverter{}),
)

通过 A2A 传递 Graph 中断与恢复

当远程 A2A Server 运行 Graph Agent 并使用了 graph.Interrupt 时，需要额外配置：

  • Server 端：使用 WithGraphEventObjectAllowlist("*") 放行所有 graph 内部事件（含中断）
  • Client 端：使用 WithTransferStateKey("*") 将 RuntimeState 传递到 A2A metadata，确保 checkpoint/lineage 等恢复信息能传回
 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
// Server
server, _ := a2aserver.New(
    a2aserver.WithAgent(graphAgent, true),
    a2aserver.WithGraphEventObjectAllowlist("*"),
    a2aserver.WithStreamingEventType(a2aserver.StreamingEventTypeMessage),
)

// Client
subAgent, _ := a2aagent.New(
    a2aagent.WithAgentCardURL("http://remote:8888"),
    a2aagent.WithEnableStreaming(true),
    a2aagent.WithTransferStateKey("*"),
)

完整示例：examples/graph/a2a_interrupt

协议交互规范

关于 A2A 协议中工具调用、代码执行、思考内容等事件的传递规范，以及 Metadata 字段定义、ADK 兼容模式、分布式追踪等详细说明，请参考独立文档：

A2A 协议交互规范

该文档定义了 trpc-agent-go 在 A2A 协议之上的扩展规范，是 Client 和 Server 实现的标准参考。

总结：A2A Server vs A2AAgent

组件 职责 使用场景 核心功能
A2A Server 服务提供者 将本地 Agent 暴露给其他系统调用 • 协议转换
• AgentCard 生成
• 消息路由
• 流式支持
A2AAgent 服务消费者 调用远程 A2A 服务 • 透明代理
• 自动发现
• 状态传递
• 协议适配

典型架构模式

1
2
3
4
5
6
7
8
┌─────────────┐ A2A protocol  ┌───────────────┐
│   Client    │──────────────→│ A2A Server    |
│ (A2AAgent)  │               │ (local Agent) │
└─────────────┘               └───────────────┘
      ↑                              ↑
      │                              │
   调用远程                       暴露本地
   Agent服务                     Agent服务

通过 A2A Server 和 A2AAgent 的配合使用，可以比较方便的构建的远程的 Agent 系统。

A2A Server 常用配置项一览

配置项 说明
WithAgent(agent, streaming) 设置 Agent，并声明生成的 AgentCard 是否支持流式；与 WithRunner 互斥
WithHost(host) 设置服务地址，支持带 path 的 URL
WithAgentCard(card) 自定义 AgentCard（覆盖自动生成）
WithRunner(runner) 自定义 Runner（注入 Memory、Session 等）；需配合 WithAgentCard 使用
WithSessionService(service) 为默认 Runner 设置 SessionService
WithProcessMessageHook(hook) 服务端消息处理 Hook（中间件模式）
WithProcessorBuilder(builder) 完全自定义消息处理器
WithTaskManagerBuilder(builder) 自定义任务管理器
WithGraphEventObjectAllowlist(types...) 限制 Event 转换器允许输出的 graph object 类型
WithResponseRewriter(rewriter) 改写或丢弃出站 A2A unary/streaming 结果
WithRunOptions(opts...) 为每次调用追加 RunOption
WithStreamingEventType(type) 流式输出事件类型（Artifact/Message）
WithUserIDHeader(header) 自定义 UserID HTTP Header
WithADKCompatibility(enabled) ADK 兼容模式（默认：开启）
WithErrorHandler(handler) 自定义错误处理
WithA2AToAgentConverter(conv) 自定义 A2A→Agent 消息转换
WithEventToA2AConverter(conv) 自定义 Event→A2A 消息转换
WithExtraA2AOptions(opts...) 透传底层 A2A Server 选项；其中 middleware 可读取最终认证用户。自定义认证 provider 必须返回非空 UserID，空身份会被拒绝
WithPreAuthA2AMiddleware(middlewares...) 添加必须在匿名 Cookie 认证前执行的请求 middleware
WithDebugLogging(enabled) 开启调试日志

A2AAgent 匿名身份协调

对于使用持久化 session 的匿名 A2AAgent 调用，如果配置的 SessionService 实现了 session.StateInitializationService，agent 会自动 使用该能力。它会在共享同一后端的多个 agent 实例之间协调首次匿名 Cookie 初始化。in-memory session service 只协调共享同一个 service 实例的调用方； Redis session service 可以跨 service 实例协调。

如果后端不支持该能力，agent 会保留现有的单 agent 初始化锁和持久化行为。 如果业务必须保证跨实例协调，可以开启 fail-closed 模式：

1
2
3
4
a2aAgent, err := a2aagent.New(
    a2aagent.WithAgentCardURL("http://remote:8888"),
    a2aagent.WithRequireAnonymousIdentityCoordination(true),
)

该选项默认为 false。启用后，strict 模式在缺少稳定持久化 session key，或 SessionService 不支持协调能力时，会在联系远端 agent 前使调用失败。该能力检查 无法判断不同 service 实例是否实际共享协调状态。需要跨进程协调的部署必须使用 Redis 等共享后端，而不能使用相互独立的 in-memory service。

在支持协调的路径中，规范 .record.v1 值和 legacy 投影会一起提交，因此新版本 写入的状态仍可被变更前的旧版本读取。这种兼容性是有意设计为单向的：规范记录 存在后，旧版本只能更新或删除 legacy key，而新版本仍会把规范记录视为权威值。 因此，对于同一个可能被两个版本写入的逻辑 session，不支持混合版本滚动部署； 应先排空旧版本的写入流量并完成升级，再继续复用这些 session。

启用 Runner candidate selector 时，每个 speculative attempt 都使用隔离的 attempt-scoped session state，不会暴露 session.StateInitializationService。首次使用 匿名 Cookie 时，lenient 模式可能让每个 attempt 独立初始化并创建多个远端 principal； strict 模式则会在联系远端 agent 前让每个 attempt 失败，即使底层后端支持协调。 已经存在的有效 Cookie 不受此限制。在 candidate-aware 集成完成前，不要将 candidate selector 与首次匿名 Cookie 协调初始化组合使用；可以先预热并持久化 Cookie，或等待 后续的专门集成。

Cookie record 仍然是私有 session state，不会出现在 event state delta 中。 lease fencing 可以阻止过期 owner 覆盖新的 owner，但如果进程在远端返回后、 持久化前退出，远端 principal 仍可能成为孤儿。

这条基于 session 的路径与 NewAnonymousA2AClient 相互独立；后者没有框架级 session service，只在自身 client 和 Cookie Jar 范围内生效。

A2AAgent 完整配置项一览

配置项 说明
WithAgentCardURL(url) 远程 A2A 服务地址
WithBuildMessageHook(hook) 客户端消息构建 Hook（中间件模式）
WithTransferStateKey(keys...) 指定要传递的 RuntimeState 键
WithEnableStreaming(enabled) 显式控制流式模式
WithStreamingChannelBufSize(size) 流式缓冲区大小
WithUserIDHeader(header) 自定义 UserID HTTP Header
WithCustomA2AConverter(conv) 自定义 Invocation→A2A 消息转换
WithCustomEventConverter(conv) 自定义 A2A Response→Event 转换
WithRequireAnonymousIdentityCoordination(enabled) 要求持久化 session 的匿名身份初始化必须经过协调