Multi Agent Task Automation MCP

基于MCP的多Agent任务自动化平台

  

摘要

A生产级 多智能体任务自动化平台 这演示了使用 模型上下文协议（MCP）系统通过Redis发布/订阅的异步消息传递来编排研究、总结和验证工作流，在PostgreSQL中持久化状态，并通过FastAPI公开管理API。

关键成就：减少手动工作流程时间 小时到分钟 通过智能代理协作自动化数据收集、处理和验证。

建筑

graph TB
    subgraph "Client Layer"
        Client[API Client]
    end
    
    subgraph "API Layer"
        FastAPI[FastAPI Application]
    end
    
    subgraph "Orchestration"
        Coordinator[Agent Coordinator]
        WorkflowRunner[Workflow Runner]
    end
    
    subgraph "Agent Layer"
        Researcher[Researcher Agent]
        Summarizer[Summarizer Agent]
        Validator[Validator Agent]
    end
    
    subgraph "Infrastructure"
        Redis[(Redis Pub/Sub)]
        PostgreSQL[(PostgreSQL)]
    end
    
    subgraph "Protocol"
        MCP[MCP Protocol Layer]
    end
    
    Client --> FastAPI
    FastAPI --> WorkflowRunner
    FastAPI --> Coordinator
    
    WorkflowRunner --> Redis
    Coordinator --> Researcher
    Coordinator --> Summarizer
    Coordinator --> Validator
    
    Researcher --> MCP
    Summarizer --> MCP
    Validator --> MCP
    
    MCP --> Redis
    
    Researcher --> PostgreSQL
    Summarizer --> PostgreSQL
    Validator --> PostgreSQL
    
    WorkflowRunner --> PostgreSQL

代理工作流

端到端管道

1. 工作流启动

- 用户通过REST API发送查询 - 工作流运行器创建上下文并发布到 researcher_input 频道

1. 研究阶段 （研究员代理）

- 收听 researcher_input 频道 - 从多个来源收集数据 - 将结果发布到 summarizer_input 通过MCP - 将活动记录到PostgreSQL

1. 总结阶段 （汇总代理）

- 收听 summarizer_input 频道 - 使用DistilBART模型生成简明摘要 - 将摘要发布到 validator_input - 在PostgreSQL中存储摘要

1. 验证阶段 （验证代理）

- 收听 validator_input 频道 - 进行质量检查（长度、相关性、内容） - 在PostgreSQL中将任务状态更新为“已完成” - 存储验证报告

1. 完成

- 收集的指标（延迟、吞吐量、成功率） - 可通过API获得结果

技术栈

• 后端：Python 3.10+、FastAPI、Uvicorn
• 消息传递：Redis（异步发布/订阅）
• 数据库：PostgreSQL 15（SQLAlchemy ORM）
• 毫升：拥抱面部变压器（DistilBART）
• 容器化：Docker，Docker Compose
• 测试：pytest、httpx
• CI/CD：GitHub操作

快速开始

先决条件

• Docker&Docker编写
• Python 3.10+（用于本地开发）
• Git

使用Docker Compose运行（推荐）

# Clone the repository
git clone https://github.com/PremC1F/Multi-Agent-Task-Automation-MCP.git
cd Multi-Agent-Task-Automation-MCP

# Start all services
docker-compose up --build

# The API will be available at http://localhost:8000

本地运行

# Install dependencies
pip install -r requirements.txt

# Set up environment variables
cp .env.example .env

# Start PostgreSQL and Redis (via Docker)
docker-compose up postgres redis -d

# Run the application
python -m src.main

API 文档

跑步后，请访问：

• 交互式API文档: http://localhost:8000/docs
• 备选文档: http://localhost:8000/redoc

关键终点

1.启动工作流

POST /api/v1/task/start
Content-Type: application/json

{
  "query": "machine learning"
}

响应:

{
  "context_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "message": "Workflow started: a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "query": "machine learning"
}

2.获取任务状态

GET /api/v1/task/{context_id}

响应:

{
  "context_id": "a1b2c3d4-e5f6-7890-abcd-ef1234567890",
  "status": "completed",
  "success": true,
  "created_at": "2024-11-15T20:00:00",
  "completed_at": "2024-11-15T20:00:05",
  "error_message": null,
  "results_count": 2,
  "metrics": {
    "total_duration": 5.2,
    "agent_timings": {
      "researcher_agent": 1.5,
      "summarizer_agent": 2.8,
      "validator_agent": 0.9
    },
    "message_count": 3
  }
}

3.代理状态

GET /api/v1/agents

响应:

{
  "agents": {
    "researcher": {
      "name": "researcher_agent",
      "running": true,
      "input_channel": "researcher_input",
      "output_channel": "summarizer_input"
    },
    "summarizer": {
      "name": "summarizer_agent",
      "running": true,
      "input_channel": "summarizer_input",
      "output_channel": "validator_input"
    },
    "validator": {
      "name": "validator_agent",
      "running": true,
      "input_channel": "validator_input",
      "output_channel": null
    }
  },
  "total": 3
}

4.系统指标

GET /api/v1/metrics

响应:

{
  "total_workflows": 42,
  "completed": 40,
  "successful": 38,
  "success_rate": 0.95,
  "avg_duration": 5.3,
  "avg_message_latency": 0.05,
  "total_messages": 126
}

示例用法

使用cURL

# Start a workflow
curl -X POST http://localhost:8000/api/v1/task/start \
  -H "Content-Type: application/json" \
  -d '{"query": "autonomous agents"}'

# Get task status (replace with actual context_id)
curl http://localhost:8000/api/v1/task/a1b2c3d4-e5f6-7890-abcd-ef1234567890

# Check agent health
curl http://localhost:8000/api/v1/health

# View metrics
curl http://localhost:8000/api/v1/metrics

使用Python

import requests

# Start workflow
response = requests.post(
    "http://localhost:8000/api/v1/task/start",
    json={"query": "distributed systems"}
)
context_id = response.json()["context_id"]
print(f"Workflow started: {context_id}")

# Check status
import time
time.sleep(5)  # Wait for completion

status = requests.get(f"http://localhost:8000/api/v1/task/{context_id}")
print(status.json())

性能指标

典型工作流性能

节省时间

• 手动研究：30-60分钟
• 自动化流水线：5-7秒
• 减少: 速度提高了约99.8% 🚀

发展

运行测试

# Run all tests
pytest tests/ -v

# Run specific test file
pytest tests/test_agents.py -v

# Run with coverage
pytest tests/ --cov=src --cov-report=html

代码质量

# Format code
black src/ tests/

# Lint
flake8 src/ tests/

# Type checking
mypy src/

项目结构

Multi-Agent-Task-Automation-MCP/
├── src/
│   ├── main.py                 # FastAPI entrypoint
│   ├── agents/
│   │   ├── base_agent.py       # Abstract agent class
│   │   ├── researcher_agent.py # Data collection agent
│   │   ├── summarizer_agent.py # Summarization agent
│   │   ├── validator_agent.py  # Validation agent
│   │   └── coordinator.py      # Agent orchestrator
│   ├── core/
│   │   ├── mcp_protocol.py     # MCP message protocol
│   │   ├── redis_manager.py    # Redis pub/sub manager
│   │   ├── db_manager.py       # PostgreSQL ORM
│   │   ├── workflow_runner.py  # Workflow orchestration
│   │   └── config.py           # Configuration
│   ├── api/
│   │   ├── routes.py           # API endpoints
│   │   └── schemas.py          # Pydantic models
│   └── utils/
│       ├── logger.py           # Centralized logging
│       └── metrics.py          # Metrics collection
├── tests/                      # Test suite
├── Dockerfile                  # Container definition
├── docker-compose.yml          # Multi-service setup
└── requirements.txt            # Python dependencies

环境变量

创建一个 .env 来自文件 .env.example:

POSTGRES_HOST=postgres
POSTGRES_PORT=5432
POSTGRES_DB=mcp_db
POSTGRES_USER=mcp_user
POSTGRES_PASSWORD=mcp_password

REDIS_HOST=redis
REDIS_PORT=6379
REDIS_DB=0

API_PORT=8000
LOG_LEVEL=INFO
ENVIRONMENT=development

架构演变

1. 第2阶段：使用Helm charts进行Kubernetes部署
2. 第三期：服务网格集成（Istio）
3. 阶段4：事件来源和CQRS模式
4. 阶段5：联合多租户架构

故障排除

常见问题

问题：代理未启动

# Check Redis connectivity
docker-compose logs redis

# Restart services
docker-compose restart

问题：数据库连接错误

# Check PostgreSQL health
docker-compose exec postgres pg_isready

# View logs
docker-compose logs postgres

问题：工作流超时

• 增加超时时间 workflow_runner.py
• 检查代理日志是否有错误
• 验证Redis发布/订阅频道

项目链接：

______________________________________________________________________

内置于❤️ 用于自主系统和智能自动化