🇻🇪🇨🇱 Dev.to Linkedin GitHub Twitter Instagram Youtube
Linktr
Getting Started with Strands Agents: Build Your First AI Agent - Curso Gratis
La comunicación agente a agente (A2A) representa la próxima evolución en la automatización de IA, donde múltiples agentes especializados colaboran para resolver problemas complejos. Con el framework Strands Agent, puedes construir sistemas multi-agente que se coordinan de manera fluida para manejar tareas más allá de las capacidades de agentes individuales.
En este artículo, aprenderás cómo crear agentes que se comunican entre sí, comparten información y trabajan juntos para realizar flujos de trabajo complejos utilizando las nuevas herramientas Strands A2A.
⭐ Entendiendo la Comunicación Agente a Agente
El protocolo Agente a Agente es un estándar abierto que define cómo los agentes de IA pueden descubrirse, comunicarse y colaborar entre sí. El soporte del protocolo A2A permite varios casos de uso potentes:
- Flujos de Trabajo Multi-Agente: Encadena múltiples agentes especializados
- Mercados de Agentes: Descubre y utiliza agentes de diferentes proveedores
- Integración Multi-Plataforma: Conecta agentes Strands con otros sistemas compatibles con A2A
- Sistemas de IA Distribuidos: Construye arquitecturas de agentes escalables y distribuidas
MCP vs A2A
Es importante entender cómo se relaciona A2A con MCP (Model Context Protocol). Estos son estándares complementarios.
- MCP conecta agentes a herramientas, APIs y recursos con entradas y salidas estructuradas. Considéralo como la manera en que los agentes acceden a sus capacidades.
- A2A facilita la comunicación dinámica entre diferentes agentes como pares. Es cómo los agentes colaboran, delegan y gestionan tareas compartidas.
Son complementarios: MCP proporciona herramientas a los agentes, mientras que A2A facilita la colaboración.
Arquitectura A2A
Los componentes principales incluyen:
- Agente Primario: Inicia la comunicación y delega tareas (el "gerente")
- Agente(s) Secundario(s): Recibe tareas y proporciona respuestas (los "especialistas")
- Herramienta A2A: Maneja los detalles del protocolo para una comunicación fluida entre los agentes
- Protocolo de Mensajes: Define el formato y estructura de los mensajes
🚀 Comenzando
Clona el repositorio de ejemplo:
git clone https://github.com/elizabethfuentes12/strands-agent-samples
cd notebook
Crea y activa un entorno virtual:
python -m venv .venv
source .venv/bin/activate # En Windows: .venv\Scripts\activate
Instala las dependencias requeridas:
pip install -r requirements.txt
🔧 Creando tu Primer Sistema Agente a Agente
En esta sección, exploraremos bloques de código seleccionados para construir strands-a2a-inter-agent, basado en el repositorio sample-agentic-ai-demos (gracias James Ward).
Nuestro ejemplo muestra una arquitectura de tres niveles:
- Un HR Agent se comunica con un Agente de Empleados
- El Employee Agent se comunica con un Servidor MCP
- El MCP Server proporciona acceso a datos de empleados
1. Datos de Empleados
Primero, tenemos un módulo simple que genera datos aleatorios de empleados con habilidades:
import random
FIRST_NAMES = ["James", "Mary", "John", "Patricia", "Robert", "Jennifer", "Michael", "Linda", "William", "Elizabeth"]
LAST_NAMES = ["Smith", "Johnson", "Williams", "Brown", "Jones", "Garcia", "Miller", "Davis", "Rodriguez", "Martinez"]
SKILLS = {
"Kotlin", "Java", "Python", "JavaScript", "TypeScript",
"React", "Angular", "Spring Boot", "AWS", "Docker",
"Kubernetes", "SQL", "MongoDB", "Git", "CI/CD",
"Machine Learning", "DevOps", "Node.js", "REST API", "GraphQL"
}
EMPLOYEES = list({emp["name"]: emp for emp in [
{
"name": f"{random.choice(FIRST_NAMES)} {random.choice(LAST_NAMES)}",
"skills": random.sample(list(SKILLS), random.randint(2, 5))
}
for i in range(100)
]}.values())
2. Servidor MCP
A continuación, tenemos un servidor MCP que expone herramientas para acceder a los datos de empleados:
from mcp.server.fastmcp import FastMCP
from employee_data import SKILLS, EMPLOYEES
mcp = FastMCP("employee-server", stateless_http=True, host="0.0.0.0", port=8002)
@mcp.tool()
def get_skills() -> set[str]:
"""todas las habilidades que los empleados pueden tener - usa esta lista para averiguar habilidades relacionadas"""
print("get_skills")
return SKILLS
@mcp.tool()
def get_employees_with_skill(skill: str) -> list[dict]:
"""empleados que tienen una habilidad específica - la salida incluye nombre completo (Nombre Apellido) y sus habilidades"""
print(f"get_employees_with_skill({skill})")
skill_lower = skill.lower()
employees_with_skill = [employee for employee in EMPLOYEES if any(s.lower() == skill_lower for s in employee["skills"])]
if not employees_with_skill:
raise ValueError(f"Ningún empleado tiene la habilidad {skill}")
return employees_with_skill
if __name__ == "__main__":
mcp.run(transport="streamable-http")
El servidor MCP expone dos herramientas:
-
get_skills(): Devuelve todas las habilidades disponibles -
get_employees_with_skill(skill): Devuelve empleados con una habilidad específica
Estas herramientas se exponen a través del Protocolo de Contexto de Modelo (MCP) usando el framework FastMCP, que proporciona una forma estandarizada para que los agentes accedan a estas funciones.
3. Agente de Empleados
El Agente de Empleados se conecta al servidor MCP y expone sus capacidades a través de un servidor A2A:
import os
from mcp.client.streamable_http import streamablehttp_client
from strands import Agent
from strands.tools.mcp.mcp_client import MCPClient
from strands.multiagent.a2a import A2AServer
from urllib.parse import urlparse
from strands.models.anthropic import AnthropicModel
# Define las URLs correctamente
EMPLOYEE_INFO_URL = "http://localhost:8002/mcp/"
EMPLOYEE_AGENT_URL = "http://localhost:8001/"
# Crea el cliente MCP
employee_mcp_client = MCPClient(lambda: streamablehttp_client(EMPLOYEE_INFO_URL))
model = AnthropicModel(
client_args={
"api_key": "TU_CLAVE_API_AQUI", # Reemplaza con tu clave API
},
max_tokens=1028,
model_id="claude-sonnet-4-20250514",
params={
"temperature": 0.7,
}
)
# Usa el cliente MCP dentro de un contexto
with employee_mcp_client:
tools = employee_mcp_client.list_tools_sync()
# Crea un agente Strands
employee_agent = Agent(
model=model,
name="Agente de Empleados",
description="Responde preguntas sobre empleados",
tools=tools,
system_prompt="debes abreviar los nombres de los empleados y listar todas sus habilidades"
)
# Crea servidor A2A
a2a_server = A2AServer(
agent=employee_agent,
host=urlparse(EMPLOYEE_AGENT_URL).hostname,
port=int(urlparse(EMPLOYEE_AGENT_URL).port)
)
# Inicia el servidor
if __name__ == "__main__":
a2a_server.serve(host="0.0.0.0", port=8001)
Opciones de Configuración del Servidor
El A2AServer acepta varias opciones de configuración:
-
agent: El Agente Strands a envolver con compatibilidad A2A -
host: Nombre de host o dirección IP para vincular (predeterminado: "0.0.0.0") -
port: Puerto para vincular (predeterminado: 9000) -
version: Versión del agente (predeterminado: "0.0.1") -
skills: Lista personalizada de habilidades del agente (predeterminado: auto-generado desde herramientas) -
http_url: URL HTTP pública donde este agente será accesible (opcional, habilita montaje basado en rutas) -
serve_at_root: Fuerza al servidor a servir en la ruta raíz sin importar la ruta http_url (predeterminado: False)
4. Agente de RRHH
Finalmente, el Agente de RRHH proporciona una API orientada al usuario y se comunica con el Agente de Empleados.
- Crea una aplicación FastAPI para manejar peticiones HTTP
- Usa el A2AClientToolProvider para descubrir e interactuar con agentes A2A sin escribir código de cliente manualmente
- Proporciona un endpoint para que los usuarios hagan preguntas sobre empleados
Este agente sirve como punto de entrada para consultas de usuarios y delega tareas especializadas al Agente de Empleados cuando es necesario.
import os
import uvicorn
from strands import Agent
from strands.models import BedrockModel
from strands_tools.a2a_client import A2AClientToolProvider
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
EMPLOYEE_AGENT_URL = "http://localhost:8001/"
app = FastAPI(title="API del Agente de RRHH")
class QuestionRequest(BaseModel):
question: str
@app.get("/health")
def health_check():
return {"status": "healthy"}
model = AnthropicModel(
client_args={
"api_key": os.getenv("api_key"),
},
max_tokens=1028,
model_id="claude-3-7-sonnet-20250219",
params={
"temperature": 0.3,
}
)
@app.post("/inquire")
async def ask_agent(request: QuestionRequest):
async def generate():
provider = A2AClientToolProvider(known_agent_urls=[EMPLOYEE_AGENT_URL])
agent = Agent(model=bedrock_model, tools=provider.tools)
stream_response = agent.stream_async(request.question)
async for event in stream_response:
if "data" in event:
yield event["data"]
return StreamingResponse(
generate(),
media_type="text/plain"
)
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
✅ Ejecutando el Ejemplo
Opción 1: Ejecutar cada componente en una terminal separada:
- Establece tu clave API:
export api_key='tu-clave-api-anthropic-aqui'
- Instala los Paquetes Requeridos:
pip install -r requirements.txt
- Inicia el Servidor MCP:
python3 strands-a2a-inter-agent/server.py
- Inicia el Agente de Empleados:
python employee-agent.py
- Inicia el Agente de RRHH:
python hr-agent.py
- Haz una petición al Agente de RRHH: Una vez que los tres componentes estén ejecutándose, puedes hacer peticiones al Agente de RRHH:
curl -X POST --location "http://0.0.0.0:8000/inquire" \
-H "Content-Type: application/json" \
-d '{"question": "lista empleados que tengan habilidades relacionadas con programación de IA"}'
Opción 2: Ejecuta el Servidor MCP, Agente de Empleados y Agente de RRHH en paralelo
- Establece tu clave API:
export api_key='tu-clave-api-anthropic-aqui'
- Instala los Paquetes Requeridos:
pip install -r requirements.txt
- Inicia el sistema:
python run_a2a_system.py
- Haz Peticiones al Agente de RRHH: Una vez que los tres componentes estén ejecutándose, puedes hacer peticiones al Agente de RRHH:
curl -X POST --location "http://0.0.0.0:8000/inquire" \
-H "Content-Type: application/json" \
-d '{"question": "lista empleados que tengan habilidades relacionadas con programación de IA"}'
🎯 Mejores Prácticas para Comunicación A2A
Para implementar sistemas A2A exitosos, considera estas recomendaciones:
- Responsabilidades Claras del Agente: Define roles específicos para cada agente
- Manejo de Errores: Implementa manejo robusto de errores y mecanismos de respaldo
- Estandarización del Formato de Mensajes: Usa formatos de mensaje consistentes entre agentes
- Monitoreo de Rendimiento: Rastrea tiempos de respuesta y tasas de éxito
- Consideraciones de Seguridad: Valida mensajes entre agentes
- Estrategia de Pruebas: Prueba agentes individuales y sus interacciones
🔮 Próximos Pasos
Esta introducción a la comunicación A2A con Strands Tools abre numerosas posibilidades para construir sistemas de IA sofisticados.
El código completo y los ejemplos están disponibles en el repositorio de GitHub. Te invito a experimentar con diferentes configuraciones de agentes y patrones de comunicación para descubrir qué funciona mejor para tus casos de uso.
¿Qué te parece este enfoque? ¿Has trabajado con sistemas multi-agente antes? Comparte tu experiencia en los comentarios.
Top comments (0)