基于 LangGraph 和 FastAPI 构建的智能 Agent 服务器，采用 JSON-RPC 2.0 协议通过 WebSocket 实现与客户端的实时双向通信。支持多 Agent 动态切换、混合工具系统（内置工具 + 客户端远程工具 + MCP 工具）、以及 RESTful API 扩展。
- JSON-RPC 2.0 协议: 标准化的请求/响应/通知格式，支持批量请求
- WebSocket 实时通信: 基于 FastAPI WebSocket 的双向通信
- 多 Agent 架构: 支持多个 Agent 并存，按需切换，自动发现注册
- 混合工具系统:
- 内置工具 — 服务端直接执行（如
get_current_time、calculator） - 客户端工具 — 通过 JSON-RPC 远程调用客户端执行
- MCP 工具 — 通过 Model Context Protocol 连接外部服务
- 内置工具 — 服务端直接执行（如
- Graph-based 执行: LangGraph 状态图驱动，内置 interrupt / resume 机制
- RESTful API: 独立的 HTTP 端点，支持批量元件型号替换等功能
- 可配置 LLM: 支持 OpenAI 兼容 API（DeepSeek、Qwen 等）
- 会话管理: 多会话并发，每个会话独立状态、独立工具集
- 日志系统: 按天轮转的控制台 + 文件双通道日志
┌──────────────┐ WebSocket ┌───────────────────────────────────┐
│ Client │ ◄──────────────────────► │ Agent Server │
│ (C# .NET) │ JSON-RPC 2.0 │ (Python) │
└──────┬───────┘ │ │
│ │ ┌─────────────┐ ┌───────────┐ │
│ 1. agent.register │ │ LangGraph │ │ MCP │ │
├─────────────────────────────────►│ │ Agent │ │ Manager │ │
│ │ │ ┌─────────┐ │ │ │ │
│ 2. agent.chat │ │ │CAD Agent│ │ │ dx2-mcp │ │
├─────────────────────────────────►│ │ ├─────────┤ │ └───────────┘ │
│ │ │ │Component│ │ │
│ 3. agent.toolCall (notification) │ │ │Replace │ │ ┌───────────┐ │
│◄─────────────────────────────────┤ │ │Agent │ │ │ Builtin │ │
│ │ │ └─────────┘ │ │ Tools │ │
┌────┴─────┐ │ └─────────────┘ └───────────┘ │
│ Execute │ │ │
│ Tool │ │ ┌────────────────────────────┐ │
└────┬─────┘ │ │ REST API │ │
│ │ │ POST /api/batch-replace │ │
│ 4. agent.submitToolResult │ └────────────────────────────┘ │
├─────────────────────────────────►│ │
│ └───────────────────────────────────┘
│ 5. agent.response (notification)
│◄─────────────────────────────────┤
Client Server
│ │
│ ── agent.register (tools=[...]) ──────────► │ 注册客户端 & 工具
│ ◄── result: {status, tools_synced} ──────── │
│ │
│ ── agent.chat (message) ──────────────────► │ 发送用户消息
│ ◄── agent.status {thinking} ─────────────── │ 状态通知
│ │
│ ┌─── LangGraph 执行 ───┐ │
│ │ call_llm → 工具调用 │ │
│ │ 内置工具 → 立即执行 │ │
│ │ 客户端工具 → interrupt │ │
│ └──────────────────────┘ │
│ │
│ ◄── agent.toolCall {call_id, args} ──────── │ 请求执行客户端工具
│ ── agent.submitToolResult {result} ────────► │ 返回工具结果
│ │
│ ┌─── resume 继续执行 ───┐ │
│ └──────────────────────┘ │
│ │
│ ◄── agent.response {content} ────────────── │ 最终响应
- Python >= 3.11
- uv (Python 包管理器)
# 克隆项目
git clone <repository-url>
cd agent-server
# 安装依赖
uv sync创建 .env 文件：
# LLM API 配置（OpenAI 兼容接口）
OPENAI_API_KEY=your-api-key
OPENAI_API_BASE=https://api.openai.com/v1
MODEL_NAME=gpt-4
TEMPERATURE=0.01
# 服务器配置
HOST=0.0.0.0
PORT=8999
# Agent 配置
AGENT_TIMEOUT=600.0
# 其他
MAX_MESSAGE_SIZE=1048576
LOG_BACKUP_COUNT=30python app.py服务器将在 http://0.0.0.0:8999 启动。可通过 /health 和 /ready 检查服务状态。
agent-server/
├── app.py # FastAPI 应用入口 & WebSocket/HTTP 端点
├── pyproject.toml # 项目配置和依赖
├── mcp_servers.json # MCP 服务器配置
├── .env # 环境变量配置
│
├── agent/ # Agent 核心
│ ├── base.py # BaseAgent 基类（LangGraph 图构建、interrupt/resume）
│ ├── state.py # AgentState / PendingToolCall 状态定义
│ ├── registry.py # AgentRegistry 注册表（自动发现、单例管理）
│ └── agents/ # 具体 Agent 实现
│ ├── cad_agent.py # CADAgent — CAD 绘图/建模助手
│ └── component_replacement_agent.py
│ # ComponentReplacementAgent — 电气元件替换工作流
│
├── jsonrpc/ # JSON-RPC 2.0 协议层
│ ├── models.py # 请求/响应/通知/错误码 Pydantic 模型
│ ├── methods.py # MethodRouter 方法路由器（装饰器注册 + 分发）
│ └── handler.py # JsonRpcHandler 消息解析 & 处理
│
├── services/ # 业务服务层
│ ├── websocket_handler.py # JSON-RPC 方法实现（register/chat/toolResult 等）
│ ├── connection_manager.py # WebSocket 连接 & 会话管理（SessionData）
│ └── agent_handler.py # Agent 执行器（execute / resume + 超时控制）
│
├── tools/ # 工具系统
│ ├── base.py # ToolRegistry 注册表 + MCP→OpenAI 格式转换
│ └── builtin.py # 内置工具（get_current_time / calculator）
│
├── api/ # RESTful API
│ └── routers/
│ └── batch_replace.py # POST /api/batch-replace — 批量元件型号替换
│
├── core/ # 核心基础设施
│ ├── config.py # Settings 配置类（Pydantic Settings + .env）
│ ├── logging.py # 日志配置（控制台 + 按天轮转文件）
│ ├── mcp_manager.py # MCPManager — MCP 服务器生命周期管理
│ └── exceptions.py # 自定义异常体系
│
├── utils/ # 工具库
│ └── dx2_accessor.py # Dx2 数据访问（厂家/系列查询 + 模糊搜索）
│
├── client/ # C# 客户端代码
│ └── Leadsoft.Common.Agent/ # C# .NET Agent 客户端库
│
├── scripts/ # 辅助脚本
│ ├── debug_component_replacement_agent.py
│ └── generate_bomset.py
│
└── assets/ # 静态数据资源
└── dx2/ # Dx2 厂家/系列数据文件
ws://localhost:8999/v1/ws?session_id={your-session-id}
也支持旧版端点
ws://localhost:8999/ws?session_id={your-session-id}
{
"jsonrpc": "2.0",
"method": "agent.register",
"params": {
"client_id": "cad-client-001",
"agent_id": "component_replacement",
"tools": [
{
"name": "fetch_bom_tool",
"description": "获取BOM明细列表",
"inputSchema": {
"type": "object",
"properties": {},
"required": []
}
}
]
},
"id": 1
}响应：
{
"jsonrpc": "2.0",
"result": {
"status": "registered",
"agent_id": "component_replacement",
"tools_count": 1,
"tools_synced": ["fetch_bom_tool", "ask_user_tool"]
},
"id": 1
}{
"jsonrpc": "2.0",
"method": "agent.chat",
"params": { "message": "把所有NDM2断路器替换为正泰NXM系列" },
"id": 2
}{
"jsonrpc": "2.0",
"method": "agent.submitToolResult",
"params": {
"call_id": "call_abc123",
"tool_name": "fetch_bom_tool",
"success": true,
"result": "[{\"id\":1, \"type\":\"NDM2-63\", ...}]"
},
"id": 3
}{
"jsonrpc": "2.0",
"method": "agent.clearHistory",
"params": { "agent_id": "component_replacement" },
"id": 4
}{ "jsonrpc": "2.0", "method": "system.ping", "id": 5 }{ "jsonrpc": "2.0", "method": "system.listAgents", "id": 6 }{
"jsonrpc": "2.0",
"method": "agent.status",
"params": { "status": "thinking", "message": "正在分析替换需求..." }
}状态值包括：thinking、executing、tools_synced 等。
{
"jsonrpc": "2.0",
"method": "agent.toolCall",
"params": {
"call_id": "call_abc123",
"tool_name": "fetch_bom_tool",
"arguments": {}
}
}{
"jsonrpc": "2.0",
"method": "agent.response",
"params": { "content": "已完成所有NDM2断路器到NXM系列的替换。" }
}| 代码 | 常量 | 含义 |
|---|---|---|
-32700 |
PARSE_ERROR |
无效的 JSON |
-32600 |
INVALID_REQUEST |
无效的请求对象 |
-32601 |
METHOD_NOT_FOUND |
方法不存在 |
-32602 |
INVALID_PARAMS |
无效的参数 |
-32603 |
INTERNAL_ERROR |
内部错误 |
-32000 |
SERVER_ERROR |
通用服务器错误 |
-32001 |
AGENT_ERROR |
Agent 执行错误 |
-32002 |
TOOL_ERROR |
工具执行错误 |
-32003 |
TIMEOUT_ERROR |
执行超时 |
-32004 |
NOT_REGISTERED |
客户端未注册 |
服务器启动时自动扫描 agent/agents/ 目录，发现所有 BaseAgent 子类并注册：
# agent/registry.py
registry.auto_discover() # 自动发现并实例化所有 Agent| Agent ID | 类名 | 说明 |
|---|---|---|
cad |
CADAgent |
CAD 绘图/建模/标注助手，使用通用 BaseAgent 图 |
component_replacement |
ComponentReplacementAgent |
电气元件替换工作流，自定义多步骤状态图 |
通用 Agent 使用的 LangGraph 图：
START → call_llm → (检查是否有工具调用)
│ │
▼ ▼
END process_tools → (interrupt 客户端工具)
│
▼ (resume)
call_llm → ...
电气元件替换的多步骤有状态工作流：
START → 步骤1: 确认替换明细
│
▼
步骤2: 确认目标厂家 → (搜索厂家数据库)
│
▼
步骤3: 确认目标系列 → (搜索系列数据库)
│
▼
自动执行替换 → END
每个步骤循环执行 LLM调用 → 工具处理 → 状态更新，直到满足步骤完成条件后自动推进。
继承 BaseAgent 并实现必要方法：
from agent.base import BaseAgent
class MyAgent(BaseAgent):
def get_agent_id(self) -> str:
return "my_agent"
def get_system_prompt(self) -> str:
return "你是一个专业的助手..."
def get_tools(self) -> list:
# 返回内置工具列表（可选）
return []
# 可选：覆盖 create_graph() 实现自定义工作流
# 可选：覆盖 get_llm_config() 使用不同的 LLM将文件放入 agent/agents/ 目录即可自动注册。
| 工具名 | 说明 |
|---|---|
get_current_time |
获取当前服务器时间 |
calculator |
执行基本数学运算 |
search_manufacturer_tool |
模糊搜索电气元件厂商 |
search_series_tool |
模糊搜索元件系列 |
batch_search_series_tool |
批量搜索元件系列 |
客户端在 agent.register 时注册的工具，LLM 调用时通过 agent.toolCall 通知客户端执行。
工具定义遵循 MCP 协议格式（name / description / inputSchema），服务端自动转换为 OpenAI 格式。
通过 mcp_servers.json 配置外部 MCP 服务器，启动时自动连接并加载工具：
{
"mcpServers": {
"dx2-mcp": {
"command": "python",
"args": ["path/to/dx2-mcp/main.py"],
"env": {},
"agents": ["component_replacement"]
}
}
}agents 字段指定哪些 Agent 可以使用该 MCP 服务的工具。
根据一组替换示例（old → new），利用 LLM 推导规律并批量替换：
请求：
{
"old": "NDM2-63/3200 ln=32A",
"new": "NXM-63H/3200 32A",
"tasks": ["NDM2-100/3200 ln=25A", "NDM2-100M/3340 ln=63A"]
}响应：
{
"results": [
{ "original": "NDM2-100/3200 ln=25A", "replaced": "NXM-100H/3200 25A" },
{ "original": "NDM2-100M/3340 ln=63A", "replaced": "NXM-100H/3340 63A" }
]
}| 方法 | 路径 | 说明 |
|---|---|---|
GET |
/ |
API 信息和可用 Agent |
GET |
/health |
健康检查 |
GET |
/ready |
就绪检查（验证 Agent 已加载） |
所有配置通过环境变量或 .env 文件设置，由 Pydantic Settings 管理：
| 配置项 | 默认值 | 说明 |
|---|---|---|
OPENAI_API_KEY |
— | OpenAI 兼容 API 密钥 |
OPENAI_API_BASE |
https://api.openai.com/v1 |
API 基础 URL |
MODEL_NAME |
gpt-4 |
使用的模型名称 |
TEMPERATURE |
0.01 |
LLM 温度参数 |
HOST |
127.0.0.1 |
服务器监听地址 |
PORT |
8999 |
服务器端口 |
AGENT_TIMEOUT |
60.0 |
Agent 单次执行超时（秒） |
LOG_LEVEL |
INFO |
日志级别 |
LOG_DIR |
logs |
日志文件目录 |
LOG_BACKUP_COUNT |
30 |
日志保留天数 |
MAX_MESSAGE_SIZE |
1048576 |
最大消息大小（字节） |
| 依赖包 | 用途 |
|---|---|
fastapi |
Web 框架 & WebSocket |
uvicorn |
ASGI 服务器 |
langgraph |
Agent 状态图引擎 |
langchain-openai |
OpenAI 兼容 LLM 调用 |
langchain-core |
LangChain 核心抽象（Tool / Message） |
langchain-mcp-adapters |
MCP 工具适配器 |
mcp |
Model Context Protocol 客户端 |
pydantic |
数据验证和序列化 |
pydantic-settings |
配置管理 |
websockets |
WebSocket 支持 |
import asyncio
import websockets
import json
async def test_agent():
uri = "ws://localhost:8999/v1/ws?session_id=test-001"
async with websockets.connect(uri) as ws:
# 1. 注册
await ws.send(json.dumps({
"jsonrpc": "2.0",
"method": "agent.register",
"params": {"client_id": "python-client", "agent_id": "cad", "tools": []},
"id": 1
}))
print("注册响应:", await ws.recv())
# 2. 发送消息
await ws.send(json.dumps({
"jsonrpc": "2.0",
"method": "agent.chat",
"params": {"message": "现在几点了？"},
"id": 2
}))
# 3. 接收消息循环
while True:
msg = json.loads(await ws.recv())
print("收到:", msg)
if msg.get("method") == "agent.toolCall":
# 处理工具调用...
pass
elif msg.get("method") == "agent.response":
print("最终响应:", msg["params"]["content"])
break
asyncio.run(test_agent())- API 密钥安全: 使用
.env或环境变量，不要提交到版本控制 - WebSocket 认证: 生产环境添加 Token 验证机制
- HTTPS / WSS: 生产环境使用加密连接
- CORS 配置: 生产环境应限制
allow_origins为具体域名 - 速率限制: 添加请求频率限制
- 输入验证: JSON-RPC 层已内置 Pydantic 参数验证
MIT License