Configuración del entorno

Herramientas requeridas

• Editor de código como Visual Studio Code
• Terminal (Linux/Mac) o PowerShell (Windows)

Entorno Python

Utilizaremos uv como gestor de paquetes. La biblioteca A2A requiere Python ≥3.12:

echo 'import sys; print(sys.version)' | uv run -

Salida esperada:

3.12.3 (main, Feb 4 2025, 14:48:35) [GCC 13.3.0]

Creación del proeycto

uv init --package mi-proyecto
cd mi-proyecto
uv venv .venv
source .venv/bin/activate

Estructura inicial:

.
├── pyproject.toml
├── README.md
├── src
│   └── mi-proyecto
│       ├── __init__.py

Instalación de dependencias

uv add git+https://github.com/google/A2A#subdirectory=samples/python
touch src/mi-proyecto/agente.py
touch src/mi-proyecto/gestor_tareas.py

Habilidades del agente

Definición de una habilidad de eco:

import google_a2a
from google_a2a.common.types import HabilidadAgente

def principal():
    habilidad = HabilidadAgente(
        id="mi-proyecto-eco",
        nombre="Herramienta Eco",
        descripcion="Repite la entrada recibida",
        etiquetas=["eco", "repetidor"],
        ejemplos=["Este texto será repetido"],
        modosEntrada=["texto"],
        modosSalida=["texto"],
    )
    print(habilidad)

Tarjeta del agente

import logging
import click
from google_a2a.common.types import CapacidadesAgente, TarjetaAgente

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@click.command()
@click.option("--host", default="localhost")
@click.option("--port", default=10002)
def principal(host, port):
    habilidad = HabilidadAgente(...)
    capacidades = CapacidadesAgente()
    tarjeta = TarjetaAgente(
        nombre="Agente Eco",
        descripcion="Repite la entrada proporcionada",
        url=f"http://{host}:{port}/",
        version="0.1.0",
        capacidades=capacidades,
        habilidades=[habilidad]
    )

Servidor A2A

Implementación del gestor de tareas:

from typing import AsyncIterable
import google_a2a
from google_a2a.common.server.gestor_tareas import GestorTareasMemoria

class GestorTareasPersonalizado(GestorTareasMemoria):
    def __init__(self):
        super().__init__()

    async def manejar_tarea(self, solicitud):
        texto = solicitud.params.mensaje.partes[0].texto
        await self.actualizar_tarea(
            id_tarea=solicitud.params.id,
            estado="COMPLETADO",
            respuesta=f"Recibido: {texto}"
        )

Inicialización del servidor:

from google_a2a.common.server import ServidorA2A
from mi-proyecto.gestor_tareas import GestorTareasPersonalizado

def principal(host, port):
    gestor = GestorTareasPersonalizado()
    servidor = ServidorA2A(
        tarjeta_agente=tarjeta,
        gestor_tareas=gestor,
        host=host,
        port=port
    )
    servidor.iniciar()

Interacción con el servidor

uv run google-a2a-cli --agent http://localhost:10002

Respuesta esperada:

"mensaje": {"rol": "agente", "partes": [{"tipo": "texto", "texto": "Recibido: Hola!"}]}

Transmisión en tiempo real

class GestorTareasPersonalizado(GestorTareasMemoria):
    async def manejar_suscripcion(self, solicitud):
        async def _transmitir():
            mensajes = ["uno", "dos", "tres"]
            for texto in mensajes:
                await self.enviar_evento(
                    id_tarea=solicitud.params.id,
                    contenido=f"{texto}",
                    final=(texto == mensajes[-1])
                )
        asyncio.create_task(_transmitir())

Integración con Ollama

from langchain_ollama import ChatOllama

def crear_agente_ollama(url_base, modelo):
    return ChatOllama(
        base_url=url_base,
        model=modelo,
        temperature=0.2
    )

async def ejecutar_ollama(agente, entrada):
    respuesta = await agente.ainvoke({"mensajes": entrada})
    return str(respuesta.mensajes[-1].contenido)

class GestorTareasPersonalizado(GestorTareasMemoria):
    def __init__(self, url_ollama, modelo_ollama):
        self.agente_ia = crear_agente_ollama(url_ollama, modelo_ollama)

    async def manejar_tarea(self, solicitud):
        texto = solicitud.params.mensaje.partes[0].texto
        respuesta = await ejecutar_ollama(self.agente_ia, texto)
        await self.actualizar_tarea(..., respuesta)

Inicio del servidor con Ollama:

uv run mi-proyecto --ollama-host http://127.0.0.1:11434 --ollama-model modelo-ollama