使用说明
Sistema Multi-Agente con MCP y A2A
Este proyecto demuestra una implementación completa de un sistema multi-agente en Python que integra:
- MCP (Model Context Protocol): Para consumir fuentes de datos externas
- A2A (Agent-to-Agent): Para comunicación entre agentes
Arquitectura
┌─────────────────────────────────────────────────────────────┐
│ Orchestrator Agent │
│ (Coordina todos los agentes) │
└────────────┬────────────────────────────────┬────────────────┘
│ │
│ A2A Protocol │ A2A Protocol
│ │
┌────────▼────────┐ ┌───────▼────────┐
│ Data Agent │ │ Analysis Agent │
│ │ │ │
│ - Recuperación │◄────A2A─────►│ - Análisis │
│ de datos │ │ - Razonamiento │
└────────┬────────┘ └────────────────┘
│
│ MCP Protocol
│
┌────────▼────────┐
│ MCP Server │
│ │
│ - Weather API │
│ - Recursos │
│ - Herramientas │
└─────────────────┘
Características
MCP (Model Context Protocol)
- Cliente MCP para conectar con servidores de datos
- Lectura de recursos (resources)
- Ejecución de herramientas (tools)
- Servidor mock de clima incluido para pruebas
A2A (Agent-to-Agent)
- Protocolo de comunicación entre agentes
- Mensajes request/response
- Sistema de capacidades (capabilities)
- Registro y descubrimiento de agentes
- Comunicación asíncrona
Agentes Especializados
-
DataAgent: Especializado en recuperación de datos
- Conecta con servidores MCP
- Expone datos a otros agentes vía A2A
- Capacidad:
DATA_RETRIEVAL
-
AnalysisAgent: Especializado en análisis
- Analiza datos de otros agentes
- Genera insights y resúmenes
- Capacidades:
DATA_ANALYSIS,REASONING
-
OrchestratorAgent: Coordina flujos complejos
- Coordina múltiples agentes
- Ejecuta workflows
- Gestiona el ciclo de vida
Instalación
-
Clonar o descargar el proyecto
-
Instalar dependencias:
pip install -r requirements.txt
- Configurar variables de entorno (opcional):
cp .env.example .env
# Editar .env con tus configuraciones
Uso
1. Iniciar el servidor MCP mock
En una terminal:
python src/mcp/mcp_mock_server.py
Esto iniciará un servidor en http://localhost:8000 que simula una API de clima.
2. Ejecutar los ejemplos
En otra terminal:
python example_usage.py
Ejemplos Incluidos
Ejemplo 1: Consulta Básica de Clima
# El DataAgent recupera datos del servidor MCP
result = await orchestrator.data_agent.get_weather_data("tokyo")
Ejemplo 2: Consulta con Análisis
# Workflow completo: MCP → DataAgent → AnalysisAgent
result = await orchestrator.get_and_analyze_weather("london")
Ejemplo 3: Comparación de Múltiples Ciudades
# Procesamiento paralelo de múltiples fuentes
cities = ["new_york", "london", "tokyo"]
result = await orchestrator.compare_multiple_cities(cities)
Ejemplo 4: Listar Recursos MCP
# Descubrir recursos disponibles en el servidor MCP
result = await orchestrator.get_mcp_resources()
Ejemplo 5: Estado de Agentes
# Verificar estado y capacidades de todos los agentes
status = await orchestrator.get_agent_status()
Estructura del Proyecto
multi-agent/
├── src/
│ ├── mcp/
│ │ ├── mcp_client.py # Cliente MCP
│ │ └── mcp_mock_server.py # Servidor mock de pruebas
│ ├── a2a/
│ │ ├── a2a_protocol.py # Protocolo A2A
│ │ └── a2a_server.py # Servidor A2A
│ └── agents/
│ ├── data_agent.py # Agente de datos
│ ├── analysis_agent.py # Agente de análisis
│ └── orchestrator.py # Orquestador
├── example_usage.py # Ejemplos de uso
├── requirements.txt # Dependencias
└── README.md # Esta documentación
Cómo Funciona
Flujo MCP
- Inicialización: El DataAgent se conecta al servidor MCP
- Descubrimiento: Obtiene lista de recursos y herramientas disponibles
- Consumo: Lee recursos o ejecuta herramientas según sea necesario
- Procesamiento: Devuelve datos estructurados
Flujo A2A
- Registro: Los agentes se registran entre sí compartiendo capacidades
- Request: Un agente envía un mensaje de solicitud a otro
- Processing: El agente receptor procesa la solicitud
- Response: Devuelve una respuesta al emisor
- Timeout: Sistema de timeout para evitar bloqueos
Coordinación
El OrchestratorAgent coordina flujos complejos:
- Solicita datos al DataAgent (que consulta MCP)
- Envía datos al AnalysisAgent para análisis
- Agrega y devuelve resultados finales
Personalización
Crear un Nuevo Agente
from src.a2a.a2a_protocol import A2AProtocol, AgentCapability
class CustomAgent:
def __init__(self, agent_id: str):
self.agent_id = agent_id
self.a2a = A2AProtocol(
agent_id=agent_id,
agent_name="CustomAgent",
capabilities=[AgentCapability.TASK_EXECUTION]
)
self._register_handlers()
def _register_handlers(self):
self.a2a.register_handler("custom_action", self._handle_custom)
async def _handle_custom(self, payload):
# Tu lógica aquí
return {"result": "success"}
Conectar a un Servidor MCP Real
# Cambia la URL del servidor MCP
orchestrator = OrchestratorAgent(
mcp_server_url="https://your-mcp-server.com"
)
Características Avanzadas
Procesamiento Paralelo
# Los agentes pueden procesar múltiples tareas en paralelo
tasks = [agent.process(data) for data in dataset]
results = await asyncio.gather(*tasks)
Sistema de Capacidades
# Buscar agentes por capacidad
agents = protocol.get_agents_by_capability(
AgentCapability.DATA_ANALYSIS
)
Timeout y Manejo de Errores
# Timeout configurable en solicitudes A2A
response = await protocol.request_and_wait(
receiver_id="agent_001",
action="analyze",
params={"data": data},
timeout=30.0 # 30 segundos
)
Solución de Problemas
El servidor MCP no responde
- Verifica que el servidor mock esté ejecutándose
- Revisa la URL en la configuración
- Verifica los logs para errores de conexión
Los agentes no se comunican
- Asegúrate de que los agentes estén registrados entre sí
- Verifica los agent_ids sean correctos
- Revisa que los handlers estén registrados
Timeouts en solicitudes
- Aumenta el timeout en
request_and_wait - Verifica que el agente receptor esté activo
- Revisa los logs para identificar el cuello de botella
Referencias
Licencia
Este es un proyecto de ejemplo con fines educativos.
