DEV Community

Cover image for Construyendo Agentes Strands con pocas líneas de código: Comunicación Agente a Agente (A2A)
AWS Español profile image Elizabeth Fuentes L
Elizabeth Fuentes L for AWS Español

Posted on

Construyendo Agentes Strands con pocas líneas de código: Comunicación Agente a Agente (A2A)

#aws #ai #programming #beginners

🇻🇪🇨🇱 Dev.to Linkedin GitHub Twitter Instagram Youtube
Linktr

elizabethfuentes12 image

Elizabeth Fuentes L

AWS Developer Advocate

Getting Started with Strands Agents: Build Your First AI Agent - Curso Gratis

GitHub repository

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
Enter fullscreen mode Exit fullscreen mode

Crea y activa un entorno virtual:

python -m venv .venv
source .venv/bin/activate  # En Windows: .venv\Scripts\activate
Enter fullscreen mode Exit fullscreen mode

Instala las dependencias requeridas:

pip install -r requirements.txt
Enter fullscreen mode Exit fullscreen mode

🔧 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:

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())
Enter fullscreen mode Exit fullscreen mode

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")
Enter fullscreen mode Exit fullscreen mode

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)
Enter fullscreen mode Exit fullscreen mode

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)
Enter fullscreen mode Exit fullscreen mode

✅ Ejecutando el Ejemplo

Opción 1: Ejecutar cada componente en una terminal separada:

  1. Establece tu clave API: 
export api_key='tu-clave-api-anthropic-aqui'
Enter fullscreen mode Exit fullscreen mode
  1. Instala los Paquetes Requeridos: 
pip install -r requirements.txt
Enter fullscreen mode Exit fullscreen mode
  1. Inicia el Servidor MCP: 
python3 strands-a2a-inter-agent/server.py
Enter fullscreen mode Exit fullscreen mode

  1. Inicia el Agente de Empleados: 
python employee-agent.py
Enter fullscreen mode Exit fullscreen mode
  1. Inicia el Agente de RRHH: 
python hr-agent.py
Enter fullscreen mode Exit fullscreen mode
  1. 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"}'
Enter fullscreen mode Exit fullscreen mode

Opción 2: Ejecuta el Servidor MCP, Agente de Empleados y Agente de RRHH en paralelo

  1. Establece tu clave API: 
export api_key='tu-clave-api-anthropic-aqui'
Enter fullscreen mode Exit fullscreen mode
  1. Instala los Paquetes Requeridos: 
pip install -r requirements.txt
Enter fullscreen mode Exit fullscreen mode
  1. Inicia el sistema: 
python run_a2a_system.py
Enter fullscreen mode Exit fullscreen mode
  1. 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"}'
Enter fullscreen mode Exit fullscreen mode

🎯 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.

📚 Recursos

Top comments (1)

Collapse
 
jthan24 profile image
Jhonny Pong •

Wooow, totally awesome! thanks for sharing!