AG-UI 协议调研

// 核心 Agent 执行接口
type RunAgent = () => Observable<BaseEvent>;

class MyAgent extends AbstractAgent {
  run(input: RunAgentInput): RunAgent {
    const { threadId, runId } = input;
    return () =>
      from([
        { type: EventType.RUN_STARTED, threadId, runId },
        {
          type: EventType.MESSAGES_SNAPSHOT,
          messages: [
            { id: "msg_1", role: "assistant", content: "Hello, world!" },
          ],
        },
        { type: EventType.RUN_FINISHED, threadId, runId },
      ]);
  }
}

interface BaseEvent {
  type: EventType; // 事件类型枚举
  timestamp?: number; // 可选时间戳
  rawEvent?: any; // 可选原始事件（用于调试/兼容）
}

// 消息开始
{ type: EventType.TEXT_MESSAGE_START, messageId: "msg_123" }

// 内容流式传输（多次）
{ type: EventType.TEXT_MESSAGE_CONTENT, messageId: "msg_123", delta: "Hello" }
{ type: EventType.TEXT_MESSAGE_CONTENT, messageId: "msg_123", delta: ", world" }
{ type: EventType.TEXT_MESSAGE_CONTENT, messageId: "msg_123", delta: "!" }

// 消息结束
{ type: EventType.TEXT_MESSAGE_END, messageId: "msg_123" }

// 工具调用开始
{
  type: EventType.TOOL_CALL_START,
  toolCallId: "tool-123",
  toolCallName: "confirmAction",
  parentMessageId: "msg-456"  // 可选，关联消息
}

// 参数流式传输（JSON 片段）
{ type: EventType.TOOL_CALL_ARGS, toolCallId: "tool-123", delta: '{"act' }
{ type: EventType.TOOL_CALL_ARGS, toolCallId: "tool-123", delta: 'ion":"Depl' }
{ type: EventType.TOOL_CALL_ARGS, toolCallId: "tool-123", delta: 'oy the app"}' }

// 工具调用结束
{ type: EventType.TOOL_CALL_END, toolCallId: "tool-123" }

interface BaseMessage {
  id: string; // 消息唯一标识
  role: string; // 发送者角色
  content?: string; // 可选文本内容
  name?: string; // 可选发送者名称
}

interface UserMessage {
  id: string;
  role: "user";
  content: string | InputContent[]; // 支持多模态
  name?: string;
}

type InputContent = TextInputContent | BinaryInputContent;

interface BinaryInputContent {
  type: "binary";
  mimeType: string; // 如 "image/png"
  id?: string; // 引用 ID
  url?: string; // 远程 URL
  data?: string; // Base64 数据
  filename?: string; // 文件名
}

interface ActivityMessage {
  id: string;
  role: "activity";
  activityType: string; // 如 "PLAN", "SEARCH", "SCRAPE"
  content: Record<string, any>; // 结构化 payload
}

interface Tool {
  name: string         // 工具唯一标识
  description: string  // 人类可读描述（LLM 使用）
  parameters: {        // JSON Schema 参数定义
    type: "object"
    properties: { ... }
    required: string[]
  }
}

// 前端定义确认工具
const confirmAction = {
  name: "confirmAction",
  description: "Ask the user to confirm a specific action before proceeding",
  parameters: {
    type: "object",
    properties: {
      action: {
        type: "string",
        description: "The action that needs user confirmation",
      },
      importance: {
        type: "string",
        enum: ["low", "medium", "high", "critical"],
        description: "The importance level of the action",
      },
    },
    required: ["action"],
  },
};

// 运行 Agent 时传入工具
agent.runAgent({
  tools: [confirmAction], // 前端控制工具可用性
  // ...
});

interface StateSnapshotEvent {
  type: EventType.STATE_SNAPSHOT;
  snapshot: any; // 完整状态对象
}

interface StateDeltaEvent {
  type: EventType.STATE_DELTA;
  delta: JsonPatchOperation[];
}

interface JsonPatchOperation {
  op: "add" | "remove" | "replace" | "move" | "copy" | "test";
  path: string; // JSON Pointer (RFC 6901)
  value?: any; // add, replace 时使用
  from?: string; // move, copy 时使用
}

// 添加用户偏好
{ "op": "add", "path": "/user/preferences", "value": { "theme": "dark" } }

// 替换对话状态
{ "op": "replace", "path": "/conversation_state", "value": "paused" }

// 移除临时数据
{ "op": "remove", "path": "/temporary_data" }

// 移动待办项到已完成
{ "op": "move", "path": "/completed_items", "from": "/pending_items/0" }

case EventType.STATE_DELTA: {
  const { delta } = event as StateDeltaEvent;
  try {
    // 原子性应用 JSON Patch，不修改原状态
    const result = applyPatch(state, delta, true, false);
    state = result.newDocument;
    return emitUpdate({ state });
  } catch (error: unknown) {
    console.warn(`Failed to apply state patch...`);
    return emitNoUpdate();
  }
}

type RunStartedEvent = BaseEvent & {
  type: EventType.RUN_STARTED;
  threadId: string;
  runId: string;
  /** 用于分支/时间旅行的父运行 ID */
  parentRunId?: string;
  /** 本次运行的精确 Agent 输入（可省略已在历史中的消息） */
  input?: AgentInput;
};

function compactEvents(events: BaseEvent[]): BaseEvent[];

// 压缩前：冗余的流式事件
[
  { type: "TEXT_MESSAGE_START", messageId: "m1", role: "assistant" },
  { type: "TEXT_MESSAGE_CONTENT", messageId: "m1", delta: "Hello" },
  { type: "TEXT_MESSAGE_CONTENT", messageId: "m1", delta: " " },
  { type: "CUSTOM", name: "thinking" },
  { type: "TEXT_MESSAGE_CONTENT", messageId: "m1", delta: "world" },
  { type: "TEXT_MESSAGE_END", messageId: "m1" },
]

// 压缩后：精简的快照
[
  { type: "TEXT_MESSAGE_START", messageId: "m1", role: "assistant" },
  { type: "TEXT_MESSAGE_CONTENT", messageId: "m1", delta: "Hello world" },
  { type: "TEXT_MESSAGE_END", messageId: "m1" },
  { type: "CUSTOM", name: "thinking" },  // 交错事件移到末尾
]

// 原始运行
{
  type: "RUN_STARTED",
  threadId: "thread1",
  runId: "run1",
  input: { messages: ["Tell me about Paris"] }
}

// 从 run1 分支
{
  type: "RUN_STARTED",
  threadId: "thread1",
  runId: "run2",
  parentRunId: "run1",  // 关键：指向父运行
  input: { messages: ["Actually, tell me about London instead"] }
}

import { AbstractAgent } from "@ag-ui/client";

const agent = new MyAgent();

// 中间件链：logging -> auth -> filter -> agent
agent.use(loggingMiddleware, authMiddleware, filterMiddleware);

// 运行 Agent 时，事件流经所有中间件
await agent.runAgent();

import { MiddlewareFunction } from "@ag-ui/client";
import { EventType } from "@ag-ui/core";

const prefixMiddleware: MiddlewareFunction = (input, next) => {
  return next.run(input).pipe(
    map((event) => {
      if (event.type === EventType.TEXT_MESSAGE_CHUNK) {
        return { ...event, delta: `[AI]: ${event.delta}` };
      }
      return event;
    })
  );
};

agent.use(prefixMiddleware);

import { Middleware } from "@ag-ui/client";
import { Observable } from "rxjs";
import { tap, finalize } from "rxjs/operators";

class MetricsMiddleware extends Middleware {
  private eventCount = 0;

  constructor(private metricsService: MetricsService) {
    super();
  }

  run(input: RunAgentInput, next: AbstractAgent): Observable<BaseEvent> {
    const startTime = Date.now();

    return next.run(input).pipe(
      tap((event) => {
        this.eventCount++;
        this.metricsService.recordEvent(event.type);
      }),
      finalize(() => {
        const duration = Date.now() - startTime;
        this.metricsService.recordDuration(duration);
        this.metricsService.recordEventCount(this.eventCount);
      })
    );
  }
}

agent.use(new MetricsMiddleware(metricsService));

{
  "tool": "generateUserInterface",
  "arguments": {
    "description": "A form that collects a user's shipping address.",
    "data": {
      "firstName": "Ada",
      "lastName": "Lovelace",
      "city": "London"
    },
    "output": {
      "type": "object",
      "required": [
        "firstName",
        "lastName",
        "street",
        "city",
        "postalCode",
        "country"
      ],
      "properties": {
        "firstName": { "type": "string", "title": "First Name" },
        "lastName": { "type": "string", "title": "Last Name" },
        "street": { "type": "string", "title": "Street Address" },
        "city": { "type": "string", "title": "City" },
        "postalCode": { "type": "string", "title": "Postal Code" },
        "country": {
          "type": "string",
          "title": "Country",
          "enum": ["GB", "US", "DE", "AT"]
        }
      }
    }
  }
}

type RunFinishedOutcome = "success" | "interrupt";

type RunFinished = {
  type: "RUN_FINISHED";
  // ... existing fields

  outcome?: RunFinishedOutcome; // 可选，向后兼容

  // outcome === "success" 时存在
  result?: any;

  // outcome === "interrupt" 时存在
  interrupt?: {
    id?: string; // 中断 ID
    reason?: string; // 如 "human_approval", "upload_required", "policy_hold"
    payload?: any; // 任意 JSON（表单、提案、diff 等）
  };
};

type RunAgentInput = {
  // ... existing fields

  // 恢复中断的通道
  resume?: {
    interruptId?: string; // 回传中断 ID
    payload?: any; // 任意 JSON：审批结果、编辑内容、文件引用等
  };
};

// 推理开始
type ReasoningStartEvent = BaseEvent & {
  type: EventType.REASONING_START;
  messageId: string;
  encryptedContent?: string; // 可选加密内容（用于隐私保护）
};

// 推理消息开始
type ReasoningMessageStartEvent = BaseEvent & {
  type: EventType.REASONING_MESSAGE_START;
  messageId: string;
  role: "assistant";
};

// 推理内容流式传输
type ReasoningMessageContentEvent = BaseEvent & {
  type: EventType.REASONING_MESSAGE_CONTENT;
  messageId: string;
  delta: string; // 非空字符串
};

// 推理消息结束
type ReasoningMessageEndEvent = BaseEvent & {
  type: EventType.REASONING_MESSAGE_END;
  messageId: string;
};

// 推理消息 Chunk（便捷事件）
type ReasoningMessageChunkEvent = BaseEvent & {
  type: EventType.REASONING_MESSAGE_CHUNK;
  messageId?: string;
  delta?: string;
};

// 推理结束
type ReasoningEndEvent = BaseEvent & {
  type: EventType.REASONING_END;
  messageId: string;
};

interface TextInputContent {
  type: "text";
  text: string;
}

interface BinaryInputContent {
  type: "binary";
  mimeType: string; // 如 "image/jpeg", "audio/wav", "application/pdf"
  id?: string; // 预上传内容的引用 ID
  url?: string; // 远程 URL
  data?: string; // Base64 编码数据
  filename?: string; // 文件名
}

type InputContent = TextInputContent | BinaryInputContent;

// 图像 + 文本
{
  "id": "msg-002",
  "role": "user",
  "content": [
    { "type": "text", "text": "What's in this image?" },
    {
      "type": "binary",
      "mimeType": "image/jpeg",
      "data": "base64-encoded-image-data..."
    }
  ]
}

// 多图片对比
{
  "id": "msg-003",
  "role": "user",
  "content": [
    { "type": "text", "text": "What are the differences between these images?" },
    { "type": "binary", "mimeType": "image/png", "url": "https://example.com/image1.png" },
    { "type": "binary", "mimeType": "image/png", "url": "https://example.com/image2.png" }
  ]
}

// 文档分析
{
  "id": "msg-005",
  "role": "user",
  "content": [
    { "type": "text", "text": "Summarize the key points from this PDF" },
    {
      "type": "binary",
      "mimeType": "application/pdf",
      "filename": "quarterly-report.pdf",
      "url": "https://example.com/reports/q4-2024.pdf"
    }
  ]
}

# 创建新项目
npx create-ag-ui-app@latest

# 启动开发服务器
npm run dev

# 访问应用
# http://localhost:3000/copilotkit

npm install @ag-ui/client @ag-ui/core rxjs

import { HttpAgent } from "@ag-ui/client";
import { EventType } from "@ag-ui/core";

// 创建 HTTP Agent 客户端
const agent = new HttpAgent({
  url: "https://your-agent-endpoint.com/agent",
  agentId: "unique-agent-id",
  threadId: "conversation-thread",
});

// 定义前端工具
const confirmAction = {
  name: "confirmAction",
  description: "Ask the user to confirm a specific action",
  parameters: {
    type: "object",
    properties: {
      action: { type: "string", description: "The action to confirm" },
      importance: {
        type: "string",
        enum: ["low", "medium", "high", "critical"],
      },
    },
    required: ["action"],
  },
};

// 启动 Agent 并处理事件
agent
  .runAgent({
    tools: [confirmAction],
    context: [{ type: "text", text: "User is on the checkout page" }],
  })
  .subscribe({
    next: (event) => {
      switch (event.type) {
        case EventType.RUN_STARTED:
          console.log("Agent started:", event.runId);
          break;

        case EventType.TEXT_MESSAGE_CONTENT:
          console.log("Content:", event.delta);
          break;

        case EventType.TOOL_CALL_START:
          console.log("Tool call:", event.toolCallName);
          // 渲染确认对话框...
          break;

        case EventType.STATE_DELTA:
          console.log("State update:", event.delta);
          break;

        case EventType.RUN_FINISHED:
          console.log("Agent finished");
          break;
      }
    },
    error: (error) => console.error("Agent error:", error),
    complete: () => console.log("Agent run complete"),
  });

import os
import uuid
import uvicorn
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from ag_ui.core import (
    RunAgentInput,
    EventType,
    RunStartedEvent,
    RunFinishedEvent,
    RunErrorEvent,
)
from ag_ui.encoder import EventEncoder
from openai import OpenAI

app = FastAPI(title="AG-UI OpenAI Server")

# 初始化 OpenAI 客户端（使用环境变量 OPENAI_API_KEY）
client = OpenAI()

@app.post("/")
async def agentic_chat_endpoint(input_data: RunAgentInput, request: Request):
    """AG-UI 兼容的 OpenAI 聊天端点"""

    # 获取客户端期望的编码格式
    accept_header = request.headers.get("accept")
    encoder = EventEncoder(accept=accept_header)

    async def event_generator():
        try:
            # 1. 发射 RUN_STARTED 事件
            yield encoder.encode(
                RunStartedEvent(
                    type=EventType.RUN_STARTED,
                    thread_id=input_data.thread_id,
                    run_id=input_data.run_id
                )
            )

            # 2. 调用 OpenAI API（启用流式传输）
            stream = client.chat.completions.create(
                model="gpt-4o",
                stream=True,
                # 转换 AG-UI 工具格式为 OpenAI 格式
                tools=[
                    {
                        "type": "function",
                        "function": {
                            "name": tool.name,
                            "description": tool.description,
                            "parameters": tool.parameters,
                        }
                    }
                    for tool in input_data.tools
                ] if input_data.tools else None,
                # 转换 AG-UI 消息为 OpenAI 消息格式
                messages=[
                    {
                        "role": message.role,
                        "content": message.content or "",
                        # 包含工具调用（如果是 assistant 消息）
                        **({"tool_calls": message.tool_calls}
                           if message.role == "assistant"
                           and hasattr(message, 'tool_calls')
                           and message.tool_calls else {}),
                        # 包含工具调用 ID（如果是 tool 消息）
                        **({"tool_call_id": message.tool_call_id}
                           if message.role == "tool"
                           and hasattr(message, 'tool_call_id') else {}),
                    }
                    for message in input_data.messages
                ],
            )

            message_id = str(uuid.uuid4())

            # 3. 流式转发 OpenAI 响应
            for chunk in stream:
                # 处理文本内容
                if chunk.choices[0].delta.content:
                    yield encoder.encode({
                        "type": EventType.TEXT_MESSAGE_CHUNK,
                        "message_id": message_id,
                        "delta": chunk.choices[0].delta.content,
                    })
                # 处理工具调用
                elif chunk.choices[0].delta.tool_calls:
                    tool_call = chunk.choices[0].delta.tool_calls[0]
                    yield encoder.encode({
                        "type": EventType.TOOL_CALL_CHUNK,
                        "tool_call_id": tool_call.id,
                        "tool_call_name": tool_call.function.name if tool_call.function else None,
                        "parent_message_id": message_id,
                        "delta": tool_call.function.arguments if tool_call.function else None,
                    })

            # 4. 发射 RUN_FINISHED 事件
            yield encoder.encode(
                RunFinishedEvent(
                    type=EventType.RUN_FINISHED,
                    thread_id=input_data.thread_id,
                    run_id=input_data.run_id
                )
            )

        except Exception as error:
            # 5. 发射 RUN_ERROR 事件
            yield encoder.encode(
                RunErrorEvent(
                    type=EventType.RUN_ERROR,
                    message=str(error)
                )
            )

    return StreamingResponse(
        event_generator(),
        media_type=encoder.get_content_type()
    )

def main():
    """启动 uvicorn 服务器"""
    port = int(os.getenv("PORT", "8000"))
    uvicorn.run("example_server:app", host="0.0.0.0", port=port, reload=True)

if __name__ == "__main__":
    main()

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

app = FastAPI()

# 创建 LangGraph Agent
llm = ChatOpenAI(model="gpt-4")
agent = create_react_agent(llm, tools=[...])

@app.post("/agent")
async def run_agent(request: dict):
    """AG-UI 兼容的 Agent 端点"""

    async def event_stream():
        # 发射 RUN_STARTED
        yield f"data: {json.dumps({'type': 'RUN_STARTED', 'runId': request['runId']})}\n\n"

        # 运行 Agent
        async for event in agent.astream(request["messages"]):
            if "content" in event:
                # 流式文本消息
                yield f"data: {json.dumps({'type': 'TEXT_MESSAGE_CONTENT', 'delta': event['content']})}\n\n"
            elif "tool_calls" in event:
                # 工具调用
                for tc in event["tool_calls"]:
                    yield f"data: {json.dumps({'type': 'TOOL_CALL_START', 'toolCallId': tc['id'], 'toolCallName': tc['name']})}\n\n"

        # 发射 RUN_FINISHED
        yield f"data: {json.dumps({'type': 'RUN_FINISHED', 'runId': request['runId']})}\n\n"

    return StreamingResponse(event_stream(), media_type="text/event-stream")

import { CopilotKit, useCopilotAction } from "@copilotkit/react-core";
import { CopilotChat } from "@copilotkit/react-ui";

function App() {
  return (
    <CopilotKit
      runtimeUrl="http://localhost:8000/agent"
      agent="langgraph-agent"
    >
      <MyChat />
    </CopilotKit>
  );
}

function MyChat() {
  // 定义前端工具
  useCopilotAction({
    name: "confirmAction",
    description: "Confirm an action",
    parameters: [
      { name: "action", type: "string", required: true },
      { name: "importance", type: "string", enum: ["low", "medium", "high"] },
    ],
    handler: async ({ action, importance }) => {
      // 显示确认对话框
      const confirmed = await showConfirmDialog(action, importance);
      return confirmed ? "User confirmed" : "User rejected";
    },
  });

  return <CopilotChat />;
}

from google.adk import Agent, Tool
from copilotkit.integrations.adk import ADKMiddleware

# 定义 ADK Agent
@Agent
class MyAgent:
    @Tool
    def search_web(self, query: str) -> str:
        """Search the web for information."""
        return f"Results for: {query}"

    @Tool
    def confirm_action(self, action: str, importance: str = "medium") -> str:
        """Ask user to confirm an action."""
        # 这将触发前端工具调用
        return "Awaiting user confirmation..."

# 使用 AG-UI 中间件包装
from fastapi import FastAPI
app = FastAPI()

middleware = ADKMiddleware(MyAgent())

@app.post("/agent")
async def run_agent(request: dict):
    return await middleware.handle(request)

import { Middleware, RunAgentInput, AbstractAgent } from "@ag-ui/client";
import { BaseEvent, EventType } from "@ag-ui/core";
import { Observable } from "rxjs";
import { map, tap } from "rxjs/operators";

// 日志中间件
class LoggingMiddleware extends Middleware {
  run(input: RunAgentInput, next: AbstractAgent): Observable<BaseEvent> {
    console.log(`[${new Date().toISOString()}] Agent run started`);

    return next.run(input).pipe(
      tap((event) => {
        console.log(`[${event.type}]`, event);
      })
    );
  }
}

// 认证中间件
class AuthMiddleware extends Middleware {
  constructor(private getToken: () => string) {
    super();
  }

  run(input: RunAgentInput, next: AbstractAgent): Observable<BaseEvent> {
    // 注入认证 token 到上下文
    const authenticatedInput = {
      ...input,
      context: [
        ...(input.context || []),
        { type: "text", text: `Bearer ${this.getToken()}` },
      ],
    };
    return next.run(authenticatedInput);
  }
}

// 事件转换中间件
class TransformMiddleware extends Middleware {
  run(input: RunAgentInput, next: AbstractAgent): Observable<BaseEvent> {
    return next.run(input).pipe(
      map((event) => {
        if (event.type === EventType.TEXT_MESSAGE_CONTENT) {
          // 添加前缀
          return { ...event, delta: `🤖 ${event.delta}` };
        }
        return event;
      })
    );
  }
}

// 组合使用
const agent = new HttpAgent({ url: "..." });
agent.use(
  new LoggingMiddleware(),
  new AuthMiddleware(() => localStorage.getItem("token")),
  new TransformMiddleware()
);

MCP (工具)  +  A2A (协作)  +  AG-UI (用户)  =  完整 Agentic 架构