多智能体协调协议

一个基于FastAPI和SSE的多智能体协调协议(MCP)演示,支持共享状态、规划、验证和工具使用。

  • by omar-florez
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 omar-florez/mcp_demo

使用说明

模型上下文协议(MCP)演示

该存储库展示了使用最小模型上下文协议(MCP)架构的多代理协调,该架构由FastAPI和服务器发送事件(SSE)构建。每个示例都实现了具有共享状态、计划、验证和工具使用的结构化代理协作。

特性

  • 多代理协调(计划员、执行员、评论家)
  • 通过结构化上下文对象共享内存
  • 流式更新的服务器发送事件(SSE)
  • 真正的工具集成(例如,维基百科API、数学表达式评估)
  • 模块化和可扩展的代理定义

例子

1. sse_server.py

说明:
使用SSE计算简单数学表达式的单代理演示。

运行:

python sse_server.py

测试:

curl "http://localhost:8000/sse?task=2*(3+4)"

2. sse_server_planner_executor.py

说明:
两个代理之间的交互 PlannerAgent 和a ExecutorAgent规划器定义计算计划;执行器使用Python的 eval().

运行:

python sse_server_planner_executor.py

测试:

curl "http://localhost:8000/sse?task=3*(5+1)"

3. sse_server_planner_executor_wikipedia.py

说明:
完整的MCP管道 PlannerAgent, ExecutorAgent以及维基百科的整合。计划者提出一个摘要检索任务,执行者在验证计划后通过Wikipedia REST API运行该任务。

MCP代理协调图 下图说明了任务如何通过模型上下文协议从用户流向代理:

+---------+        +--------+        +---------+        +--------+
|  User   | -----> | Planner| -----> | Critic  | -----> | Executor|
+---------+        +--------+        +---------+        +--------+
     |                 |                  |                  |
     |  Task:          |                  |                  |
     |  "Alan_Turing"  |                  |                  |
     |                 |     Plan:        |                  |
     |                 |  "Fetch from     |                  |
     |                 |   Wikipedia +    |                  |
     |                 |   validate bias" |                  |
     |                 | ---------------->                  |
     |                 |                  | Validates plan   |
     |                 |                  | ---------------->|
     |                 |                  |                  |
     |                 |                  |     Executes     |
     |                 |                  |   Wikipedia API  |
     |                 |                  |        |         |
     |                 |                  |        V         |
     |                 |                  |   Tool Output    |
     |                 |                  |  "Alan Turing was..." |
     |                 |                  |                  |
     |<------------------------------------------------------|
     |            Streams live result via SSE               |

运行:

python sse_server_planner_executor_wikipedia.py

测试:

curl "http://localhost:8000/sse?task=Alan_Turing"

运作原理

所有示例都使用FastAPI EventSourceResponse 模拟长时间运行的代理工作流。代理人通过a SharedState 包含任务元数据、计划内容、审批标志和工具结果的对象。

依赖项

pip install fastapi sse-starlette uvicorn nest_asyncio requests

许可证

MIT许可证