AG-UI 与 A2A：Agent 与前端的协议层

用户界面（React / 移动端 / 桌面端）
        ▲
        │  AG-UI 事件流（SSE / WebSocket）
        │
Agent Runtime（LangGraph / CrewAI / 自研）
        │
        ├── MCP ──→ 工具与数据
        └── A2A ──→ 其他 Agent

{
  "threadId": "thread-42",
  "runId": "run-1001",
  "messages": [
    {
      "id": "m-1",
      "role": "user",
      "content": "帮我查一下订单 A1024，如果已发货就申请退款"
    }
  ],
  "tools": [],
  "context": [],
  "state": {},
  "forwardedProps": {}
}

RUN_STARTED            { threadId, runId }
  STEP_STARTED         { stepName: "plan" }
  STEP_FINISHED        { stepName: "plan" }

  TOOL_CALL_START      { toolCallId, toolCallName: "get_order" }
  TOOL_CALL_ARGS       { toolCallId, delta: "{\"orderId\":" }
  TOOL_CALL_ARGS       { toolCallId, delta: "\"A1024\"}" }
  TOOL_CALL_END        { toolCallId }
  TOOL_CALL_RESULT     { toolCallId, messageId, content: "..." }

  STATE_SNAPSHOT       { snapshot: { order: {...}, step: "confirm_refund" } }
  STATE_DELTA          { delta: [{ "op": "replace", "path": "/step", "value": "await_confirm" }] }

  TEXT_MESSAGE_START   { messageId, role: "assistant" }
  TEXT_MESSAGE_CONTENT { messageId, delta: "订单已发货，" }
  TEXT_MESSAGE_CONTENT { messageId, delta: "需要你确认后退款。" }
  TEXT_MESSAGE_END     { messageId }

RUN_FINISHED           { threadId, runId, outcome: { type: "success" } }

POST /ag-ui/agent HTTP/1.1
Content-Type: application/json
Accept: text/event-stream

{ "threadId": "t-1", "runId": "r-1", "messages": [ ... ] }

HTTP/1.1 200 OK
Content-Type: text/event-stream
Cache-Control: no-cache

id: 0
data: {"type":"RUN_STARTED","threadId":"t-1","runId":"r-1"}

id: 1
data: {"type":"TEXT_MESSAGE_START","messageId":"a-1","role":"assistant"}

断线
 ├─ 不要假设服务端取消了 run（SSE 断开不等于取消）
 ├─ 用 threadId + runId 重新订阅或查询 run 状态
 ├─ 收到 STATE_SNAPSHOT / MESSAGES_SNAPSHOT 后整体替换本地状态
 └─ 如果 run 已结束，补拉最后一次快照，避免丢最终结果

STATE_SNAPSHOT  → 全量替换（连接开始、重连、检测到漂移）
STATE_DELTA     → JSON Patch 增量（高频小变更）
MESSAGES_SNAPSHOT → 对话历史全量（初始化、恢复）

type State = { order?: Order; step: string; pending?: string };

function reduce(state: State, event: AgUiEvent): State {
  switch (event.type) {
    case "STATE_SNAPSHOT":
      return event.snapshot as State; // 替换，不合并
    case "STATE_DELTA":
      return applyPatch(state, event.delta).newDocument;
    case "MESSAGES_SNAPSHOT":
      return { ...state, messages: event.messages };
    default:
      return state;
  }
}

{
  "type": "RUN_FINISHED",
  "threadId": "t-1",
  "runId": "r-1",
  "outcome": {
    "type": "interrupt",
    "interrupts": [
      {
        "id": "int-1",
        "reason": "tool_call",
        "message": "即将为订单 A1024 发起退款 128.00 元，是否确认？",
        "toolCallId": "tc-9",
        "responseSchema": {
          "type": "object",
          "properties": {
            "approved": { "type": "boolean" },
            "editedArgs": { "type": "object" }
          },
          "required": ["approved"]
        },
        "expiresAt": "2026-11-14T12:05:00Z"
      }
    ]
  }
}

{
  "threadId": "t-1",
  "runId": "r-2",
  "messages": [],
  "state": {},
  "resume": [
    {
      "interruptId": "int-1",
      "status": "resolved",
      "payload": { "approved": true }
    }
  ]
}

run-1: TOOL_CALL_ARGS        ← Agent 提议的参数
run-2: resume.payload        ← 用户的决定与修改
run-2: TOOL_CALL_RESULT      ← 实际执行结果（针对原 toolCallId）

async function onApprove(intr: Interrupt, approved: boolean) {
  const res = await fetch("/ag-ui/agent", {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify({
      threadId: intr.threadId,
      runId: crypto.randomUUID(),
      messages: [],
      state: {},
      resume: [
        { interruptId: intr.id, status: "resolved", payload: { approved } },
      ],
    }),
  });
  await consumeSse(res.body!);
}

{
  "name": "travel-booking-agent",
  "description": "负责机票与酒店预订",
  "provider": { "organization": "Example Travel" },
  "supportedInterfaces": [
    { "url": "https://agent.example.com/a2a", "protocolBinding": "JSONRPC", "protocolVersion": "1.0" }
  ],
  "capabilities": { "streaming": true, "pushNotifications": true },
  "securitySchemes": { "oauth": { "type": "oauth2", "flows": { ... } } },
  "skills": [
    {
      "id": "book-flight",
      "name": "预订航班",
      "description": "根据出发地、目的地、日期预订航班",
      "tags": ["travel", "booking"],
      "examples": ["帮我订 12 月 1 日北京到东京的航班"]
    }
  ]
}

SUBMITTED ──→ WORKING ──→ COMPLETED
     │            │  │
     │            │  ├──→ FAILED
     │            │  ├──→ CANCELED
     │            │  └──→ REJECTED
     │            ├──→ INPUT_REQUIRED ──→ WORKING
     │            └──→ AUTH_REQUIRED  ──→ WORKING

┌──────────────────────────────────────────────────────────────┐
│ 用户界面                                                      │
│  聊天 / 工具卡片 / 审批弹窗 / 生成式 UI                        │
└───────────────────────────┬──────────────────────────────────┘
                            │ AG-UI（事件流：文本 / 工具 / 状态 / 中断）
┌───────────────────────────▼──────────────────────────────────┐
│ Agent Runtime（规划、记忆、编排、审批）                        │
│                                                              │
│   ┌──────────────────────┐      ┌─────────────────────────┐  │
│   │ MCP Client           │      │ A2A Client              │  │
│   │ 工具 / 资源 / 提示词  │      │ 任务委派 / 产物接收      │  │
│   └──────────┬───────────┘      └────────────┬────────────┘  │
└──────────────┼───────────────────────────────┼───────────────┘
               │ MCP（JSON-RPC / stdio / HTTP） │ A2A（Task / Message / Artifact）
        ┌──────▼──────┐                 ┌──────▼──────────────┐
        │ MCP Server  │                 │ 远程 Agent          │
        │ 文件/DB/API │                 │ （不透明，可能再用 MCP）│
        └─────────────┘                 └─────────────────────┘

前端                  Agent Runtime              MCP Server          用户
 │  POST /ag-ui/agent      │                         │                │
 │────────────────────────>│                         │                │
 │                         │ tools/call get_order    │                │
 │                         │────────────────────────>│                │
 │                         │<────────────────────────│                │
 │  RUN_STARTED            │                         │                │
 │<────────────────────────│                         │                │
 │  TOOL_CALL_START/ARGS   │                         │                │
 │<────────────────────────│                         │                │
 │  TOOL_CALL_RESULT       │                         │                │
 │<────────────────────────│                         │                │
 │  STATE_SNAPSHOT         │                         │                │
 │<────────────────────────│                         │                │
 │  RUN_FINISHED(interrupt)│                         │                │
 │<────────────────────────│                         │                │
 │  展示审批卡片 ────────────────────────────────────────────────────>│
 │<──────────────────────────────────────────── 点击"确认退款"        │
 │  POST /ag-ui/agent (resume)                       │                │
 │────────────────────────>│ tools/call refund_order │                │
 │                         │────────────────────────>│                │
 │  TOOL_CALL_RESULT       │                         │                │
 │<────────────────────────│                         │                │
 │  TEXT_MESSAGE_*         │                         │                │
 │<────────────────────────│                         │                │
 │  RUN_FINISHED(success)  │                         │                │
 │<────────────────────────│                         │                │

import { Client } from "@modelcontextprotocol/sdk/client/index.js";
import { StreamableHTTPClientTransport } from "@modelcontextprotocol/sdk/client/streamableHttp.js";

const mcp = new Client({ name: "agent-runtime", version: "1.0.0" });
await mcp.connect(
  new StreamableHTTPClientTransport(new URL("https://mcp.internal/mcp")),
);

type Emit = (event: Record<string, unknown>) => void;

async function run(input: RunAgentInput, emit: Emit) {
  emit({ type: "RUN_STARTED", threadId: input.threadId, runId: input.runId });

  try {
    // 1. 调用 MCP 工具，并把调用过程映射成 AG-UI 事件
    const toolCallId = `tc-${input.runId}`;
    emit({ type: "TOOL_CALL_START", toolCallId, toolCallName: "get_order" });
    emit({
      type: "TOOL_CALL_ARGS",
      toolCallId,
      delta: JSON.stringify({ orderId: "A1024" }),
    });
    emit({ type: "TOOL_CALL_END", toolCallId });

    const order = await mcp.callTool({
      name: "get_order",
      arguments: { orderId: "A1024" },
    });

    if ((order as any).isError) {
      emit({ type: "RUN_ERROR", code: "TOOL_ERROR", message: "订单查询失败" });
      return;
    }

    emit({
      type: "TOOL_CALL_RESULT",
      messageId: `msg-${input.runId}`,
      toolCallId,
      content: JSON.stringify(order.structuredContent ?? order.content),
    });

    // 2. 中断前必须先发状态快照
    emit({
      type: "STATE_SNAPSHOT",
      snapshot: { order: order.structuredContent, step: "await_confirm" },
    });

    // 3. 需要人工确认：以 interrupt 结束本次 run
    emit({
      type: "RUN_FINISHED",
      threadId: input.threadId,
      runId: input.runId,
      outcome: {
        type: "interrupt",
        interrupts: [
          {
            id: `int-${input.runId}`,
            reason: "tool_call",
            message: "确认对订单 A1024 发起退款？",
            toolCallId,
            responseSchema: {
              type: "object",
              properties: { approved: { type: "boolean" } },
              required: ["approved"],
            },
          },
        ],
      },
    });
  } catch (err) {
    emit({ type: "RUN_ERROR", message: "运行时异常", code: "RUNTIME_ERROR" });
  }
}

async function resume(input: RunAgentInput, emit: Emit) {
  const decision = input.resume?.[0];
  if (!decision || decision.status !== "resolved") {
    emit({
      type: "RUN_FINISHED",
      outcome: { type: "interrupt", interrupts: pending },
    });
    return;
  }

  emit({ type: "RUN_STARTED", threadId: input.threadId, runId: input.runId });

  if (decision.payload?.approved !== true) {
    emit({ type: "TEXT_MESSAGE_START", messageId: "m-no", role: "assistant" });
    emit({
      type: "TEXT_MESSAGE_CONTENT",
      messageId: "m-no",
      delta: "已取消退款。",
    });
    emit({ type: "TEXT_MESSAGE_END", messageId: "m-no" });
    emit({ type: "RUN_FINISHED", outcome: { type: "success" } });
    return;
  }

  const result = await mcp.callTool({
    name: "refund_order",
    arguments: {
      orderId: "A1024",
      idempotencyKey: `refund:${input.threadId}:int-1`,
    },
  });

  // 恢复的 run 不重发 TOOL_CALL_START/ARGS，只发 RESULT 挂回原 toolCallId
  emit({
    type: "TOOL_CALL_RESULT",
    messageId: "m-refund",
    toolCallId: "tc-run-1",
    content: JSON.stringify(result.structuredContent ?? result.content),
  });
  emit({ type: "TEXT_MESSAGE_START", messageId: "m-done", role: "assistant" });
  emit({
    type: "TEXT_MESSAGE_CONTENT",
    messageId: "m-done",
    delta: "退款已提交。",
  });
  emit({ type: "TEXT_MESSAGE_END", messageId: "m-done" });
  emit({ type: "RUN_FINISHED", outcome: { type: "success" } });
}

async function consumeSse(body: ReadableStream<Uint8Array>) {
  const reader = body.getReader();
  const decoder = new TextDecoder();
  let buffer = "";

  while (true) {
    const { value, done } = await reader.read();
    if (done) break;
    buffer += decoder.decode(value, { stream: true });

    // SSE 以空行分隔事件；只处理 data: 行
    const chunks = buffer.split("\n\n");
    buffer = chunks.pop() ?? "";
    for (const chunk of chunks) {
      const data = chunk
        .split("\n")
        .filter((l) => l.startsWith("data:"))
        .map((l) => l.slice(5).trim())
        .join("");
      if (data) dispatch(JSON.parse(data));
    }
  }
}