Content
# 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
1. **DataAgent**: Especializado en recuperación de datos
- Conecta con servidores MCP
- Expone datos a otros agentes vía A2A
- Capacidad: `DATA_RETRIEVAL`
2. **AnalysisAgent**: Especializado en análisis
- Analiza datos de otros agentes
- Genera insights y resúmenes
- Capacidades: `DATA_ANALYSIS`, `REASONING`
3. **OrchestratorAgent**: Coordina flujos complejos
- Coordina múltiples agentes
- Ejecuta workflows
- Gestiona el ciclo de vida
## Instalación
1. Clonar o descargar el proyecto
2. Instalar dependencias:
```bash
pip install -r requirements.txt
```
3. Configurar variables de entorno (opcional):
```bash
cp .env.example .env
# Editar .env con tus configuraciones
```
## Uso
### 1. Iniciar el servidor MCP mock
En una terminal:
```bash
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:
```bash
python example_usage.py
```
## Ejemplos Incluidos
### Ejemplo 1: Consulta Básica de Clima
```python
# El DataAgent recupera datos del servidor MCP
result = await orchestrator.data_agent.get_weather_data("tokyo")
```
### Ejemplo 2: Consulta con Análisis
```python
# Workflow completo: MCP → DataAgent → AnalysisAgent
result = await orchestrator.get_and_analyze_weather("london")
```
### Ejemplo 3: Comparación de Múltiples Ciudades
```python
# Procesamiento paralelo de múltiples fuentes
cities = ["new_york", "london", "tokyo"]
result = await orchestrator.compare_multiple_cities(cities)
```
### Ejemplo 4: Listar Recursos MCP
```python
# Descubrir recursos disponibles en el servidor MCP
result = await orchestrator.get_mcp_resources()
```
### Ejemplo 5: Estado de Agentes
```python
# 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
1. **Inicialización**: El DataAgent se conecta al servidor MCP
2. **Descubrimiento**: Obtiene lista de recursos y herramientas disponibles
3. **Consumo**: Lee recursos o ejecuta herramientas según sea necesario
4. **Procesamiento**: Devuelve datos estructurados
### Flujo A2A
1. **Registro**: Los agentes se registran entre sí compartiendo capacidades
2. **Request**: Un agente envía un mensaje de solicitud a otro
3. **Processing**: El agente receptor procesa la solicitud
4. **Response**: Devuelve una respuesta al emisor
5. **Timeout**: Sistema de timeout para evitar bloqueos
### Coordinación
El OrchestratorAgent coordina flujos complejos:
1. Solicita datos al DataAgent (que consulta MCP)
2. Envía datos al AnalysisAgent para análisis
3. Agrega y devuelve resultados finales
## Personalización
### Crear un Nuevo Agente
```python
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
```python
# Cambia la URL del servidor MCP
orchestrator = OrchestratorAgent(
mcp_server_url="https://your-mcp-server.com"
)
```
## Características Avanzadas
### Procesamiento Paralelo
```python
# 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
```python
# Buscar agentes por capacidad
agents = protocol.get_agents_by_capability(
AgentCapability.DATA_ANALYSIS
)
```
### Timeout y Manejo de Errores
```python
# 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
- [Model Context Protocol (MCP)](https://modelcontextprotocol.io/)
- [Agent-to-Agent Communication Patterns](https://www.anthropic.com/)
- [Anthropic Claude](https://claude.ai/)
## Licencia
Este es un proyecto de ejemplo con fines educativos.
# A2A---MCP
