Metadata-Version: 2.3
Name: agent-api-server
Version: 2.1.10
Summary: A Langgraph agent API server that implements Langgraph agent's web capabilities and can interact with chatbot
Author: zijie.zhang@advantech.com.cn
Requires-Python: >=3.10,<3.13
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Provides-Extra: a2a
Requires-Dist: a2a-sdk[http-server] (==1.1.2)
Requires-Dist: a2a-sdk[http-server] (==1.1.2) ; extra == "a2a"
Requires-Dist: agent-client-protocol (>=0.11.0,<1.0.0)
Requires-Dist: agent-client-protocol (>=0.11.0,<1.0.0) ; extra == "a2a"
Requires-Dist: aiofiles (>=24.1.0,<25.0.0)
Requires-Dist: authlib (>=1.6.5)
Requires-Dist: cryptography (>=45.0.4,<46.0.0)
Requires-Dist: fastapi (>=0.117.0,<0.121.2)
Requires-Dist: fastmcp (>=2.13.0,<3.0.0)
Requires-Dist: httpx (>=0.27.0,<1.0.0)
Requires-Dist: httpx (>=0.27.0,<1.0.0) ; extra == "a2a"
Requires-Dist: langchain-community (==0.3.29)
Requires-Dist: langchain-core (==0.3.76)
Requires-Dist: langchain-openai (==0.3.33)
Requires-Dist: langchain-text-splitters (==0.3.11)
Requires-Dist: langfuse (>=3.14.5,<4.0.0)
Requires-Dist: langgraph (==0.6.11)
Requires-Dist: langgraph-checkpoint (>=3.0.1,<4.0.0)
Requires-Dist: langgraph-checkpoint-postgres (==3.0.0)
Requires-Dist: langgraph-prebuilt (==0.6.5)
Requires-Dist: langsmith (==0.4.28)
Requires-Dist: llm-sdk (==0.0.10)
Requires-Dist: model-manage-client (>=0.0.1.8,<0.0.2.0)
Requires-Dist: nats-py (>=2.11.0,<3.0.0)
Requires-Dist: psycopg-binary (>=3.2.9,<4.0.0)
Requires-Dist: psycopg-pool (>=3.2.6,<4.0.0)
Requires-Dist: pydantic-settings (>=2.9.1,<3.0.0)
Requires-Dist: redis (>=6.2.0,<7.0.0)
Requires-Dist: starlette (>=0.49.3,<0.50.0)
Requires-Dist: tenacity (>=9.1.2,<10.0.0)
Requires-Dist: xinference-client (>=1.8.0,<2.0.0)
Description-Content-Type: text/markdown

## 背景：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](https://github.com/microsoft/mcp-for-beginners/blob/main/translations/zh-CN/01-CoreConcepts/mcp-2026-07-28-release-candidate.md)）。用通俗的话说，这次更新解决了四个问题：

**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 通信
```

**代码示例**：
```python
# 之前：工具只返回数据
@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](https://github.com/openclaw/openclaw) 是由 OpenClaw 基金会维护的开源个人 AI 助手，支持 25+ 消息通道和多模型提供商。

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

---

基于 [LangGraph](https://github.com/langchain-ai/langgraph) + [FastAPI](https://fastapi.tiangolo.com/) 的 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 直接调用
- **多租户隔离** — 支持按租户隔离模型配置和认证


---

## 系统架构

![系统架构图](docs/images/system_architecture.png)

### 分层说明

**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)](https://github.com/google/A2A) 协议，使你的 Agent 可以被其他 A2A 兼容系统发现和调用。

### A2A 协议流程

![A2A 协议流程](docs/images/a2a_flow.png)

**① 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 链式调用**。以下是一个典型场景：

![A2A 链式调用](docs/images/a2a_chain.png)

**场景说明：**

用户向 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` 中设置：

```bash
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 间调用示例

```python
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)](https://modelcontextprotocol.io/) 协议，将你的 LangGraph Agent 暴露为 MCP 工具，供 Claude、GPT、IDE 等 LLM 客户端直接调用。

### MCP 协议流程

![MCP 协议流程](docs/images/mcp_flow.png)

**① 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` 中设置：

```bash
ENABLE_MCP_SERVER=true
MCP_SERVER_PORT=8081
```

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

### MCP 客户端调用示例

```python
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)](https://agentclientprotocol.com/) 协议，通过轻量级 CLI 代理将你的 LangGraph Agent 暴露为 VS Code、Zed、JetBrains 等编辑器的 AI 助手。

### ACP 协议架构

![ACP 协议架构](docs/images/acp_architecture.png)

### 工作原理

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**

```bash
# 从 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` 中添加：

```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](https://agentclientprotocol.com/)），远端调用能力尚未标准化，暂不支持 |
| 远端适配 |  待定 | 待 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](https://github.com/nemoclaw) 和 [OpenClaw](https://github.com/openclaw) 整合，实现**每用户沙箱调度 + 边缘 Agent 部署**的架构。

### 架构图

![NemoClaw + OpenClaw 整合架构](docs/images/nemoclaw_openclaw_architecture.png)

### 架构说明

**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
```

**代码示例：**

```python
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          # 自定义异常
```


---

## 快速开始

### 安装

```bash
pip install agent-api-server

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

### 配置

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

```bash
# 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` 配置文件：

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

### 启动服务

```python
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` 中提供的启动方式：

```python
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`：

```bash
python demo.py
```


---

## 模块说明

### DynamicLLM

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

```python
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 助手：

```python
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 两种消息队列：

```python
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/` — 获取服务配置


---

## 开发

```bash
# 安装开发依赖
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 会自动读取此文件：

```json
{
  "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 的状态结构，包括输入状态和完整状态：

```python
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 的可配置参数，支持环境变量回退：

```python
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 工作流：

```python
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` 参数：

```python
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` 装饰器定义：

```python
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](https://langfuse.com/) 集成，用于 Agent 调用的可观测性和追踪。

### 启用 Langfuse

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

```bash
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` 中：

```python
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` 中：

```python
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` 中：

```python
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` 添加自定义属性：

```python
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
```


---

