Implementación de Python para el Protocolo A2A de Google: Guía Técnica Completa

Python A2A es una implementación del protocolo de comunicaciones de agente a agente (A2A) de Google, diseñado para estandarizar la interacción entre agentes de inteligencia artificial. Este protocolo aborda un desafío clave en el ecosistema de IA: permitir que diferentes servicios de IA se comuniquen de manera fluida sin necesidad de capas de conversión personalizadas.

El ecosistema de IA se ha especializado, con múltiples servicios que exponen sus propios formatos de API y parámetros. Los desarrolladores inviertne demasiado tiempo en construir infraestructura de comunicación. Python A2A resuelve este problema proporcionando una forma estandarizada para que los agentes se comuniquen entre sí, independientemente de su implementación subyacente.

Instalación y Configuración Inicial

# Instalación base del paquete
pip install python-a2a

# Para integración con OpenAI
pip install "python-a2a[openai]"

# Para integración con Anthropic Claude
pip install "python-a2a[anthropic]"

# Instalación completa con todas las dependencias opcionales
pip install "python-a2a[all]"

Principios Fundamentales del Protocolo

La implementación de Python A2A se basa en varios conceptos centrales:

  • Formato de Mensaje Estructurado: Define esquemas para contenido textual, invocaciones de funciones y respuestas.
  • Gestión de Hilos Conversacionales: Permite mantener el contexto a través de múltiples interacciones.
  • Mecanismo de Invocación de Funciones: Ofrece un método estandarizado para que los agentes expongan y ejecuten funciones.
  • Manejo de Errores Consistente: Establece un formato unificado para los mensajes de error.

Creación de un Agente de Ejemplo

Construyamos un agente simple que replica mensajes con un prefijo. Primero, el servidor del agente:

from python_a2a import A2AServer, Message, TextContent, MessageRole, run_server

class AgenteEco(A2AServer):
    """Agente que repite los mensajes recibidos con un prefijo."""

    def process_request(self, incoming_msg):
        if incoming_msg.content.type == "text":
            original_text = incoming_msg.content.text
            response_text = f"Reflejo: {original_text}"
            return Message(
                content=TextContent(text=response_text),
                role=MessageRole.AGENT,
                parent_message_id=incoming_msg.message_id,
                conversation_id=incoming_msg.conversation_id
            )

if __name__ == "__main__":
    servidor_agente = AgenteEco()
    run_server(servidor_agente, host="0.0.0.0", port=5000)

Posteriormente, implementamos un cliente para interactuar con nuestro agente:

from python_a2a import A2AClient, Message, TextContent, MessageRole

cliente_agente = A2AClient("http://localhost:5000/a2a")

mensaje_usuario = Message(
    content=TextContent(text="¿Puedes escucharme?"),
    role=MessageRole.USER
)
respuesta_del_agente = cliente_agente.send_message(mensaje_usuario)

print(f"El agente respondió: {respuesta_del_agente.content.text}")

Invocación de Funciones entre Agentes

Una característica potente de A2A es la invocación estandarizada de funciones. A continuación, un agente que ofrece operaciones matemáticas:

import math
from python_a2a import (
    A2AServer, Message, TextContent, FunctionCallContent,
    FunctionResponseContent, FunctionParameter, MessageRole, run_server
)

class ServicioCalculadora(A2AServer):
    """Agente que ofrece funciones de cálculo matemático."""

    def process_request(self, solicitud):
        if solicitud.content.type == "text":
            guia = (
                "Soy un servicio de cálculo. Funciones disponibles:\n"
                "- operar: Aritmética básica (operando1, operando2, operación)\n"
                "- raiz_cuadrada: Calcula la raíz cuadrada (numero)"
            )
            return Message(
                content=TextContent(text=guia),
                role=MessageRole.AGENT,
                parent_message_id=solicitud.message_id,
                conversation_id=solicitud.conversation_id
            )

        elif solicitud.content.type == "function_call":
            nombre_funcion = solicitud.content.name
            argumentos = {arg.name: arg.value for arg in solicitud.content.parameters}

            try:
                if nombre_funcion == "operar":
                    op = argumentos.get("operacion", "sumar")
                    num1 = float(argumentos.get("operando1", 0))
                    num2 = float(argumentos.get("operando2", 0))

                    if op == "sumar":
                        resultado = num1 + num2
                    elif op == "restar":
                        resultado = num1 - num2
                    elif op == "multiplicar":
                        resultado = num1 * num2
                    elif op == "dividir":
                        if num2 == 0:
                            raise ArithmeticError("División por cero no definida")
                        resultado = num1 / num2
                    else:
                        raise ValueError(f"Operación no reconocida: {op}")

                    return Message(
                        content=FunctionResponseContent(
                            name=nombre_funcion,
                            response={"resultado": resultado}
                        ),
                        role=MessageRole.AGENT,
                        parent_message_id=solicitud.message_id,
                        conversation_id=solicitud.conversation_id
                    )

                elif nombre_funcion == "raiz_cuadrada":
                    valor = float(argumentos.get("numero", 0))
                    if valor < 0:
                        raise ValueError("Valor debe ser no negativo")
                    raiz = math.sqrt(valor)
                    return Message(
                        content=FunctionResponseContent(
                            name=nombre_funcion,
                            response={"resultado": raiz}
                        ),
                        role=MessageRole.AGENT,
                        parent_message_id=solicitud.message_id,
                        conversation_id=solicitud.conversation_id
                    )

            except Exception as error:
                return Message(
                    content=FunctionResponseContent(
                        name=nombre_funcion,
                        response={"error": str(error)}
                    ),
                    role=MessageRole.AGENT,
                    parent_message_id=solicitud.message_id,
                    conversation_id=solicitud.conversation_id
                )

if __name__ == "__main__":
    mi_servicio = ServicioCalculadora()
    run_server(mi_servicio, host="0.0.0.0", port=5001)

Para invocar una función en el agente calculadora, el cliente genera una solicitud específica:

from python_a2a import (
    A2AClient, Message, FunctionCallContent,
    FunctionParameter, MessageRole
)

cliente_calculadora = A2AClient("http://localhost:5001/a2a")

solicitud_funcion = Message(
    content=FunctionCallContent(
        name="operar",
        parameters=[
            FunctionParameter(name="operacion", value="multiplicar"),
            FunctionParameter(name="operando1", value=7),
            FunctionParameter(name="operando2", value=6)
        ]
    ),
    role=MessageRole.USER
)

respuesta = cliente_calculadora.send_message(solicitud_funcion)

if respuesta.content.type == "function_response":
    valor = respuesta.content.response.get("resultado")
    if valor is not None:
        print(f"Resultado de la operación: {valor}")  # Salida: 42

Agentes Basados en Modelos de Lenguaje Grandes (LLM)

Python A2A incluye integraciones listas para usar con proveedores populares de LLM. Ejemplo con OpenAI:

import os
from python_a2a import OpenAIA2AServer, run_server

agente_openai = OpenAIA2AServer(
    api_key=os.environ.get("MI_CLAVE_OPENAI"),
    model="gpt-4",
    system_prompt="Eres un asistente de IA útil y conciso."
)

if __name__ == "__main__":
    run_server(agente_openai, host="0.0.0.0", port=5002)

De manera similar, se puede configurar un agente basado en Anthropic Claude:

import os
from python_a2a import ClaudeA2AServer, run_server

agente_claude = ClaudeA2AServer(
    api_key=os.environ.get("MI_CLAVE_ANTHROPIC"),
    model="claude-3-opus-20240229",
    system_prompt="Responde con precisión y claridad."
)

if __name__ == "__main__":
    run_server(agente_claude, host="0.0.0.0", port=5003)

Orquestación de Múltiples Agentes

El verdadero potencial de A2A radica en conectar varios agentes. Implementemos un flujo de trabajo para un asistente de investigación:

from python_a2a import (
    A2AClient, Message, TextContent, MessageRole, Conversation
)

def flujo_investigacion(pregunta):
    # Conexión a agentes especializados
    cliente_llm = A2AClient("http://localhost:5002/a2a")
    cliente_busqueda = A2AClient("http://localhost:5003/a2a")
    cliente_resumen = A2AClient("http://localhost:5004/a2a")

    # Conversación para rastrear el flujo completo
    dialogo = Conversation()
    dialogo.create_text_message(
        text=f"Tema de investigación: {pregunta}",
        role=MessageRole.USER
    )

    # Paso 1: Generación de consultas de búsqueda
    solicitud_consultas = Message(
        content=TextContent(
            text=f"Basado en el tema: '{pregunta}', "
                 f"genera 3 consultas de búsqueda específicas para encontrar información relevante."
        ),
        role=MessageRole.USER
    )
    respuesta_consultas = cliente_llm.send_message(solicitud_consultas)
    dialogo.add_message(respuesta_consultas)

    # Paso 2: Recopilación de información
    solicitud_busqueda = Message(
        content=TextContent(
            text=f"Busca información para responder: {pregunta}\n\n"
                 f"Utiliza estas consultas:\n{respuesta_consultas.content.text}"
        ),
        role=MessageRole.USER
    )
    resultados_busqueda = cliente_busqueda.send_message(solicitud_busqueda)
    dialogo.add_message(resultados_busqueda)

    # Paso 3: Síntesis de la información
    solicitud_sintesis = Message(
        content=TextContent(
            text=f"Sintetiza la siguiente información para responder: '{pregunta}'\n\n"
                 f"Datos recopilados:\n{resultados_busqueda.content.text}"
        ),
        role=MessageRole.USER
    )
    resumen_final = cliente_resumen.send_message(solicitud_sintesis)
    dialogo.add_message(resumen_final)

    # Respuesta final en la conversación
    dialogo.create_text_message(
        text=f"Respuesta a tu pregunta de investigación:\n\n{resumen_final.content.text}",
        role=MessageRole.AGENT
    )

    return dialogo

if __name__ == "__main__":
    tema = input("¿Cuál es tu pregunta de investigación? ")
    resultado_dialogo = flujo_investigacion(tema)
    print("\nInvestigación completada.")
    print("=" * 50)
    print(resultado_dialogo.messages[-1].content.text)

Integración de Servicios Heterogéneos

Este ejemplo ilustra cómo A2A simplifica la comunicación entre agentes que proveen servicios diferentes, como el clima y la planificación:

from python_a2a import A2AClient, Message, TextContent, MessageRole

def organizar_viaje(destino):
    cliente_clima = A2AClient("http://localhost:5001/a2a")
    cliente_planificador = A2AClient("http://localhost:5002/a2a")

    # Consulta al agente del clima
    msg_clima = Message(
        content=TextContent(text=f"¿Cuál es el pronóstico del tiempo para {destino}?"),
        role=MessageRole.USER
    )
    pronostico = cliente_clima.send_message(msg_clima)

    # Solicitud al agente planificador con datos del clima
    msg_plan = Message(
        content=TextContent(
            text=f"Estoy planificando un viaje a {destino}. Pronóstico: {pronostico.content.text}"
                 f" Sugiéreme actividades adecuadas."
        ),
        role=MessageRole.USER
    )
    sugerencias = cliente_planificador.send_message(msg_plan)

    return sugerencias.content.text

Gestión de Conversaciones Multivuelta

La clase Conversation permite administrar interacciones prolongadas, manteniendo el historial de mensajes:

from python_a2a import Conversation, MessageRole, A2AClient

conversacion = Conversation()

conversacion.create_text_message(
    text="¿Hace buen tiempo hoy?",
    role=MessageRole.USER
)

agente_meteo = A2AClient("http://localhost:5001/a2a")

ultimo_mensaje = conversacion.messages[-1]
respuesta_inicial = agente_meteo.send_message(ultimo_mensaje)
conversacion.add_message(respuesta_inicial)

conversacion.create_text_message(
    text="Gracias. ¿Necesitaré un paraguas?",
    role=MessageRole.USER
)

respuesta_secundaria = agente_meteo.send_message(
    conversacion.messages[-1],
    conversation_id=conversacion.conversation_id
)
conversacion.add_message(respuesta_secundaria)

for mensaje in conversacion.messages:
    identificador = "Humano" if mensaje.role == MessageRole.USER else "Agente"
    print(f"{identificador}: {mensaje.content.text}")

Manejo de Errores Estandarizado

La biblioteca proporciona mecanismos para capturar errores comunes en la comunicación:

from python_a2a import A2AClient, Message, TextContent, MessageRole

cliente = A2AClient("http://localhost:5001/a2a")

try:
    paquete = Message(
        content=TextContent(text="Saludo de prueba"),
        role=MessageRole.USER
    )
    respuesta = cliente.send_message(paquete)
except ConnectionError as ce:
    print(f"Error de conexión con el agente: {ce}")
except TimeoutError as te:
    print(f"Tiempo de espera excedido: {te}")
except Exception as ex:
    print(f"Error inesperado durante la comunicación: {ex}")

Plantilla para Agentes Personalizados

Para crear un agente compatible con A2A, se puede seguir esta estructura base:

from python_a2a import A2AServer, Message, TextContent, MessageRole, run_server

class MiAgentePersonalizado(A2AServer):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)
        # Inicialización de componentes específicos del agente

    def process_request(self, solicitud):
        """Punto de entrada principal para el procesamiento de mensajes."""
        if solicitud.content.type == "text":
            contenido_recibido = solicitud.content.text
            respuesta_generada = self.mi_logica_especifica(contenido_recibido)

            return Message(
                content=TextContent(text=respuesta_generada),
                role=MessageRole.AGENT,
                parent_message_id=solicitud.message_id,
                conversation_id=solicitud.conversation_id
            )

        elif solicitud.content.type == "function_call":
            # Lógica para manejar invocaciones de funciones
            pass

        return Message(
            content=TextContent(text="Tipo de mensaje no soportado por este agente."),
            role=MessageRole.AGENT,
            parent_message_id=solicitud.message_id,
            conversation_id=solicitud.conversation_id
        )

    def mi_logica_especifica(self, texto):
        # Implementación de la lógica de negocio única
        return f"Procesado: {texto}"

if __name__ == "__main__":
    agente = MiAgentePersonalizado()
    run_server(agente, host="0.0.0.0", port=5005)

Etiquetas: Python agent-to-agent protocol A2A google-ai

Publicado el 7-23 09:18

Etiquetas populares

©2026 Friki Work