agent-api-server 2.1.11

背景：Agent 协作的协议格局

问题：Agent 孤岛

随着 AI Agent 在企业中的广泛应用，一个核心挑战日益凸显：不同团队、不同框架、不同组织开发的 Agent 如何有效协作？ 当前业界存在多个 Agent 开发框架（LangGraph、CrewAI、AutoGen、Semantic Kernel、OpenAI Agents SDK 等），每个框架产出的 Agent 天然是"孤岛"，缺乏统一的互操作标准。

与此同时，Agent 互操作协议正在快速演进。2025 年 4 月，Google 发布 A2A 协议；2025 年 11 月，MCP 以实验特性形式发布 Tasks 扩展；2025 年 12 月，Google 将 A2A 捐赠给 Linux 基金会；2026 年 3 月，Agent API Initiative (AAIF) 成立；2026 年 7 月 28 日，MCP 发布了里程碑式的 Release Candidate 规范。AgentHub SDK 的核心使命就是解决这一问题——提供一套面向多协议、跨框架的 Agent 服务化基础设施。

三大主流 Agent 调用协议

当前业界围绕 Agent 互操作形成了三个互补的协议标准，各自解决不同层次的协作问题：

协议	核心定位	应用场景	传输方式	治理方
MCP (Model Context Protocol)	LLM 调用外部工具	LLM 访问数据库、API、文件系统；Agent 作为工具被 LLM 调用	Streamable HTTP (stdio 可选)	Anthropic → AAIF
ACP (Agent Client Protocol)	编辑器集成 Agent	VS Code、Zed、JetBrains 等编辑器集成 AI 编码助手	stdio JSON-RPC	Zed Industries / IBM
A2A (Agent-to-Agent)	Agent 间对等协作	跨系统、跨组织的 Agent 编排；多 Agent 链式调用	JSON-RPC 2.0 over HTTP(S)	Linux 基金会 / AAIF

MCP — Model Context Protocol（LLM 调用外部工具）

由 Anthropic 于 2024 年推出，定位为 LLM 与外部工具/数据源之间的标准化接口。核心理念是让 LLM 以统一的方式发现和调用工具——Agent 在这里扮演"工具"的角色，被 LLM 作为 Function Calling 的目标。

• 适用场景：LLM 需要访问数据库、API、文件系统等外部资源；Agent 暴露为 tool 供 Claude/GPT 等模型调用
• 生态：Anthropic 主导，已获广泛社区支持，大量 MCP Server 实现
• 局限：面向工具调用，不解决 Agent 之间的对等协作问题

ACP — Agent Client Protocol（编辑器集成 Agent）

由 Zed Industries 发起，IBM 参与推动，定位为 代码编辑器/IDE 与编码 Agent 之间的标准化通信协议。

• 适用场景：VS Code、Zed、JetBrains 等编辑器集成 AI 编码助手
• 特点：基于 stdio JSON-RPC，支持本地和远程两种模式；复用 MCP 的 JSON 表示但增加了编码 UX 专用类型（如 diffs）
• 生态：Zed、IBM 主导，已被部分编辑器采纳
• 局限：当前仅支持本地 stdio 模式，远端调用能力仍处于 RFC 意见收集阶段

A2A — Agent-to-Agent Protocol（Agent 间对等协作）

由 Google 于 2025 年 4 月发布，同年 12 月捐赠给 Linux 基金会，2026 年 3 月由新成立的 Agent API Initiative (AAIF) 接管治理。定位为 独立 Agent 系统之间的对等协作协议。核心理念是让 Agent 作为"对等方"（peer）直接通信，而非被当作工具调用。

• 适用场景：跨组织 Agent 编排、多 Agent 链式调用、长期异步任务协作
• 核心特性：
  • Agent Card 发现：Agent 通过 AgentCard 暴露能力、技能、认证方式
  • Task 生命周期：支持异步长任务，状态包括 working、completed、failed、canceled、input_required
  • 企业级安全：协议规范定义认证、授权、可观测性标准
  • 多模态交互：协议规范支持文本、文件、JSON 结构化数据
• 生态：截至 2026 年 7 月，已有 150+ 组织参与，14 个官方 SDK 覆盖主流语言

---

MCP 2026-07-28 新规范：核心收益

2026 年 7 月 28 日，MCP 发布了迄今为止最大的一次协议修订（Release Candidate）。用通俗的话说，这次更新解决了四个问题：

1. 更易扩展（从"只能官方开发"到"社区都能开发"）

之前 MCP 协议是固定的，所有功能都是官方定义的。现在建立了 Extensions 框架，允许第三方开发扩展功能，就像浏览器的插件系统。

Extensions 能扩展什么？

举个例子：

• 之前：MCP 工具只能返回文本或 JSON 数据。比如你有一个"数据可视化"工具，它只能返回 {"chart": "sales_data"}，然后 LLM 用文字描述给用户："这是一个销售数据图表..."
• 现在：通过 MCP Apps 扩展，工具可以直接返回一个可交互的 HTML 页面（在 LLM 界面里以 iframe 形式显示）。用户可以在聊天界面里直接看到一个可操作的图表，能缩放、筛选、点击查看详情

MCP Apps 是什么？

MCP Apps 是一套协议规范 + 配套 SDK，不是单独的产品。它定义了：

• MCP Server 如何返回 UI 组件
• Host（Claude/VS Code/ChatGPT）如何渲染这些 UI
• UI 如何与 Host 通信

工作流程：

1. MCP Server 注册一个 HTML 资源（如 ui://chart.html）
2. 工具被调用时，返回数据 + UI 资源引用
3. Host（Claude/VS Code）获取 HTML，在 iframe 里渲染
4. iframe 里的 UI 通过 postMessage 与 Host 通信

代码示例：

# 之前：工具只返回数据
@tool
def sales_chart():
    return {"data": [...], "type": "bar"}
# 用户在聊天里看到："这是一个柱状图..."（纯文字描述）

# 现在：工具返回数据 + UI
@tool
def sales_chart():
    return {
        "data": [...],
        "ui": "ui://chart.html"  # 引用一个 HTML 资源
    }
# 用户在聊天里直接看到一个可交互的图表（iframe 渲染）

实际场景：

• 数据可视化工具：返回可操作的图表（ECharts/D3.js），用户可以在聊天界面里缩放、筛选
• 表单工具：返回一个填写表单，用户直接在 LLM 界面里输入信息
• 地图工具：返回可缩放的地图，标记出关键位置
• 代码编辑器：返回一个可编辑的代码片段，用户修改后提交

类比理解：

• MCP Apps 就像 React（规范 + SDK），不是单独的产品
• 它让 MCP 工具可以"画界面"，而不只是返回数据

其他扩展：

• Tasks：长时间任务可以异步执行。比如数据库查询要跑 5 分钟，不用一直等着，可以提交任务后去做别的，完成了再来看结果
• 未来可能的扩展：审批流、工作流引擎、特定行业的协议（如医疗 HL7、金融 FIX）等

无状态协议：之前每次调用都要保持连接（像打电话），现在每次调用独立（像发短信），服务器可以轻松扩容

2. 更安全（从"验证 Token 有效"到"验证 Token 是谁发的"）

之前只检查"你的身份证是不是真的"，现在还要检查"身份证是哪个公安局发的"：

OAuth 2.0/OIDC 是什么？

• OAuth 2.0（授权框架）：像"授权书"。你授权第三方应用访问你的资源，但不需要把密码给它。比如你授权微信读取通讯录，但不需要把邮箱密码给微信
• OIDC（OpenID Connect，身份认证协议）：基于 OAuth 2.0，用于验证"你是谁"。像"身份证验证"，确认用户身份
• 简单说：OAuth 2.0 解决"能做什么"（授权），OIDC 解决"是谁"（认证）

• 统一认证标准：所有 MCP Server 都用 OAuth 2.0/OIDC 标准，不用每家自己实现一套
• 验证 Token 签发方：强制检查 Token 的 issuer（签发方），防止有人伪造认证。比如你的系统只信任"公司 SSO"签发的 Token，其他来源的 Token 即使有效也会被拒绝

3. 更强大（从"能用"到"好用"）

• 参数定义更灵活：完整支持 JSON Schema 2020-12，工具的参数可以定义得更复杂。比如之前只能定义"这个参数是字符串"，现在可以定义"这个参数是枚举值，只能是 A/B/C 之一"
• 调用链可追踪：多个 Agent 协作时（A 调用 B，B 调用 C），所有系统用统一的追踪格式，Langfuse/Jaeger 等工具可以自动把整个调用链串起来，一眼看清请求经过了哪些系统

4. 治理转型（从"Anthropic 一家说了算"到"大家一起商量"）

• 捐赠给 AAIF：Anthropic 把 MCP 捐给了 Agent API Initiative（AAIF），和 A2A 在同一个组织下管理
• 多方共同治理：成立 MCP 工作组，由 Anthropic、Google、Microsoft 等公司共同决策。这意味着 MCP 不再是 Anthropic 的"私产"，而是行业标准

---

MCP Tasks 扩展 vs A2A Task：区别在哪？

MCP 在 2025 年 11 月以实验特性形式发布了 Tasks 扩展，2026 年 7 月的 Release Candidate 中正式纳入规范。表面上看，MCP Tasks 和 A2A Task 都涉及"任务生命周期"，但二者解决的是不同层次的问题：

维度	MCP Tasks 扩展	A2A Task
角色关系	LLM 调用工具（主从）	Agent 间对等协作
任务发起	LLM 调用 tools/call，服务器决定是否作为任务执行	Agent 直接发送消息，任务由协议层管理
任务控制	客户端通过 tasks/get、tasks/update、tasks/cancel 推动	Agent 通过协议原生支持任务状态轮询、推送通知
上下文传递	工具调用无上下文传递，每次调用独立	支持上下文在 Agent 间传递，保持协作连续性
多模态	支持文本、图片、音频等（通过 Content 类型）	支持文本、文件、JSON 结构化数据（通过 Part 类型）
适用场景	单个工具的异步执行（如长时间数据库查询）	多个 Agent 的协作编排（如 Planner → Analyst → Reporter）

关键区别：MCP Tasks 是"工具调用的异步化"，仍然是 LLM 调用工具的模式；A2A Task 是"Agent 间的协作协议"，支持 Agent 直接通信、协商、传递中间结果。

多模态支持：

• MCP：支持文本、图片、音频等多模态内容（通过 Content 类型的 text、image、audio 等字段）
• A2A：支持文本、文件、JSON 结构化数据（通过 Part 类型的 text、file、data 等字段）

二者都支持多模态，但侧重点不同：MCP 更偏向 LLM 可处理的内容类型（图片、音频），A2A 更偏向 Agent 间传递的结构化数据（文件、JSON）。

---

A2A 的核心价值（MCP 新规范后的真实定位）

MCP 2026-07-28 新规范发布后，A2A 的部分价值被 MCP 新特性覆盖：

能力	MCP 新规范前	MCP 新规范后
异步长任务	❌ 不支持	✅ Tasks 扩展支持
交互式 UI	❌ 不支持	✅ MCP Apps 支持
分布式追踪	❌ 各自实现	✅ 标准化 W3C Trace Context
水平扩展	❌ 有状态	✅ 无状态协议

A2A 的"独家优势"只剩：

1. 对等协作模式（核心差异）

   • MCP：LLM 作为中心协调者，所有决策和信息传递都经过 LLM（中心辐射）
   • A2A：Agent 之间直接通信，不需要 LLM 作为中间人（对等网络）

2. 上下文自动传递

   • MCP：每次工具调用独立，中间结果需要 LLM 手动传递
   • A2A：上下文在 Agent 间自动流转，无需 LLM 介入

3. input_required 状态

   • A2A 支持 Agent 在需要用户输入时暂停任务，等待输入后继续
   • MCP Tasks 不支持这种交互式暂停

4. Agent 级发现

   • MCP：工具级发现（tools/list），只知道工具签名
   • A2A：Agent 级发现（AgentCard），知道完整能力、认证方式、端点

场景对比：

场景：用户要求"分析销售数据并生成报告"

【MCP 模式 - LLM 是"大脑"】
用户 → LLM（协调者）
  ├─ 调用 Analyst Agent（工具）→ 返回分析结果
  ├─ LLM 处理中间结果（LLM 要理解并传递）
  ├─ 调用 Reporter Agent（工具）→ 传入分析结果 → 返回报告
  └─ LLM 返回最终报告给用户

【A2A 模式 - Agent 自主协作】
用户 → Planner Agent
  ├─ Planner 直接发消息给 Analyst Agent（Planner 不介入）
  ├─ Analyst 完成后直接发消息给 Reporter Agent（上下文自动传递）
  └─ Reporter 完成后返回结果给 Planner → 返回给用户

什么时候用 MCP，什么时候用 A2A？

• 用 MCP：你的场景是"LLM 调用几个工具完成任务"，比如查询数据库、调用 API、生成报告
• 用 A2A：你的场景是"多个独立 Agent 需要自主协作"，比如跨组织的复杂工作流、需要 Agent 间直接协商的场景

---

A2A 生态采用情况

截至 2026 年 7 月，A2A 协议的生态正在快速成长：

• 治理：Linux 基金会托管，Apache 2.0 许可证，确保厂商中立
• 里程碑：2025 年 4 月 Google 发布 → 2025 年 12 月捐赠 Linux 基金会 → 2026 年 3 月 AAIF 成立统一治理
• 参与者：150+ 组织，包括 Salesforce、SAP、Atlassian、Box、ServiceNow、Cisco、LangChain、CrewAI 等
• SDK：14 种语言/平台（Python、Go、JavaScript/TypeScript、Java、.NET、Rust、C++、PHP、Dart、Kotlin、Ruby、Swift 等）
• 教育：DeepLearning.AI 推出 A2A 专项课程

---

OpenClaw 的协议支持情况

OpenClaw 是由 OpenClaw 基金会维护的开源个人 AI 助手，支持 25+ 消息通道和多模型提供商。

当前协议支持：

• MCP：已原生支持，通过 MCP Registry 集成工具生态
• ACP：已支持，作为编辑器/IDE Agent 集成的标准协议
• A2A：尚未原生支持，但社区有强烈需求（GitHub Issues 中已有多个讨论帖呼吁支持 A2A）

---

基于 LangGraph + FastAPI 的 AI Agent API 服务端 SDK，为 WISE-PaaS / EnSAAS 平台提供完整的 Agent 服务构建能力。支持 A2A (Agent-to-Agent)、MCP (Model Context Protocol) 和 ACP (Agent Client Protocol) 三种标准协议，实现跨系统、跨组织的 Agent 互操作。

核心能力

• 三协议支持 — 同时支持 A2A、MCP 和 ACP 协议，Agent 可被任意兼容标准的客户端发现和调用
• 编辑器集成 — 通过 ACP 协议将 Agent 暴露为 VS Code / Zed / JetBrains 等编辑器的 AI 助手，支持实时流式对话
• 远程 Agent 链式调用 — 通过 A2A 协议实现多 Agent 跨服务编排，支持 Agent A → Agent B → Agent C 的链式协作
• 动态 LLM 调用 — 支持 CHAT / EMBEDDING / RERANK 多种模型类型，按租户动态切换
• 对话记忆 — 基于 PostgreSQL 的 LangGraph checkpoint 持久化
• MCP 工具暴露 — 将 LangGraph Agent 自动转换为 MCP 工具，供 Claude/GPT 等 LLM 直接调用
• 多租户隔离 — 支持按租户隔离模型配置和认证

---

系统架构

分层说明

Client Layer（客户端层）

• Web App：通过 REST API 调用 Agent
• LLM App (Claude/GPT)：通过 MCP 协议将 Agent 作为工具调用
• External A2A Agent：通过 A2A 协议发现和调用本系统的 Agent
• Editor (VS Code/Zed/JetBrains)：通过 ACP 协议将 Agent 作为编辑器 AI 助手调用

ACP Bridge（ACP 桥接层）

• agenthub-acp CLI：轻量级 stdio 代理进程，将 ACP 协议转换为 HTTP SSE 请求
• 无需加载 LangGraph 或连接数据库，仅需 HTTP 访问 AgentHub FastAPI 服务
• 支持 --service-url、--agent-name、--extra-inputs 等参数配置

Protocol Layer（协议层）

• REST API (FastAPI)：标准的 HTTP RESTful 接口
• A2A Protocol (JSON-RPC)：Agent 间互操作协议
• MCP Server (Streamable HTTP)：Model Context Protocol 服务端
• ACP Bridge (stdio JSON-RPC)：Agent Client Protocol 编辑器集成协议

Core Layer（核心层）

• API Router & Thread Mgmt：路由分发和会话线程管理
• Graph Loader (LangGraph)：动态加载 LangGraph Agent
• DynamicLLM (Multi-Tenant)：多租户动态 LLM 调用
• AgentCard Builder：构建 A2A AgentCard，描述 Agent 能力
• MCP Protocol Converter：将 LangGraph Agent 转换为 MCP 工具
• A2A Executor (LangGraph)：A2A 协议执行器
• Agent Registry：向 Model Manager 注册 Agent

Service Layer（服务层）

• config_builder：统一构建 LangGraph configurable 和 Langfuse 回调，消除 REST/MCP/A2A 三入口重复代码
• stream_service：统一图执行 + SSE 流式输出 + JWT user_id 提取

Infrastructure（基础设施）

• PostgreSQL：LangGraph checkpoint 持久化（advisory lock 保证多 worker schema 迁移安全）
• Redis：线程缓存、消息发布订阅、A2A TaskStore（多 worker 共享任务状态）
• Model Manager Service：模型配置管理
• SSO Auth：单点登录认证

Multi-Worker Safety（多 worker 安全）

• PostgreSQL advisory lock：schema 迁移仅由一个 worker 执行，其余等待
• RedisTaskStore：A2A 任务状态存储在 Redis，跨 worker 共享
• DynamicLLM async lock：模型初始化使用 asyncio.Lock，避免并发重复初始化
• 幂等启动：ApplicationStartup._initialized 标志防止 lifespan 重复初始化

---

A2A 协议支持

AgentHub SDK 完整实现了 A2A (Agent-to-Agent) 协议，使你的 Agent 可以被其他 A2A 兼容系统发现和调用。

A2A 协议流程

① Discovery（发现阶段）

• 调用方通过 GET /.well-known/agent-card.json 获取 AgentCard
• AgentCard 包含 Agent 的名称、描述、技能列表、输入输出格式等元数据
• 每个 skill 的 description 中包含详细的参数说明和调用示例

② Invocation（调用阶段）

• 调用方根据 AgentCard 中的 skill schema 构建请求
• 通过 POST /a2a (JSON-RPC) 发送消息
• metadata 字段包含 graph_name（目标 Agent）和所有业务参数

③ Execution（执行阶段）

• LangGraphAgentExecutor 根据 metadata.graph_name 路由到对应的 LangGraph Agent
• 将 metadata 中的参数传递给 Agent 执行
• 当前实现为非流式模式（通过 Task artifacts 返回结果）

④ Response（响应阶段）

• 执行结果通过 Task 的 artifacts 返回
• 包含 Agent 的最终输出文本

远程 Agent 链式调用

A2A 协议的核心价值在于实现跨服务的 Agent 链式调用。以下是一个典型场景：

场景说明：

用户向 AgentHub A 的 Planner Agent 发送请求："分析销售数据并生成报告"

① 任务分解

• Planner Agent 分析用户需求，决定需要调用 Data Analyst Agent 和 Report Agent
• 通过 A2A 协议发现 AgentHub B 的 Analyst Agent

② 第一次链式调用

• Planner Agent 作为 A2A Client，向 AgentHub B 发送调用请求
• metadata 中设置 graph_name: "analyst"，并传递数据分析所需的参数
• Analyst Agent 执行数据分析，返回结果

③ 第二次链式调用

• Planner Agent 将分析结果传递给 AgentHub C 的 Report Agent
• metadata 中设置 graph_name: "reporter"，并传递分析结果
• Report Agent 生成最终报告

④ 结果汇总

• Planner Agent 收到报告，整合所有结果，返回给用户

关键技术点：

1. AgentCard 驱动：每个 AgentHub 实例通过 AgentCard 暴露自己的能力，调用方通过 AgentCard 了解可用的 Agent 和参数要求
2. metadata 路由：通过 metadata.graph_name 实现多 Agent 路由，单个 A2A 端点可以服务多个 Agent
3. 参数透传：业务参数通过 metadata 直接传递，无需嵌套在 input 或 params 中
4. 异步执行：A2A 协议支持异步任务，长时间运行的 Agent 不会阻塞调用方

启用 A2A

在 .env 中设置：

ENABLE_A2A=true

其他配置会自动从 langgraph.json 推导：

配置项	默认值	来源
ENABLE_A2A	false	唯一需要显式配置的开关
Agent 名称	自动读取	langgraph.json 的 graph 名称
Agent 描述	自动读取	langgraph.json 的 agent_description
Agent URL	复用现有	SERVICE_EXTERNAL_URL

A2A 端点

启用后，以下端点自动可用：

• GET /.well-known/agent-card.json — Root AgentCard，描述所有可用 Agent
• GET /.well-known/agent-card/{agent_name}.json — 单个 Agent 的 AgentCard
• POST /a2a — JSON-RPC 端点
• POST /message:send — REST 发送消息
• POST /message:stream — REST 流式消息
• GET /tasks/{id} — 获取任务状态

Agent 间调用示例

from agent_api_server.a2a_bridge.call_a2a_client import AgentHubClient

# 调用远程 Agent
client = AgentHubClient("http://other-agent-hub:8080")
result = await client.call(
    raw_question="分析今天的销售数据",
    graph_name="DataAnalyst-Agent",
    tenant_id="tenant_123",
    date="2025-01-15",
)
print(result)

---

MCP 协议支持

AgentHub SDK 支持 MCP (Model Context Protocol) 协议，将你的 LangGraph Agent 暴露为 MCP 工具，供 Claude、GPT、IDE 等 LLM 客户端直接调用。

MCP 协议流程

① Tool Discovery（工具发现）

• LLM Host (Claude/GPT/IDE) 通过 tools/list 发现可用的 Agent 工具
• MCP Server 返回所有注册的 Agent，每个 Agent 作为一个 tool
• tool 的 name 是 graph_name，description 是 Agent 描述，inputSchema 是 Agent 的输入参数 schema

② Tool Invocation（工具调用）

• LLM 决定调用某个 Agent 工具
• 通过 tools/call 发送调用请求，包含 name（Agent 名称）和 arguments（参数）

③ Execution（执行阶段）

• MCP Server 根据 name 路由到对应的 LangGraph Agent
• 将 arguments 作为输入参数传递给 Agent
• Agent 执行并流式返回结果

④ Response（响应阶段）

• 执行结果通过 CallToolResult 返回给 LLM
• 包含 Agent 的最终输出文本

MCP 实现细节

独立进程运行

• MCP Server 运行在独立进程中，与主 FastAPI 服务隔离
• 通过 multiprocessing.Process 启动
• 使用 streamable-http 传输协议

自动工具转换

• create_mcp_tool_from_agent() 自动将 LangGraph Agent 转换为 MCP 工具
• 从 graph_instance.get_input_jsonschema() 提取参数 schema
• 动态生成 tool 函数签名，包括参数类型和默认值

流式支持

• Agent 执行过程中的中间状态通过 get_stream_writer() 实时推送
• 最终结果通过 CallToolResult 返回

启用 MCP

在 .env 中设置：

ENABLE_MCP_SERVER=true
MCP_SERVER_PORT=8081

MCP Server 会自动启动在指定端口，传输协议为 streamable-http。

MCP 客户端调用示例

from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

async def call_agent_via_mcp():
    url = "http://localhost:8081/mcp"
    
    async with streamablehttp_client(url) as (read, write, _):
        async with ClientSession(read, write) as session:
            await session.initialize()
            
            # 列出可用工具
            tools = await session.list_tools()
            print(f"Available agents: {[t.name for t in tools.tools]}")
            
            # 调用 Agent
            result = await session.call_tool(
                "DataAnalyst-Agent",
                arguments={"query": "分析销售数据", "date": "2025-01-15"}
            )
            print(result.content[0].text)

---

ACP 协议支持

AgentHub SDK 支持 ACP (Agent Client Protocol) 协议，通过轻量级 CLI 代理将你的 LangGraph Agent 暴露为 VS Code、Zed、JetBrains 等编辑器的 AI 助手。

ACP 协议架构

工作原理

ACP 采用代理模式（Proxy Mode），CLI 进程作为编辑器与 FastAPI 服务之间的桥梁：

┌─────────────────────────────────────────────────────────────┐
│  Docker Container (port 8087)                               │
│  ┌───────────────────────────────────────────────────────┐  │
│  │  FastAPI Service                                       │  │
│  │  - LangGraph agents                                   │  │
│  │  - Postgres, Redis                                    │  │
│  │  - Model Manager                                      │  │
│  └───────────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────────┘
                          ↑ HTTP SSE
                          │
┌─────────────────────────────────────────────────────────────┐
│  External Host (Developer Machine)                          │
│  ┌───────────────────────────────────────────────────────┐  │
│  │  ACP CLI (stdio process)                              │  │
│  │  agenthub-acp                                         │  │
│  │    --service-url http://docker-host:8087              │  │
│  │    --agent-name DataInsight-Agent-Local               │  │
│  │    --user-input-field user_input                      │  │
│  │    --extra-inputs '{"app_id":"aJ1nQnxvreAg"}'         │  │
│  └───────────────────────────────────────────────────────┘  │
│                          ↑ stdio JSON-RPC                   │
│  ┌───────────────────────────────────────────────────────  │
│  │  VS Code / Zed Editor                                 │  │
│  │  ACP Client Extension                                 │  │
│  └───────────────────────────────────────────────────────┘  │
└─────────────────────────────────────────────────────────────┘

协议流程

① 编辑器启动 ACP CLI

• 编辑器通过 acp.agents 配置找到 CLI 命令
• 以 stdio 子进程方式启动 agenthub-acp
• CLI 通过 JSON-RPC over stdio 与编辑器通信

② 会话创建

• 编辑器发送 new_session 请求
• CLI 生成 UUID，通过 HTTP GET/POST 在 FastAPI 创建 thread
• 建立 ACP session → FastAPI thread 的映射关系

③ 用户发送消息

• 编辑器发送 prompt 请求，包含用户输入文本
• CLI 将输入合并 extra-inputs，POST 到 /api/v1/thread/{id}/stream
• FastAPI 启动 LangGraph Agent 执行

** 实时流式响应**

• FastAPI 返回 SSE 事件流（token_stream, agent_message, tools_message 等）
• CLI 将 SSE 事件转换为 ACP session_update 通知
• 编辑器实时展示 Agent 的思考和执行过程

CLI 参数说明

参数	环境变量	必填	说明
--service-url	AGENTHUB_SERVICE_URL	否	FastAPI 服务地址，默认 http://localhost:8080
--agent-name	AGENTHUB_AGENT_NAME	是	Agent 名称（graph_name）
--user-input-field	AGENTHUB_USER_INPUT_FIELD	否	用户输入字段名，默认 user_input
--extra-inputs	AGENTHUB_EXTRA_INPUTS	否	额外输入参数 JSON，如 {"app_id": "xxx"}
--ts-tenant	AGENTHUB_TS_TENANT	否	TSTenant 认证值
--ei-token	AGENTHUB_EI_TOKEN	否	EIToken 认证值
--agent-version	AGENTHUB_AGENT_VERSION	否	Agent 版本号，默认 0.1.0

安装与使用

1. 安装 SDK

# 从 PyPI 安装
pip install agent-api-server

# 或从本地 wheel 安装
pip install dist/agent_api_server-2.1.9-py3-none-any.whl

安装后 agenthub-acp 命令自动注册到 PATH。

2. 启动 AgentHub 服务

确保你的 AgentHub FastAPI 服务已在 Docker 或本地运行（默认端口 8087）。

3. 配置编辑器

在 VS Code 的 settings.json 中添加：

{
  "acp.agents": {
    "DataInsight-Agent": {
      "command": "agenthub-acp",
      "args": [
        "--service-url", "http://127.0.0.1:8087",
        "--agent-name", "DataInsight-Agent-Local",
        "--user-input-field", "user_input",
        "--extra-inputs", "{\"app_id\": \"aJ1nQnxvreAg\", \"selected_entities\": \"\"}"
      ]
    }
  }
}

4. 使用

• 在 VS Code 中打开 ACP 面板（ACP Client 扩展）
• 选择配置好的 Agent
• 直接输入问题，Agent 将实时流式响应

启用 ACP

ACP Bridge 作为独立 CLI 工具运行，无需在 .env 中额外配置：

代理模式 vs 独立模式

特性	代理模式（当前）	独立模式（未来）
进程位置	外部主机	与 FastAPI 同进程
依赖	仅需 HTTP 访问	需加载 LangGraph
部署	CLI 安装到外部主机	无需额外部署
适用场景	Docker 部署、远程服务	本地开发、单体部署

当前实现采用代理模式，CLI 是一个轻量级进程，不加载 LangGraph 或连接数据库，仅通过 HTTP 与已运行的 FastAPI 服务通信。这使得 Agent 可以部署在 Docker 容器中，而开发者在外部主机通过 CLI 调用。

Roadmap

阶段	状态	说明
本地代理模式	✅ 已完成	CLI 作为 stdio 代理，通过 HTTP SSE 调用本地/容器内 FastAPI 服务
远端 ACP 调用	🔲 规划中	ACP 协议官方目前仍在意见收集阶段（ACP Remote Agent RFC），远端调用能力尚未标准化，暂不支持
远端适配	待定	待 ACP 远端协议稳定后，再评估并适配远程 Agent 调用场景

说明：当前 ACP 协议仅支持本地 stdio 模式，远端调用能力官方仍在意见收集阶段。我们暂不考虑支持远端模式，待协议稳定后再进行适配。

ACP 与 Chatbot 服务：架构演进

ACP 的核心价值：标准化用户与 Agent 的交互入口

ACP 让 Agent 在交互方式上，提供了统一的标准化管理和入口，包括：

• 历史对话管理（通过 session/thread 管理）
• 流式响应（实时展示 Agent 思考过程）
• 多 Agent 切换（用户可以选择不同的 Agent）

当前架构：需要中间 Chatbot 服务

【当前架构】
用户 → chatbot UI → Chatbot API（中间服务）→ Agent API
                    ↓
              - 对话管理
              - 历史存储
              - Agent 路由

Chatbot 服务承担了：

• 用户界面（Web UI）
• 对话管理（session/thread）
• 历史存储（数据库）
• Agent 路由（调用不同的 Agent）

未来架构：ACP 远程调用标准化后

当 ACP 远程调用协议稳定后，编辑器可以直接通过 ACP 调用远程 Agent，不再需要中间的 Chatbot 服务：

【未来架构】
用户 → chatbot UI → ACP 远程调用 → Agent
                                  ↓
                         - 对话管理（Agent 端）
                         - 历史存储（Agent 端）
                          - 流式响应（ACP 原生支持）

架构对比：

维度	当前架构（需要 Chatbot）	未来架构（ACP 远程调用）
中间服务	需要 Chatbot 服务	不需要，直接调用 Agent
对话管理	Chatbot 服务管理	Agent 端管理
历史存储	Chatbot 服务存储	Agent 端存储
部署复杂度	需要部署 Chatbot 服务	只需部署 Agent
延迟	多一跳（Chatbot 转发）	直接调用，延迟更低
适用场景	当前 ACP 仅支持本地	未来 ACP 支持远程调用

---

NemoClaw + OpenClaw 整合架构

AgentHub SDK 支持与 NemoClaw 和 OpenClaw 整合，实现每用户沙箱调度 + 边缘 Agent 部署的架构。

架构图

架构说明

Step 1: User Sandbox（用户沙箱）

• OpenClaw：部署在每个用户的 NemoClaw 沙箱中，作为调度大脑（Orchestrator）
• AgentHub ACP CLI：轻量级代理进程，也在沙箱内，负责将 OpenClaw 的 ACP 调用转发到边缘侧 Agent
• User Config：每用户配置自己的凭证和设置，包含：
  • OpenClaw API Key：用于 OpenClaw 调度器的认证
  • EI Token / 服务 API Key：用于调用 MCP Gateway 时的认证
  • 用户特定的设置和偏好

User Config 详细说明

User Config 包含以下凭证：

1. OpenClaw API Key

   • 用途：认证 OpenClaw 调度器
   • 位置：存储在用户沙箱（NemoClaw）中
   • 作用：允许 OpenClaw 调用 AgentHub ACP CLI

2. EI Token / 服务 API Key

   • 用途：MCP Gateway 认证
   • 位置：通过 ACP CLI 传递给 Edge Agent
   • 作用：Agent 调用 MCP Tool 时携带此 Token，MCP Gateway 通过 SSO 验证

3. 用户设置

   • 用途：个性化配置
   • 示例：默认 Agent 选择、输出格式偏好等

凭证流转：

User Config
  ├─ OpenClaw API Key → OpenClaw 调度器认证
  └─ EI Token → ACP CLI → Edge Agent → MCP Gateway → SSO 验证

Step 2: Edge Agents（边缘 Agent）

• Query Agent：LangGraph 实现的数据查询 Agent，负责数据检索
• Analysis Agent：LangGraph 实现的数据分析 Agent，负责洞察生成
• Agent 是系统的"手和脚"，部署在边缘侧，通过 ACP HTTP SSE 接收调用

Step 3: Gateway & Authentication（网关与认证）

• MCP Gateway：接收 Agent 的 MCP Tool 调用，结合 SSO 进行 Token 认证
• SSO Service：验证 Token 有效性
• Model Manager：多租户模型密钥管理，动态分发 Key 给下游 Agent

Step 4: MCP Servers（下游工具）

• Database MCP：SQL 查询工具
• Analytics MCP：数据分析工具
• Visualization MCP：图表生成工具
• Agent 通过 MCP Tool 调用（而非 Agent-to-Agent）执行具体任务

Step 5: Infrastructure（基础设施）

• PostgreSQL：LangGraph Checkpoint 持久化
• Redis：缓存和 PubSub
• Langfuse：调用追踪
• Model Registry：模型密钥存储

核心流程

OpenClaw → ACP CLI → Edge Agent → MCP Gateway → SSO Auth → MCP Server

1. OpenClaw 调度：用户在沙箱中发起任务，OpenClaw 作为调度大脑决定调用哪个 Agent
2. ACP 代理转发：AgentHub ACP CLI 通过 stdio 接收 OpenClaw 调用，通过 HTTP SSE 转发到边缘 Agent
3. Agent 执行：边缘 Agent（LangGraph）执行任务，需要调用 MCP Tool 时携带 Token
4. 网关认证：MCP Gateway 接收调用，通过 SSO Service 验证 Token
5. 工具执行：认证通过后，Gateway 代理调用下游 MCP Server 执行具体任务

最佳实践

Agent 暴露为 MCP Tool（而非 Agent-to-Agent 调用）

• ✅ 推荐：Agent 通过 MCP 协议暴露为 Tool，由 LLM 决定何时调用
• 不推荐：Agent 之间直接调用（Agent-to-Agent）

原因：

• 更符合 LLM 工具使用范式（Tool-use paradigm）
• 更好的可组合性和发现性（Composability & Discovery）
• MCP Gateway 统一认证和代理，简化安全模型

多租户模型管理

Model Manager 核心作用：

1. 租户密钥分发：为每个租户分发独立的模型 API Key（OpenAI / Azure / 私有模型）
2. 运行时获取：Agent 的 DynamicLLM 在运行时从 Model Manager 获取密钥，无需硬编码
3. 动态切换：支持按租户动态切换模型，实现多租户隔离

架构图中的体现：

在架构图中，Model Manager（Step 3）通过紫色箭头向两个 Edge Agent 的 DynamicLLM 分发模型密钥：

Model Manager --[Model Keys Distribution]--> Query Agent.DynamicLLM
Model Manager --[Model Keys Distribution]--> Analysis Agent.DynamicLLM

代码示例：

from agent_api_server.shared.model_config import get_env
from agent_api_server.dynamic_llm.dynamic_llm import DynamicLLM, ConfigType

# 根据租户 ID 获取模型配置（从 Model Manager 获取密钥）
config = get_env(ts_tenant="tenant_123")

# DynamicLLM 使用租户特定的模型配置
llm = DynamicLLM(tool_name="default", config_type=ConfigType.CHAT)
response = llm.invoke(messages, config=config)

工作流程：

1. Agent 执行时需要调用 LLM
2. DynamicLLM 从 config 中获取 ts_tenant（租户 ID）
3. get_env() 查询 Model Manager，获取该租户的模型 API Key
4. DynamicLLM 使用获取的 Key 调用对应的模型服务
5. 不同租户使用不同的 Key，实现隔离

部署模式

组件	部署位置	说明
OpenClaw	用户沙箱（NemoClaw）	每用户独立实例
AgentHub ACP CLI	用户沙箱（NemoClaw）	轻量级，无状态
Edge Agents	边缘侧服务器	LangGraph Agent，可水平扩展
MCP Gateway	中心化服务	SSO 认证 + MCP 代理
MCP Servers	中心化服务	下游工具服务
Model Manager	中心化服务	多租户密钥管理

---

项目结构

agent_api_server/
├── app.py                     # FastAPI 应用工厂、lifespan 和配置
├── startup.py                 # 应用启动编排（SSO/注册/MCP，幂等初始化）
├── agent_registration.py      # Agent 注册和监听器管理
├── mcp_server.py              # MCP Server 管理（独立进程模式）
├── service.py                 # 向后兼容接口
├── services/                  # 统一服务层（消除 REST/MCP/A2A 重复代码）
│   ├── config_builder.py      # 构建 configurable + Langfuse 回调
│   └── stream_service.py      # 图执行 + SSE 流式输出 + user_id 提取
├── a2a_bridge/                # A2A 协议桥接
│   ├── agent_card_builder.py  # AgentCard 构建器
│   ├── langgraph_executor.py  # LangGraph 执行器适配器
│   ├── server.py              # A2A 服务集成
│   ├── call_a2a_client.py     # A2A 客户端（调用其他 Agent）
│   └── redis_task_store.py    # Redis TaskStore（多 worker 安全）
├── acp_bridge/                # ACP 协议桥接
│   ├── acp_agent.py           # ACP Agent 实现（stdio 代理）
│   └── acp_server.py          # CLI 入口和参数解析
├── listener/                  # 消息监听器
│   ├── base.py                # 基类和工具
│   ├── redis_listener.py      # Redis 监听器
│   └── nats_listener.py       # NATS 监听器
├── api/v1/                    # API 路由
│   ├── thread.py              # 线程管理 API
│   ├── graph.py               # Graph 查询 API
│   ├── schema.py              # JSON Schema API
│   └── config.py              # 配置 API
├── cache/                     # Redis 缓存
├── config_center/             # 配置中心（httpx 异步客户端）
├── configs/                   # 配置管理
├── dynamic_llm/               # 动态 LLM 调用层（async lock 安全）
├── memory/                    # 对话记忆 (PostgreSQL, advisory lock)
├── mcp_convert/               # MCP 协议转换
├── register/                  # Agent 注册中心
├── service_hub/               # 凭证管理
├── sso_service/               # SSO 认证
└── shared/                    # 共享工具
    ├── graph_loader.py        # Graph 加载
    ├── model_config.py        # 模型配置管理
    ├── decode_token.py        # JWT 解码
    └── exceptions.py          # 自定义异常

---

快速开始

安装

pip install agent-api-server

# 如需 A2A 协议支持
pip install agent-api-server[a2a]

配置

创建 .env 文件或设置环境变量：

# SSO 配置
SSO_URL=http://sso/v4.0
CLIENT_ID=your_client_id
CLIENT_SECRET=your_client_secret

# 数据库
POSTGRES_URL=postgresql://user:password@host:port/dbname
REDIS_URL=redis://localhost:6379/0

# 模型管理
MODEL_MANAGER_SERVICE_URL=https://api-am-ensaas.axa.wise-paas.com.cn
SERVICE_EXTERNAL_URL=http://your-agent-url:8080

# 消息队列 (二选一)
MODEL_MANAGER_REDIS_URL=redis://localhost:6379/0
# MODEL_MANAGER_NATS_URL=nats://localhost:4222

# 服务配置
SERVER_PORT=8080
ENABLE_MCP_SERVER=false
MCP_SERVER_PORT=8081
AGENT_AUTO_REGISTRATION=false

# A2A 协议 (可选)
ENABLE_A2A=false

定义 Agent

创建 langgraph.json 配置文件：

{
  "graphs": {
    "my_agent": "path/to/my_agent.py:graph"
  },
  "agent_description": {
    "my_agent": "My AI Agent"
  },
  "agent_api_version": "v1.0.0"
}

启动服务

from agent_api_server.app import create_fastapi_app
import uvicorn

app = create_fastapi_app()

if __name__ == "__main__":
    uvicorn.run("agent_api_server.app:app", host="0.0.0.0", port=8080, reload=True)

或者使用 demo.py 中提供的启动方式：

from agent_api_server.app import create_fastapi_app
from agent_api_server.startup import ApplicationStartup
import uvicorn

app = create_fastapi_app()

@app.on_event("startup")
async def startup():
    startup_service = ApplicationStartup()
    await startup_service.initialize()

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=8080)

注意：A2A 协议在 FastAPI 的 lifespan 中自动初始化（参见 app.py），无需手动调用。

或直接运行 demo.py：

python demo.py

---

模块说明

DynamicLLM

动态 LLM 调用层，支持按租户动态切换模型配置：

from agent_api_server.dynamic_llm.dynamic_llm import DynamicLLM
from llm_sdk.model_providers.base import ConfigType

# CHAT 模型
llm = DynamicLLM(tool_name="default", config_type=ConfigType.CHAT)
response = llm.invoke(messages, config=llm_config)

# EMBEDDING 模型
llm = DynamicLLM(tool_name="default", config_type=ConfigType.EMBEDDING)
embedding = llm.invoke("text to embed", config=llm_config)

# RERANK 模型
llm = DynamicLLM(tool_name="default", config_type=ConfigType.RERANK)
result = llm.rerank("query", documents=["doc1", "doc2"], top_n=1)

ACP Bridge

ACP 协议桥接模块，将 LangGraph Agent 暴露为编辑器的 AI 助手：

from agent_api_server.acp_bridge.acp_agent import AgentHubACPAgent
from agent_api_server.acp_bridge.acp_server import main as acp_main

# 创建 ACP Agent（代理模式）
agent = AgentHubACPAgent(
    service_url="http://localhost:8087",
    agent_name="DataInsight-Agent-Local",
    user_input_field="user_input",
    extra_inputs={"app_id": "aJ1nQnxvreAg"},
)

# 启动 CLI（stdio 代理）
# agenthub-acp --service-url http://localhost:8087 --agent-name DataInsight-Agent-Local

核心类：

• AgentHubACPAgent：实现 ACP Agent 协议，将 ACP 请求转换为 HTTP SSE 调用
• acp_server.main()：CLI 入口，解析参数并启动 stdio 服务

工作流程：

1. 编辑器通过 stdio JSON-RPC 发送 ACP 请求
2. CLI 将请求转换为 HTTP 调用 FastAPI 的 /api/v1/thread/ 和 /stream 接口
3. SSE 事件流被转换为 ACP session_update 通知返回编辑器

消息监听器

支持 Redis 和 NATS 两种消息队列：

from agent_api_server.listener import create_listener, ListenerType

# Redis 监听器
listener = create_listener("my_agent", client_token, ListenerType.REDIS)

# NATS 监听器
listener = create_listener("my_agent", client_token, ListenerType.NATS)

listener.run()  # 阻塞运行

---

API 文档

启动服务后访问 http://localhost:8080/docs 查看 Swagger API 文档。

主要 API：

线程管理

• POST /api/v1/thread/ — 创建对话线程
• GET /api/v1/thread/ — 列出所有活跃线程
• GET /api/v1/thread/{thread_id}/status — 获取线程状态
• POST /api/v1/thread/{thread_id}/run — 执行 Agent（非流式）
• POST /api/v1/thread/{thread_id}/stream — 流式调用 Agent（SSE）
• POST /api/v1/thread/{thread_id}/stop — 停止正在运行的线程
• DELETE /api/v1/thread/{thread_id} — 删除线程

Agent 查询

• GET /api/v1/graph/ — 获取所有可用 Agent
• GET /api/v1/schema/?graph_name={name} — 获取指定 Agent 的 JSON Schema

配置

• GET /api/v1/config/ — 获取服务配置

---

开发

# 安装开发依赖
poetry install

# 运行测试
pytest

---

版本历史

• v2.1.9 — 代码结构优化，模块拆分，新增 A2A 协议支持

---

Agent 构建指南

本章节以 MetricInsight-Agent 为例，说明如何基于 AgentHub SDK 构建一个完整的 LangGraph Agent。

1. 项目结构

MetricInsight-Agent/
├── langgraph.json                    # Agent 配置文件（入口）
├── iot_data_analyse_agent/
│   ├── graph.py                      # LangGraph 工作流定义
│   ├── state.py                      # Agent 状态定义
│   ├── configuration.py              # 可配置参数定义
│   ├── agents/                       # 节点实现
│   │   ├── intent_slot_extraction_node.py
│   │   ├── metric_retrieval_node.py
│   │   ├── generate_mql_node.py
│   │   └── format_mql_node.py
│   ├── tools/                        # 工具实现
│   │   ├── intent_slot_extraction_tool/
│   │   ├── metric_retrieval_tool/
│   │   ├── generate_mql_tool/
│   │   └── time_extract_tool/
│   └── shared/                       # 共享工具
│       ├── qdrant_vector.py          # 向量数据库封装
│       ── extract_metric.py         # 指标提取工具
├── main.py                           # 服务启动入口
└── .env                              # 环境变量配置

2. 定义 langgraph.json

langgraph.json 是 Agent 的入口配置文件，AgentHub SDK 会自动读取此文件：

{
  "dependencies": ["."],
  "graphs": {
    "DataInsight-Agent-Local": "./iot_data_analyse_agent/graph.py:graph"
  },
  "agent_description": {
    "DataInsight-Agent-Local": "Intelligent Metric Query Agent - Analyzes user's natural language input, extracts relevant metric information, retrieves corresponding metrics, derives new metrics, and generates visualizations"
  },
  "agent_api_version": "v0.0.1",
  "agent_features": {
    "show_anonymous": false
  },
  "has_site": true,
  "agent_labels": ["DataInsight-Agent-Local"],
  "env": ".env"
}

关键配置项：

字段	说明	示例
graphs	Agent 图定义，格式为 {graph_name}: {file_path}:{graph_variable}	"./iot_data_analyse_agent/graph.py:graph"
agent_description	Agent 描述，会显示在 AgentCard 中	自然语言描述
agent_api_version	Agent API 版本	"v0.0.1"
agent_features	Agent 特性配置	{"show_anonymous": false}
has_site	是否有前端站点	true / false
agent_labels	Agent 标签，用于分类和搜索	["DataInsight-Agent-Local"]
env	环境变量文件路径	".env"

3. 定义 State

state.py 定义 Agent 的状态结构，包括输入状态和完整状态：

from typing import List, Dict, Any, NotRequired
from langchain_core.messages import AnyMessage
from langgraph.graph import add_messages
from pydantic import Field
from typing_extensions import Annotated, TypedDict

class InputState(TypedDict):
    """用户输入状态"""
    user_input: Annotated[
        str,
        Field(
            description="用户的原始输入文本，必填",
            examples=["今年销售额是多少？", "上个月华东区的订单量"],
        )
    ]
    
    app_id: Annotated[
        str,
        Field(
            description="所查询问题设计的业务域ID，必填",
            examples=["aJ1nQnxvreAg", "az70OY8PL9KQ"],
        )
    ]
    
    selected_entities: Annotated[
        NotRequired[str],
        Field(
            description="JSON 字符串，表示用户选中的实体列表",
            examples=['[{"type": "metric", "displayName": "销售额"}]'],
        )
    ]

class AgentState(InputState):
    """完整 Agent 状态"""
    graph_start_time: float
    graph_start_time_str: str
    query_data: list
    is_chat: str
    intent_results: List[Dict]
    metric_results: Dict
    metric_results_txt: str
    generate_mql_from_llm: List[Dict]
    format_mql_result: Dict[str, Any]
    messages: Annotated[list[AnyMessage], add_messages]

设计要点：

• InputState：定义用户输入参数，使用 Pydantic Field 添加描述和示例
• AgentState：继承 InputState，添加中间状态字段
• messages：使用 Annotated[list[AnyMessage], add_messages] 支持消息累积

4. 定义 Configuration

configuration.py 定义 Agent 的可配置参数，支持环境变量回退：

from __future__ import annotations
import os
from dataclasses import dataclass, field
from typing import Annotated, Literal
from agent_api_server.shared.common import ConfigCategory

@dataclass(kw_only=True)
class BaseConfiguration:
    CHAT_PROVIDER: str = field(
        default=os.environ.get("CHAT_PROVIDER", "openai"),
        metadata={
            "description": "默认Chat模型供应商",
            "title": ConfigCategory.CHAT_PROVIDER.value
        },
    )
    
    CHAT_MODEL: str = field(
        default=os.environ.get("CHAT_MODEL", "qwen-plus"),
        metadata={
            "description": "默认Chat模型名称",
            "title": ConfigCategory.CHAT_MODEL.value
        },
    )
    
    CHAT_CREDENTIALS: str = field(
        default=os.environ.get("CHAT_CREDENTIALS", ""),
        metadata={
            "description": "默认Chat模型的配置信息",
            "title": ConfigCategory.CHAT_CREDENTIALS.value
        },
    )
    
    EMBEDDING_PROVIDER: str = field(
        default=os.environ.get("EMBEDDING_PROVIDER", "azure"),
        metadata={
            "description": "默认Embedding模型供应商",
            "title": ConfigCategory.EMBEDDING_PROVIDER.value,
        },
    )
    
    EMBEDDING_MODEL: str = field(
        default=os.environ.get("EMBEDDING_MODEL", "text-embedding-3-small"),
        metadata={
            "description": "默认Embedding 模型名称",
            "title": ConfigCategory.EMBEDDING_MODEL.value
        },
    )
    
    QDRANT_URL: Annotated[
        str,
        field(metadata={"description": "Qdrant向量数据库URL"})
    ] = os.environ.get("QDRANT_URL", "http://localhost:6333")

设计要点：

• 使用 dataclass + field 定义配置参数
• 通过 os.environ.get() 支持环境变量回退
• metadata 中的 description 和 title 用于生成 JSON Schema

5. 定义 Graph

graph.py 定义 LangGraph 工作流：

from langgraph.constants import START, END
from langgraph.graph import StateGraph

from iot_data_analyse_agent.agents.intent_slot_extraction_node import intent_slot_extraction_node
from iot_data_analyse_agent.agents.metric_retrieval_node import metric_retrieval_node
from iot_data_analyse_agent.agents.generate_mql_node import mql_generation_node
from iot_data_analyse_agent.agents.format_mql_node import format_mql_node
from iot_data_analyse_agent.configuration import BaseConfiguration
from iot_data_analyse_agent.state import InputState, AgentState

# 创建状态图
workflow = StateGraph(
    state_schema=AgentState,
    input_schema=InputState,
    context_schema=BaseConfiguration
)

# 添加节点
workflow.add_node("intent_slot_extraction_node", intent_slot_extraction_node)
workflow.add_node("metric_retrieval_node", metric_retrieval_node)
workflow.add_node("format_mql_node", format_mql_node)
workflow.add_node("mql_generation_node", mql_generation_node)

# 定义边
workflow.add_edge(START, "intent_slot_extraction_node")

def route_after_intent_extraction(state: AgentState) -> str:
    """根据意图提取结果路由"""
    if state.get("is_chat") == "True":
        return END
    else:
        return "metric_retrieval_node"

workflow.add_conditional_edges(
    source="intent_slot_extraction_node",
    path=route_after_intent_extraction,
    path_map={END: END, "metric_retrieval_node": "metric_retrieval_node"}
)

def route_after_metric_retrieval(state: AgentState) -> str:
    """根据指标检索结果路由"""
    metric_results = state.get("metric_results_txt", "")
    if metric_results and "No matching metrics located" in metric_results:
        return END
    else:
        return "mql_generation_node"

workflow.add_conditional_edges(
    source="metric_retrieval_node",
    path=route_after_metric_retrieval,
    path_map={END: END, "mql_generation_node": "mql_generation_node"}
)

workflow.add_edge("mql_generation_node", "format_mql_node")
workflow.add_edge("format_mql_node", END)

# 编译图
graph = workflow.compile()

设计要点：

• StateGraph 需要指定 state_schema、input_schema、context_schema
• 使用 add_node() 添加节点
• 使用 add_edge() 添加固定边
• 使用 add_conditional_edges() 添加条件边，需要定义路由函数
• 最后调用 workflow.compile() 编译图

6. 实现节点

每个节点是一个异步函数，接收 state 和 config 参数：

from langchain_core.runnables import RunnableConfig
from iot_data_analyse_agent.state import AgentState
from iot_data_analyse_agent.tools.intent_slot_extraction_tool.intent_slot_extraction_tool import intent_extraction_tool

async def intent_slot_extraction_node(state: AgentState, config: RunnableConfig = None):
    """意图槽位提取节点"""
    try:
        # 调用工具
        intent_response = await intent_extraction_tool.ainvoke(
            {"question": state['user_input']},
            config=config
        )
        
        # 处理结果
        intent_results = parse_intent_response(intent_response)
        
        # 返回状态更新
        return {
            "messages": [intent_response],
            "intent_results": intent_results,
            "is_chat": 'False'
        }
    except Exception as e:
        return {
            "messages": [AIMessage(content=f"Error: {str(e)}")],
            "is_chat": 'True',
            "error": str(e)
        }

节点设计模式：

• 节点函数签名：async def node_name(state: AgentState, config: RunnableConfig = None)
• 返回字典：只返回需要更新的状态字段
• 错误处理：捕获异常并返回错误状态

7. 实现工具

工具使用 @tool 装饰器定义：

from langchain_core.tools import tool
from langchain_core.runnables import RunnableConfig
from agent_api_server.dynamic_llm.dynamic_llm import DynamicLLM, ConfigType

@tool(
    description="""
    Universal intent extraction tool that uses LLM to extract structured analysis requirements.
    
    Outputs JSON array format containing:
    - base_metric: base metric name
    - time_info: time range description
    - dimensions: analysis dimensions
    - filters: filter conditions
    """
)
async def intent_extraction_tool(question: str, config: RunnableConfig = None):
    """意图提取工具"""
    llm = DynamicLLM(tool_name="default", config_type=ConfigType.CHAT)
    
    response = await llm.ainvoke(
        input_dict=[
            SystemMessage(content=NER_INTENT_SYSTEM_PROMPT),
            HumanMessage(content=question)
        ],
        config=config
    )
    
    return response

工具设计要点：

• 使用 @tool 装饰器，添加 description 参数
• 工具函数可以是同步或异步
• 使用 DynamicLLM 调用模型，支持多租户动态切换
• config 参数用于传递租户信息和回调配置

---

Langfuse 集成

AgentHub SDK 内置了 Langfuse 集成，用于 Agent 调用的可观测性和追踪。

启用 Langfuse

在 .env 中配置 Langfuse 环境变量：

LANGFUSE_SECRET_KEY=sk-lf-xxxxx
LANGFUSE_PUBLIC_KEY=pk-lf-xxxxx
LANGFUSE_BASE_URL=https://cloud.langfuse.com

当三个环境变量都配置后，SDK 会自动启用 Langfuse 回调。

集成方式

1. REST API 调用

在 agent_api_server/shared/message.py 中：

from langfuse.langchain import CallbackHandler
from langfuse import propagate_attributes

# 检查 Langfuse 配置
langfuse_keys = [
    os.getenv("LANGFUSE_SECRET_KEY"),
    os.getenv("LANGFUSE_PUBLIC_KEY"),
    os.getenv("LANGFUSE_BASE_URL")
]

callbacks_config = {}
if all(langfuse_keys):
    langfuse_handler = CallbackHandler()
    callbacks_config = {"callbacks": [langfuse_handler], "run_name": state.graph_name}

config = {"configurable": configurable_params, **callbacks_config}

# 使用 propagate_attributes 设置追踪属性
with propagate_attributes(
    session_id=state.thread_id,
    user_id=user_id,
    trace_name=f"{state.graph_name}"
):
    async for stream_event in graph_instance.astream(
        inputs or {},
        config=config,
        stream_mode=["updates", "messages", "custom"],
        subgraphs=True
    ):
        # 处理流式事件
        pass

2. MCP 调用

在 agent_api_server/mcp_convert/mcp_convert.py 中：

from langfuse import propagate_attributes
from langfuse.langchain import CallbackHandler

# 配置 Langfuse 回调
langfuse_keys = [
    os.getenv("LANGFUSE_SECRET_KEY"),
    os.getenv("LANGFUSE_PUBLIC_KEY"),
    os.getenv("LANGFUSE_BASE_URL")
]

callbacks_config = {}
if all(langfuse_keys):
    langfuse_handler = CallbackHandler()
    callbacks_config = {"callbacks": [langfuse_handler], "run_name": f"{graph_name}_MCP_Call"}

config = {"configurable": configurable_params, **callbacks_config}

# 使用 propagate_attributes 设置追踪属性
with propagate_attributes(
    session_id=thread_id,
    user_id=user_id,
    trace_name=f"{graph_name}_MCP_Call"
):
    async for stream_event in graph_instance.astream(
        input_dict,
        config=config,
        stream_mode=["updates"],
        subgraphs=True
    ):
        # 处理流式事件
        pass

3. A2A 调用

在 agent_api_server/a2a_bridge/langgraph_executor.py 中：

from agent_api_server.services.config_builder import build_configurable, build_langfuse_callbacks
from langfuse import propagate_attributes

# 使用统一的 config_builder 构建配置
configurable = build_configurable(
    graph_name=graph_name,
    thread_id=context.context_id,
    ts_tenant=ts_tenant,
    ei_token=ei_token,
    extra={"user_id": user_id},
)
config = {"configurable": configurable}

# 使用统一的 langfuse 回调构建
callbacks_config = build_langfuse_callbacks(run_name=f"A2A_{graph_name}")
a2a_config = {**config, **callbacks_config}

# 使用 propagate_attributes 设置追踪属性
if callbacks_config and propagate_attributes:
    with propagate_attributes(
        session_id=configurable.get("thread_id", ""),
        user_id=configurable.get("user_id", ""),
        trace_name=f"A2A_{graph_name}"
    ):
        result = await graph.ainvoke(inputs, config=a2a_config)

A2A 追踪特点：

• Trace 名称以 A2A_ 前缀标识，便于区分调用来源
• session_id 使用 A2A 的 context_id
• user_id 从 ei_token（JWT）中解析，通过 call_a2a_client.py 提取并传递
• 使用 RedisTaskStore 支持多 worker 水平扩展
• 支持跨服务链式调用的完整追踪

追踪内容

Langfuse 会记录以下信息：

• Trace：每次 Agent 调用的完整追踪
• Span：每个节点的执行过程
• Generation：LLM 调用的详细信息（输入、输出、token 使用）
• Session：按 session_id（thread_id）分组
• User：按 user_id 分组

查看追踪

1. 访问 Langfuse 控制台：https://cloud.langfuse.com
2. 查看 Traces 页面，可以看到所有 Agent 调用
3. 点击单个 Trace 查看详细的执行过程、LLM 调用和 token 使用

自定义追踪属性

可以通过 propagate_attributes 添加自定义属性：

with propagate_attributes(
    session_id=state.thread_id,
    user_id=user_id,
    trace_name=f"{state.graph_name}",
    tags=["production", "v1"],
    metadata={"app_id": app_id, "tenant_id": ts_tenant}
):
    # Agent 执行代码
    pass

---