A
licenseNot graded
qualityC
maintenanceAn advanced MCP-based AI agent system with intelligent tool orchestration, multi-LLM support, and enterprise-grade reliability features like semantic routing and circuit breakers.
MIT
MultiAgent-MCP-Workflow
by Sakiko236
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% 차단되며 휴먼 승인을 거칩니다.
Related MCP server: MCP Business AI Transformation
🏗️ 전체 아키텍처 설계 (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🛠️ 4대 핵심 모듈 상세 (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를 기반으로 사용자의 기술 스택 선호도, 출력 스타일 제약, 이전 의사결정 행동을 자동으로 유지하고, 다중 에이전트 시작 시 필요에 따라 컨텍스트에 주입합니다.
Token 중복 압축 알고리즘(Context Compressor):
슬라이딩 윈도우 메커니즘: 시스템 지시와 최근의 대화 $K$ 턴을 고정적으로 보존합니다.
시맨틱 트러스케이션(Semantic Truncation): 만료되거나 길고 성가신 중간 툴 출력(예: 수백 개 레코드가 포함된 SQL 원시 결과)은 코어 Schema와 요약을 자동 추출하여, 다중 턴 세션의 Token 중복을 38% 이상 절감합니다.
4. 프로덕션급 스트리밍 추론 및 동시성 최적화 (FastAPI + AsyncIO + SSE)
완전 비동기 비차단 아키텍처: FastAPI + AsyncIO 이벤트 루프를 채용하여 높은 처리량(120+ QPS)을 구현합니다.
SSE 이벤트 스트림 미세 푸시:
thought: 각 에이전트 노드의 사고 과정과 결정 로직를 실시간 푸시.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)
방법 1: 로컬 가상 환경 실행(권장)
환경 변수 설정:
cp .env.example .env*(기본적으로 내장된 준수한 Mock 모델을 사용하므로, API Key를 설정하지 않아도 바로 실행 체험이 가능합니다)*2. 의존성 설치:
python -m venv .venv
# Windows:
.\.venv\Scripts\pip install -r requirements.txt
# Linux / macOS:
source .venv/bin/activate && pip install -r requirements.txtFastAPI 비동기 Web 서비스 시작:
# Windows:
.\.venv\Scripts\python -m app.main
# Linux / macOS:
python -m app.main* 🌐 **Web 인터랙티브 콘솔**: 브라우저에서 <http://localhost:8000> 접속
* 📑 **Swagger API 문서**: <http://localhost:8000/docs> 접속방법 2: 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 deployed
Maintenance
ActivityMaintained
ResponsivenessSyncing
Related MCP Connectors
MCP server for your apps' tools and custom tools, plus hosted AI agents and approval-gated workflows
One MCP endpoint for Claude, GPT & Gemini: 100+ tools + no-code connectors + agent workers.
Zero-secret MCP gateway for AI agents: risk-scored, audited calls with human-in-the-loop approval.
Let AI agents query data and act across all your business apps via MCP.
Related MCP Servers
- FlicenseNot gradedqualityCmaintenanceEnterprise-grade MCP server with multi-agent system for business AI transformation across finance, healthcare, retail, and other domains. Provides specialized AI agents for data analysis, API execution, business validation, and report generation with real-time monitoring and observability.-
- AlicenseNot gradedqualityDmaintenanceEnables multi-model leader-worker agent orchestration, workflow execution, and deterministic validation via structured MCP tools.2 npmApache 2.0
- AlicenseNot gradedqualityCmaintenanceAn enterprise MCP server that exposes 16 standardized tools for document intelligence, RAG, knowledge graph, SQL analysis, LLM evaluation, cost estimation, and AI architecture design, enabling AI agents to securely access and compose enterprise AI capabilities.1MIT