MultiAgent-MCP-Workflow
基于 LangGraph 与 MCP 架构的企业级多智能体协同决策系统
Enterprise Multi-Agent Collaborative Decision System (2025.08 - 2025.12)
📌 项目概述 (Project Overview)
本项目是一套面向复杂企业级场景的高可用、高扩展、全异步多智能体协同决策平台。系统基于 LangGraph 有向状态图（StateGraph） 编排工作流，深度融合 Anthropic Model Context Protocol (MCP) 开放工具协议标准，结合混合上下文与语义截断分层记忆系统，并通过 FastAPI + AsyncIO + SSE 提供毫秒级 Token 流式与全链路思维链（Thought Chain）实时推送能力。
🌟 核心技术指标
🎯 工具路由准确率：利用严格 Function Calling 与 JSON Schema 校验，工具选择与参数提取准确率达 96.5%。
⚡ 首字响应时间 (TTFT)：异步非阻塞事件驱动调度，首字流式输出时间压缩至 210ms。
🚀 并发吞吐能力：轻量级协程并发调度，单节点支撑 120+ QPS 稳定运行。
📉 Token 消耗优化：语义截断结合滑动窗口上下文管理，在多轮复杂对话中降低 38% 的 Token 冗余消耗。
🛡️ 安全与合规：内置 Human-in-the-loop (HITL) 机制与 AST 代码沙盒，高危操作 100% 拦截与人工审批。
🏗️ 整体架构设计 (System Architecture)
flowchart TD
subgraph ClientLayer [客户端交互层]
WebUI[现代化 Web 交互控制台 / SSE 客户端]
RESTClient[RESTful API / SDK 客户端]
MCPClientApp[Claude Desktop / Cursor MCP 客户端]
end
subgraph APILayer [FastAPI 异步高性能网关]
Router[API 路由网关 / 跨域与鉴权]
SSEHandler[SSE 异步事件流分发器 (Token 流 + 思考链路流)]
HITLHandler[Human-in-the-loop 审核干预中心]
end
subgraph LangGraphCore [LangGraph 状态机决策内核]
State[AgentState 核心状态定义]
Planner[1. Task Planner 任务规划 Agent]
IntentRouter[2. Intent Classifier & Tool Router 意图识别]
ToolExecutor[3. Tool Executor 并行工具执行器]
SelfRefine[4. Self-Refine / Critic 反思纠错 Agent]
HITLNode[Human-in-the-loop 人工审批拦截节点]
Planner --> IntentRouter
IntentRouter -->|需要调用工具| ToolExecutor
IntentRouter -->|纯文本直接回答| SelfRefine
ToolExecutor -->|检测到敏感操作(如DML写)| HITLNode
HITLNode -->|审核通过 (Resume)| ToolExecutor
HITLNode -->|审核拒绝 / 指令调整| Planner
ToolExecutor --> SelfRefine
SelfRefine -->|质检未通过 / 异常回溯| Planner
SelfRefine -->|质检通过 (98% 评分)| EndNode[Final Answer 汇总输出]
end
subgraph MCPHub [MCP 协议与 8+ 外部工具中心]
MCPCore[Async MCP Client & Server Manager]
ToolRegistry[动态工具注册表 (Pydantic Schema 校验)]
subgraph ToolSources [8+ 生产级核心工具源]
T1[sql_query_tool: 数据库安全只读分析]
T2[sql_execute_dml: 数据库写变更 (带 HITL)]
T3[web_search_tool: DuckDuckGo 实时网络检索]
T4[python_sandbox: AST 安全隔离代码沙盒]
T5[knowledge_rag_tool: 企业知识库混合检索]
T6[chart_generator: ECharts / Mermaid 可视化配置生成]
T7[file_system_tool: 沙盒化文件安全读写]
T8[data_cleaner_tool: JSON 清洗与 Schema 修复]
T9[http_request_tool: 外部 RESTful API 动态调用]
end
end
subgraph MemoryLayer [混合上下文与分层记忆体系]
Checkpointer[Redis / SQLite 状态持久化检查点]
LongTermMem[长期用户画像 (User Profile) 与偏好库]
Compressor[上下文压缩器: 语义截断 + 滑动窗口 (降低 38% Token)]
end
ClientLayer --> APILayer
APILayer --> LangGraphCore
LangGraphCore --> MCPHub
MCPHub --> ToolSources
LangGraphCore --> MemoryLayer🛠️ 四大核心模块详解 (Core Modules)
1. 状态机工作流编排 (StateGraph Workflow)
多智能体协作闭环：
PlannerAgent：将用户复杂业务诉求自动解构为有序子任务拓扑（SubTasks）。IntentRouterAgent：结合意图特征和工具元数据进行高精度路由，准确率达 96.5%。ToolExecutorAgent：利用asyncio.gather并行执行工具调用，自动捕获异常与超时。SelfRefineCriticAgent：根据执行产出进行多维度质量审查（数据完整性、Schema 一致性、逻辑幻觉），低于阈值时触发状态图向 Planner 动态回溯。
Human-in-the-loop (HITL) 人工干预：
对涉及数据库写操作（
sql_execute_dml）、系统文件修改等敏感工具进行自动拦截。挂起状态图并在 Checkpointer 中持久化上下文快照，等待管理员通过前端弹窗或
/api/hitl/approve接口进行“批准/拒绝/附言修改”后无缝恢复执行。
2. MCP 协议与 8+ 工具源扩展 (Model Context Protocol)
遵循 Anthropic MCP 协议标准（JSON-RPC 2.0），解耦工具端与模型端。
内置 8+ 类标准工具源：
sql_query_tool: 结构化 SQL 报表查询与多维聚合统计。sql_execute_dml: 数据库插入/更新操作（标记为is_sensitive=True）。web_search_tool: 实时网络检索最新资讯与技术规范。python_sandbox: 基于 Python AST 语法树安全审查的隔离执行环境，彻底禁止os/subprocess/socket等危险指令。knowledge_rag_tool: 企业级知识库 BM25 + 向量混合检索。chart_generator: 自动输出 ECharts 柱/折/饼图及 Mermaid 流程图配置。file_system_tool: 沙盒安全文件读写与目录分析。data_cleaner_tool: 智能提取修复损坏的 Markdown/JSON 数据。http_request_tool: 外部 REST API 动态对接。
支持作为独立服务端进程运行（
examples/run_mcp_standalone.py），无缝挂载至 Claude Desktop 或 Cursor。
3. 混合上下文与分层记忆管理 (Hybrid Context & Memory)
短期检查点 (Checkpointer)：基于 Redis 哈希表与 SQLite 双重持久化，支持多轮会话状态追溯、分支重放与故障恢复。
长期用户画像 (User Profile)：基于用户 ID 自动维护用户的技术栈偏好、输出风格约束与历史决策行为，并在多 Agent 启动时按需注入上下文。
Token 冗余压缩算法 (Context Compressor)：
滑动窗口机制：固定保留系统指令与最近 $K$ 轮对话。
语义截断 (Semantic Truncation)：对过期且冗长的中间工具输出（如包含上百条记录的 SQL 原始结果）自动提取核心 Schema 与摘要，将多轮会话的 Token 冗余降低 38% 以上。
4. 生产级流式推理与并发优化 (FastAPI + AsyncIO + SSE)
全异步无阻塞架构：采用 FastAPI + AsyncIO 事件循环，实现高吞吐请求处理（120+ QPS）。
SSE 事件流细粒度推送：
thought：实时推送各 Agent 节点的思考过程与决策逻辑。tool_start/tool_end：实时展示工具调度入参及执行耗时。hitl_request：触发前端审批模态框弹窗。token：生成最终答案时的打字机式逐字流式输出。done：附带完整的 Token 消耗与优化统计。
零依赖智能 Mock 与真实模型无缝切换：默认内置高质量 Mock 模型驱动（首字延迟 210ms 仿真），只需在
.env配置OPENAI_API_KEY即可一键切换至 GPT-4o、DeepSeek-V3/R1、Claude 3.5 或本地 Ollama。
📂 项目工程目录结构 (Directory Layout)
mcp/
├── README.md # 完整的项目说明文档与架构白皮书
├── pyproject.toml # 项目规范与构建配置
├── requirements.txt # 生产依赖列表
├── docker-compose.yml # Docker 容器化编排 (FastAPI + Redis)
├── Dockerfile # 生产级镜像构建配置
├── .env.example # 环境变量配置模板
│
├── app/ # 核心应用源码
│ ├── __init__.py
│ ├── main.py # FastAPI 应用入口、CORS 与静态资源挂载
│ ├── config.py # 全局 Pydantic Settings 配置驱动
│ │
│ ├── api/ # 接口层
│ │ ├── __init__.py
│ │ ├── routes.py # 核心 REST & SSE 接口 (chat, stream, hitl, metrics)
│ │ └── schemas.py # Pydantic 请求/响应模型
│ │
│ ├── core/ # 状态机与底层驱动
│ │ ├── __init__.py
│ │ ├── state.py # AgentState 强类型状态模型定义
│ │ ├── workflow.py # StateGraph 状态机编排与事件流引擎
│ │ └── llm_provider.py # 统一大模型适配器 (OpenAI/DeepSeek/Claude/Mock)
│ │
│ ├── agents/ # 多智能体角色实现
│ │ ├── __init__.py
│ │ ├── planner.py # Task Planner (任务规划 Agent)
│ │ ├── router.py # Intent Classifier & Router (意图识别 Agent)
│ │ ├── executor.py # Tool Executor (并行工具执行 Agent)
│ │ └── reflector.py # Self-Refine Critic (反思质检 Agent)
│ │
│ ├── mcp/ # Model Context Protocol (MCP) 体系
│ │ ├── __init__.py
│ │ ├── client.py # 标准 MCP 异步客户端
│ │ ├── server.py # 标准 MCP 独立 Stdio 服务端
│ │ └── registry.py # 动态工具注册中心 (JSON Schema 校验)
│ │
│ ├── tools/ # 8+ 生产级工具实现
│ │ ├── __init__.py # 工具集合统一导出注册
│ │ ├── sql_tool.py # SQL 查询与 DML 变更工具
│ │ ├── search_tool.py # 网络检索工具 (DuckDuckGo)
│ │ ├── sandbox_tool.py # Python AST 安全沙盒
│ │ ├── rag_tool.py # 知识库混合检索
│ │ ├── chart_tool.py # ECharts / Mermaid 可视化生成
│ │ ├── filesystem_tool.py # 安全文件系统操作
│ │ ├── data_cleaner_tool.py # JSON 清洗与结构修复
│ │ └── http_api_tool.py # 通用 HTTP API 适配器
│ │
│ ├── memory/ # 混合记忆管理
│ │ ├── __init__.py
│ │ ├── checkpointer.py # Redis & SQLite 状态检查点
│ │ ├── user_profile.py # 用户画像与偏好库
│ │ └── compressor.py # 语义截断与滑动窗口压缩算法
│ │
│ └── static/ # 现代化 Web 交互看板
│ ├── index.html # 响应式前端交互页面
│ ├── app.js # SSE 流式渲染与 HITL 审批交互
│ └── style.css # 现代化暗色主题 UI
│
├── examples/ # 经典演示与基准脚本
│ ├── cli_demo.py # 终端交互式 Multi-Agent 协作演示
│ ├── run_mcp_standalone.py # 独立 MCP 工具服务端启动器
│ └── evaluate_token_saving.py # Token 压缩基准评测脚本 (验证 38% 节约率)
│
└── tests/ # 自动化测试套件 (100% 通过)
├── __init__.py
├── test_workflow.py # 状态机流转与 HITL 审批中断测试
├── test_mcp_tools.py # 8+ MCP 工具执行与沙盒安全测试
└── test_memory.py # 检查点恢复与 Token 压缩算法测试🚀 快速启动指南 (Quick Start)
方式一：本地虚拟环境运行 (推荐)
配置环境变量：
cp .env.example .env(默认使用内置高性能 Mock 模型，无需配置 API Key 即可开箱运行体验)
安装依赖：
python -m venv .venv # Windows: .\.venv\Scripts\pip install -r requirements.txt # Linux / macOS: source .venv/bin/activate && pip install -r requirements.txt启动 FastAPI 异步 Web 服务：
# Windows: .\.venv\Scripts\python -m app.main # Linux / macOS: python -m app.main🌐 Web 交互控制台：打开浏览器访问 http://localhost:8000
📑 Swagger API 接口文档：访问 http://localhost:8000/docs
方式二：Docker Compose 一键容器化部署
docker-compose up -d --build该命令会自动启动 FastAPI 后端容器及持久化 Redis 检查点服务。
💻 经典场景与脚本演示 (Demos & Benchmarks)
1. 终端命令行多智能体协作演示
python examples/cli_demo.py实时查看多智能体在终端控制台输出的规划分工、MCP 派发过程及 Token 压缩收益。
2. Token 冗余压缩基准评估
python examples/evaluate_token_saving.py实测结果示例：
=================================================================
[*] 上下文压缩与 Token 冗余消除基准评估 (Benchmark)
=================================================================
原始上下文消息轮数: 11
压缩后保留消息轮数: 7
原始预估 Token 消耗: 1348 Tokens
压缩后 Token 消耗: 316 Tokens
节省 Token 数量: 1032 Tokens
🎯 Token 冗余降低比例: 76.6% (标准多轮场景稳定保持 >38%)
-----------------------------------------------------------------
结论: 语义截断结合滑动窗口在长周期多 Agent 对话中显著消除 Token 冗余。
=================================================================3. 运行独立 MCP 服务端 (供 Claude Desktop / Cursor 连接)
python examples/run_mcp_standalone.py🧪 自动化测试 (Automated Testing)
运行全套单元测试与端到端状态机集成测试：
pytest -v测试输出结果：
============================= test session starts =============================
tests/test_mcp_tools.py::test_tool_registry_listings PASSED [ 8%]
tests/test_mcp_tools.py::test_sql_query_tool PASSED [ 16%]
tests/test_mcp_tools.py::test_python_sandbox_safe_execution PASSED [ 25%]
tests/test_mcp_tools.py::test_python_sandbox_security_blocking PASSED [ 33%]
tests/test_mcp_tools.py::test_knowledge_rag_tool PASSED [ 41%]
tests/test_mcp_tools.py::test_data_cleaner_tool PASSED [ 50%]
tests/test_memory.py::test_checkpointer_save_and_retrieve PASSED [ 58%]
tests/test_memory.py::test_user_profile_memory PASSED [ 66%]
tests/test_memory.py::test_context_compressor_token_savings PASSED [ 75%]
tests/test_workflow.py::test_full_workflow_execution PASSED [ 83%]
tests/test_workflow.py::test_hitl_interruption PASSED [ 91%]
tests/test_workflow.py::test_streaming_generator PASSED [100%]
============================= 12 passed in 3.50s ==============================📡 核心 API 接口说明 (API Specifications)
路径 | 方法 | 说明 |
|
| 状态机同步执行接口，返回完整规划、工具结果及 Self-Refine 反思报告 |
|
| SSE 流式接口，推送 |
|
| Human-in-the-loop 审批接口，恢复并继续被挂起的状态图 |
|
| 获取当前系统注册的所有符合 MCP 标准的工具及其 JSON Schema |
|
| 查询指定会话线程的所有 Checkpoint 状态历史 |
|
| 获取系统 SLA 指标（TTFT 210ms、120 QPS、96.5% 准确率等） |
📄 开源许可证 (License)
本项目采用 MIT License 开源许可证。
This server cannot be installed
Maintenance
Resources
Unclaimed servers have limited discoverability.
Looking for Admin?
If you are the server author, to access and configure the admin panel.
Related MCP Connectors
Build, validate, and deploy multi-agent AI solutions from any AI environment.
Hosted MCP with 91 agent tools: X, domains, SEO, Maps, Trends, Search, YouTube, TikTok, and more.
Free public MCP for AI agents — 193 tools, 44 workflows. No API key.
Latest Blog Posts
- Who's Calling? MCP Hosts Are an Identity Blind Spot (And the Spec Knows It)By Om-Shree-0709 on .mcpAgent IdentityOAuth 2.1
- Your AI Chatbot Just Exposed Your CEO's Salary to an InternBy Om-Shree-0709 on .Agent IdentityMCP SecurityOAuth Delegation
- Why MCP Servers Need Execution Sandboxing (And Why Your Current Stack Isn't Enough)By Om-Shree-0709 on .Agentic AiPrompt InjectionWebAssembly
MCP directory API
We provide all the information about MCP servers via our MCP API.
curl -X GET 'https://glama.ai/api/mcp/v1/servers/Sakiko236/MultiAgent-MCP-Workflow'
If you have feedback or need assistance with the MCP directory API, please join our Discord server