Featured image of post A2A 实战：用 Go 搭一个支付诊断 MVP
AI

A2A 实战：用 Go 搭一个支付诊断 MVP

结合 a2a-go、a2aclient 和 a2asrv，拆解一个支持 Agent Card、流式事件、Task、History 和 Artifact 的 A2A 支付诊断 MVP。

A2A 实战：用 Go 搭一个支付诊断 MVP

A2A 不是单纯的聊天协议，它更像一套“任务协议”：客户端先通过 Agent Card 发现服务能力，再把用户问题作为 Task 发送给远端 Agent，最后根据 Task 状态、History 和 Artifact 把完整结果取回来。

这篇文章基于你给出的 Go 代码，整理成一个支付失败诊断 MVP。这个 Demo 的目标很明确：

  • 客户端能自动解析 Agent Card
  • 服务端能暴露 JSON-RPC 接口和 Agent Card
  • 任务执行过程中能持续推送状态
  • 结果不只返回一句话，而是带结构化 Artifact

一、MVP 框架结构

可以把这套代码拆成四层理解：

层职责关键对象
客户端发现 Agent、发送消息、接收流式事件、再次拉取 Taska2aclient.Client
服务端暴露 Agent Card 和 JSON-RPC /invokea2asrv.NewJSONRPCHandler
业务执行层把“诊断支付失败”拆成状态更新和 Artifact 输出PaymentDiagnosisExecutor
领域模型层生成诊断报告和结构化数据DiagnosisReport

一个完整请求的路径大概是这样：

1
2
3
4
5
6
7

User
  -> Client
  -> /.well-known/agent-card.json
  -> /invoke
  -> PaymentDiagnosisExecutor
  -> Task / Status / Artifact
  -> GetTask

二、A2A 在这个 MVP 里做了什么

这份 Demo 里最关键的不是“问答”，而是“任务”。

  • AgentCard 用来告诉客户端：我是谁、我支持什么协议、输出什么格式
  • SendMessage / SendStreamingMessage 用来发起任务
  • Task 用来保存执行上下文、历史消息、状态和附件
  • Artifact 用来返回结构化结果，适合放 markdown 和 JSON
  • GetTask 用来在流式完成后再取回完整任务

也就是说，A2A 把“远端智能能力”从一次性回答，变成了可追踪、可回放、可扩展的业务任务。

三、服务端源码：Agent Card 和 JSON-RPC

服务端的入口很短，核心就是两个动作：挂载 Agent Card，挂载 /invoke。

 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

package main

import (
	"flag"
	"fmt"
	"log"
	"net"
	"net/http"

	"example.com/a2a-go-business-demo/internal/incident"
	"github.com/a2aproject/a2a-go/v2/a2a"
	"github.com/a2aproject/a2a-go/v2/a2asrv"
)

var port = flag.Int("port", 9001, "A2A JSON-RPC server port.")

func main() {
	flag.Parse()

	invokeURL := fmt.Sprintf("http://127.0.0.1:%d/invoke", *port)

	card := &a2a.AgentCard{
		Name:        "Payment Diagnosis Agent",
		Description: "Diagnoses payment failures and returns task progress plus final artifacts.",
		SupportedInterfaces: []*a2a.AgentInterface{
			a2a.NewAgentInterface(invokeURL, a2a.TransportProtocolJSONRPC),
		},
		DefaultInputModes:  []string{"text/plain", "application/json"},
		DefaultOutputModes: []string{"text/markdown", "application/json"},
		Capabilities: a2a.AgentCapabilities{
			Streaming: true,
		},
		Skills: []a2a.AgentSkill{
			{
				ID:          "payment_failure_diagnosis",
				Name:        "Payment failure diagnosis",
				Description: "Creates a task, updates task status while collecting evidence, and returns markdown/json artifacts.",
				Tags:        []string{"payment", "risk", "diagnosis", "artifact"},
				Examples:    []string{"为什么 order-10086 这笔订单支付失败？"},
			},
		},
	}

	trace := func(stage string, detail string) {
		log.Printf("[server] stage=%s detail=%s", stage, detail)
	}

	requestHandler := incident.NewPaymentDiagnosisHandler(trace)

	mux := http.NewServeMux()
	mux.Handle("/invoke", a2asrv.NewJSONRPCHandler(requestHandler))
	mux.Handle(a2asrv.WellKnownAgentCardPath, a2asrv.NewStaticAgentCardHandler(card))

	listener, err := net.Listen("tcp", fmt.Sprintf(":%d", *port))
	if err != nil {
		log.Fatalf("bind server port: %v", err)
	}

	log.Printf("A2A server listening on http://127.0.0.1:%d", *port)
	log.Printf("Agent card: http://127.0.0.1:%d%s", *port, a2asrv.WellKnownAgentCardPath)
	log.Printf("JSON-RPC endpoint: %s", invokeURL)

	if err := http.Serve(listener, mux); err != nil {
		log.Fatalf("serve: %v", err)
	}
}

这里有三个点最重要：

  1. SupportedInterfaces 把 JSON-RPC 地址写进 Agent Card
  2. Capabilities.Streaming = true 表示这个 Agent 支持流式事件
  3. Skills 描述这个 Agent 会做什么，客户端不需要预先硬编码能力

四、业务执行层：把一个问题变成 Task 流程

真正的业务逻辑在 incident 包里。它不是“直接返回诊断结论”，而是一步步把执行过程变成 A2A 事件。

1
2
3
4
5

type PaymentDiagnosisExecutor struct{ trace ServerTrace }

func NewPaymentDiagnosisHandler(trace ServerTrace) a2asrv.RequestHandler {
	return a2asrv.NewHandler(&PaymentDiagnosisExecutor{trace: trace})
}

执行流程如下：

  1. 第一次进来先创建 SubmittedTask
  2. 把任务状态改成 Working
  3. 分阶段写入进度消息
  4. 生成 Artifact
  5. 最后改成 Completed

核心 Execute 逻辑可以这样看：

 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

func (e *PaymentDiagnosisExecutor) Execute(_ context.Context, execCtx *a2asrv.ExecutorContext) iter.Seq2[a2a.Event, error] {
	return func(yield func(a2a.Event, error) bool) {
		e.record("server.receive_send_message", "a2asrv handler parsed SendMessage and built ExecutorContext")

		if execCtx.StoredTask == nil {
			e.record("task.create_submitted", "create Task with initial user message in history")
			if !yield(a2a.NewSubmittedTask(execCtx, execCtx.Message), nil) {
				return
			}
		}

		if !e.status(yield, execCtx, "task.status_working", a2a.TaskStateWorking, "已创建支付失败诊断任务，正在查询订单、支付网关和风控证据。") {
			return
		}

		report, err := DiagnoseQuestion(messageText(execCtx.Message))
		if err != nil {
			e.fail(yield, execCtx, "诊断失败："+err.Error())
			return
		}

		if !e.status(yield, execCtx, "task.progress_order_loaded", a2a.TaskStateWorking, "订单服务已返回：订单已创建，支付单为 "+report.PaymentID+"。") {
			return
		}
		if !e.status(yield, execCtx, "task.progress_gateway_loaded", a2a.TaskStateWorking, "支付网关已返回：RISK_REJECT，支付被风控拒绝。") {
			return
		}
		if !e.status(yield, execCtx, "task.progress_risk_loaded", a2a.TaskStateWorking, "风控服务已返回：高风险设备指纹规则命中。") {
			return
		}

		e.sleep("task.artifact_create")
		e.record("task.artifact_create", "append final markdown and json result parts to task artifacts")

		if !yield(newPaymentArtifact(execCtx, report), nil) {
			return
		}

		e.status(yield, execCtx, "task.status_completed", a2a.TaskStateCompleted, "诊断完成："+report.RootCause)
	}
}

这段代码体现了 A2A 的核心价值：不是“立刻吐一个答案”，而是让任务状态可见、过程可追踪、结果可落盘。

五、诊断报告：Markdown + JSON 双输出

业务模型层把诊断结果组织成两种视图：人类可读的 markdown 和机器可读的 JSON。

1
2
3
4
5
6
7
8

type DiagnosisReport struct {
	OrderID     string     `json:"orderId"`
	PaymentID   string     `json:"paymentId"`
	RootCause   string     `json:"rootCause"`
	Summary     string     `json:"summary"`
	Evidence    []Evidence `json:"evidence"`
	Suggestions []string   `json:"suggestions"`
}

生成 Artifact 时，两个 part 会一起挂到任务上：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17

func newPaymentArtifact(execCtx *a2asrv.ExecutorContext, report DiagnosisReport) *a2a.TaskArtifactUpdateEvent {
	markdownPart := a2a.NewTextPart(report.Markdown())
	markdownPart.MediaType = "text/markdown"
	markdownPart.Filename = "payment-root-cause-report.md"

	dataPart := a2a.NewDataPart(report.StructuredData())
	dataPart.MediaType = "application/json"
	dataPart.Filename = "payment-root-cause-report.json"

	event := a2a.NewArtifactEvent(execCtx, markdownPart, dataPart)
	event.LastChunk = true
	event.Artifact.Name = "payment-root-cause-report"
	event.Artifact.Description = "订单支付失败根因诊断报告"
	event.Artifact.SetMeta("business.orderId", report.OrderID)
	event.Artifact.SetMeta("business.paymentId", report.PaymentID)
	return event
}

这个设计很实用：

  • markdown 给人看
  • JSON 给后续系统接
  • LastChunk = true 表示这个 Artifact 已经写完

六、客户端源码：发现能力、发送消息、回收任务

客户端做三件事：解析 Agent Card、发消息、再取完整 Task。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10

card, err := agentcard.DefaultResolver.Resolve(ctx, *cardURL)
if err != nil {
	log.Fatalf("resolve agent card: %v", err)
}

client, err := a2aclient.NewFromCard(ctx, card)
if err != nil {
	log.Fatalf("create A2A client: %v", err)
}
defer client.Destroy()

然后按 stream 参数决定走流式还是非流式：

1
2
3
4
5

if *stream {
	taskID = runStreamingQuestion(ctx, client, *question)
} else {
	taskID = runNonStreamingQuestion(ctx, client, *question)
}

流式版本会实时打印事件：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15

for event, err := range incident.StreamPaymentQuestion(ctx, client, question) {
	if err != nil {
		log.Fatalf("stream payment question: %v", err)
	}
	if event == nil {
		continue
	}

	info := event.TaskInfo()
	if info.TaskID != "" {
		taskID = info.TaskID
	}

	fmt.Printf("  [%s] %s\n", time.Now().Format("15:04:05.000"), incident.DescribeStreamEvent(event))
}

最后再拉一次完整任务：

1
2
3
4

storedTask, err := incident.FetchTask(ctx, client, taskID)
if err != nil {
	log.Fatalf("get task: %v", err)
}

这一步很关键，因为流式事件适合看进度，GetTask 适合拿完整上下文、History 和 Artifacts。

七、示例业务：为什么 order-10086 支付失败

这里的示例诊断逻辑非常明确：

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21

func DiagnoseQuestion(question string) (DiagnosisReport, error) {
	if !strings.Contains(question, "order-10086") {
		return DiagnosisReport{}, errors.New("question must contain an order id like order-10086")
	}

	return DiagnosisReport{
		OrderID:   "order-10086",
		PaymentID: "pay-7788",
		RootCause: "支付网关返回 RISK_REJECT，订单命中高风险设备指纹规则",
		Summary:   "订单已创建成功，但支付阶段被风控拒绝。",
		Evidence: []Evidence{
			{Source: "order-service", Detail: "订单已创建，金额 1299.00 CNY"},
			{Source: "payment-gateway", Detail: "支付网关返回 RISK_REJECT"},
			{Source: "risk-engine", Detail: "设备 device-risk-9 风控分 97，超过拒绝阈值 90"},
		},
		Suggestions: []string{
			"提示用户更换支付方式或完成身份校验",
			"风控平台核验 device-risk-9 是否误杀",
		},
	}, nil
}

这说明一个 MVP 不一定要连真实系统，但必须把业务边界说清楚。A2A 负责承载任务流转，业务函数负责给出稳定、可解释的结论。

八、一次请求的完整流程

如果用户问：

为什么 order-10086 这笔订单支付失败？

整个链路会是：

  1. 客户端解析 Agent Card
  2. 客户端发送 SendStreamingMessage
  3. 服务端创建 SubmittedTask
  4. 服务端持续发 Working 状态
  5. 服务端生成 Artifact
  6. 服务端发 Completed
  7. 客户端通过 GetTask 拉回完整结果

这套流程的价值在于，前端、网关、外部系统都能按 Task ID 继续追踪，而不是只拿到一次性文本。

九、总结

这份 A2A MVP 的重点不是“做了一个问答接口”，而是把一个远端 Agent 做成了可发现、可流转、可追踪的任务系统。

如果你要把它继续往下做，最值得补的三块是：

  • 把 DiagnoseQuestion 换成真实订单、支付、风控查询
  • 把 Artifact 接到对象存储或知识库
  • 把 trace 日志接到统一观测平台

A2A 的落点很清楚：让 Agent 不只是会回答，而是能真正进入业务流程。

A2a Go Agent
本文采用 CC BY-NC-SA 4.0 协议，转载请注明出处。
© 2024 - 2026 DK Blog
记录工程实践和排障过程。
使用 Hugo 构建
主题 Stack 由 Jimmy 设计