agent mcp框架
用于构建多代理MCP（模型上下文协议）服务器的Python框架。
使用顺序、并行、条件和映射缩减流水线模式、代理生命周期挂钩和结构化输出组合多代理系统，并将结果作为MCP工具公开给Claude、VSCode或任何兼容MCP的客户端。
特性
- 代理抽象 —
Agent,LLMAgent,FunctionAgent带有生命周期挂钩 - 管道组成 --顺序、并行、条件和MapReduce模式
- MCP集成 --将代理管道作为MCP工具在stdio或SSE上公开
- 输出格式化 --JSON、Markdown和纯文本输出模式
- 命令行界面 --从命令行运行服务器和管道
- Docker就绪 --Dockerfile、健康检查和AWS ECS Fargate部署
- CI/CD --GitHub测试、linting和部署操作
安装
pip install agent-mcp-framework快速开始
from agent_mcp_framework import Agent, AgentContext, AgentResult, SequentialPipeline, AgentMCPServer
class AnalyzerAgent(Agent):
async def run(self, context: AgentContext) -> AgentResult:
code = context.get("code", "")
issues = []
if len(code.splitlines()) > 500:
issues.append("File exceeds 500 lines — consider splitting")
if "import *" in code:
issues.append("Wildcard imports detected")
context.set("issues", issues)
return AgentResult(success=True, output={"issues": issues, "count": len(issues)})
class ScorerAgent(Agent):
async def run(self, context: AgentContext) -> AgentResult:
issues = context.get("issues", [])
score = max(0, 100 - len(issues) * 15)
return AgentResult(success=True, output={"score": score, "grade": "A" if score >= 90 else "B" if score >= 70 else "C"})
# Compose agents into a pipeline
pipeline = SequentialPipeline("code-review", agents=[
AnalyzerAgent("analyzer", description="Find code issues"),
ScorerAgent("scorer", description="Score code quality"),
])
# Expose as an MCP server
server = AgentMCPServer("code-review-server", description="Multi-agent code review")
server.add_pipeline_tool(
pipeline,
name="review_code",
description="Analyze code quality and return a score",
)
if __name__ == "__main__":
server.run() # Starts MCP server on stdio码头工人
# Build and run locally
docker compose up --build
# Or build manually
docker build -t agent-mcp-framework .
docker run -p 8080:8080 agent-mcp-framework健康检查端点：
| 终点 | 目的 |
|---|---|
GET / | 服务信息 |
GET /health | 活体探针 |
GET /ready | 准备就绪探针 |
代理类型
Agent --基本类
实施 run() 定义代理的逻辑。
LLMAgent --克劳德动力代理人
内置Anthropic客户端 complete() LLM呼叫的助手。
FunctionAgent --快速内联代理
将任何异步函数包装为代理，而不进行子类化。
管道模式
| 图案 | 描述 |
|---|---|
SequentialPipeline | 逐一运行代理，每个代理都看到更新的上下文 |
ParallelPipeline | 在可选的并发限制下并发运行代理 |
ConditionalPipeline | 基于条件函数的代理路由 |
MapReducePipeline | 拆分工作，扇出并减少结果 |
命令行界面
# Start an MCP server
agent-mcp serve my_project.server
# Run a pipeline directly
agent-mcp run my_project.pipeline --input '{"code": "import *"}'
# Show framework info
agent-mcp info部署
该项目包括完整的AWS ECS Fargate部署配置：
Dockerfile--带健康检查的多阶段构建docker-compose.yml--地方发展.github/workflows/ci.yml--CI管道（lint、测试、Docker构建）.github/workflows/deploy.yml--CD管道（ECR推送、ECS部署）aws/task-definition.json--ECS Fargate任务定义
许可证
麻省理工学院