Protocolo AG-UI: Bus Reactivo de Interacción Agente-Usuario

1. La Tríada de Protocolos en la Era Agéntica

En el diseño de sistemas basados en agentes autónomos de inteligencia artificial, la arquitectura de comunicaciones ha evolucionado vertiginosamente. Durante los primeros compases de los modelos generativos, los desarrolladores se concentraron casi exclusivamente en la inferencia del modelo mediante simples endpoints HTTP sincrónicos. Sin embargo, un agente de producción moderno no es un simple oráculo que responde preguntas en 500 milisegundos: es un sistema distribuido reactivo, reflexivo y de ejecución prolongada.

Hoy en día, el panorama de estandarización agéntica se asienta sólidamente sobre tres protocolos complementarios que delimitan con nitidez sus fronteras de responsabilidad:

Ver en alta resolución (SVG vectorial)

Especificación Topológica en Texto (Esquema RFC / ASCII para Agentes & Docs)

┌────────────────────────────────────────────────────────────────────────┐
│            1. CAPA DE USUARIO (FRONTEND CLIENTS / @ag-ui/client)       │
│   Web App (React/Vue)  •  Mobile App  •  IDE Sidecar  •  Generative UI │
└─────────────────┬────────────────────────────────────▲─────────────────┘
                  │ A. HTTP POST (RunAgentInput)       │ B. Stream SSE
                  │ threadId, runId, state,            │ (BaseEvent JSON)
                  │ messages, tools, resume[]          │ text/event-stream
                  ▼                                    │
┌────────────────────────────────────────────────────────────────────────┐
│         2. BUS REACTIVO PROTOCOLO AG-UI (7 Familias de Eventos)        │
│  • Streaming Cognitivo: REASONING_* | TEXT_MESSAGE_* | TOOL_CALL_*     │
│  • Estado Reactivo (RFC 6902 JSON Patch): STATE_SNAPSHOT | STATE_DELTA │
│  • Stateless HITL: RUN_FINISHED (outcome="interrupt") ➔ ResumeEntry[]  │
└───────────────────────────────────┬────────────────────────────────────┘
                                    ▼
┌────────────────────────────────────────────────────────────────────────┐
│            3. RUNTIME DEL AGENTE / ORQUESTADOR BACKEND                 │
│   LangGraph • Pydantic AI • Microsoft Agent Framework • CrewAI         │
│   [Southbound: MCP]  •  [East-West: A2A]  •  [Runtime: OpenSandbox]    │
└────────────────────────────────────────────────────────────────────────┘

2. El Problema: El Colapso de REST y GraphQL ante Agentes Autónomos

Durante décadas, el desarrollo web se construyó sobre el paradigma Request/Response: el cliente envía una petición HTTP, el backend realiza una consulta a la base de datos o ejecuta una lógica sincrónica y devuelve un objeto JSON en unos cientos de milisegundos. Incluso las suscripciones de GraphQL o los WebSockets convencionales suelen asumir el transporte de datos transaccionales cerrados.

Sin embargo, los agentes de IA rompen por completo las asunciones fundacionales de estos protocolos tradicionales:

• Procesos asíncronos y no lineales de larga duración: Un agente que investiga una base de código o consulta varios sensores IoT puede trabajar durante 30 segundos, 5 minutos o varias horas antes de concluir su objetivo.
• Múltiples fases cognitivas simultáneas: Mientras trabaja, el modelo no solo emite texto. Primero piensa internamente (Chain-of-Thought o razonamiento), luego decide invocar una herramienta, evalúa el resultado intermedio, delega a un subagente y recién entonces produce texto visible.
• Mutación progresiva de estado (Generative UI): En lugar de esperar a que termine toda la tarea para mostrar una pantalla completa, el agente necesita actualizar widgets individuales, tablas de datos o gráficos dinámicamente mientras razona.
• Intervención Humana (Human-in-the-Loop): Operaciones críticas (como transferir fondos bancarios, borrar un contenedor Docker o publicar una orden de compra) exigen aprobación explícita del operador humano.

Antes de AG-UI, cada framework (LangGraph, CopilotKit, AutoGen, CrewAI o motores caseros) inventaba su propia estructura de mensajes sobre WebSockets o SSE. El resultado era la clásica pesadilla de fragmentación N × M: un componente de frontend diseñado para LangGraph no podía reutilizarse con un agente escrito en Pydantic AI o Microsoft Semantic Kernel sin reescribir por completo la capa de red y el gestor de eventos.

AG-UI (Agent-User Interaction Protocol) fue concebido (inicialmente impulsado por el equipo de CopilotKit y adoptado ampliamente por la comunidad y grandes fabricantes) para convertirse en el estándar abierto unificado que desacopla definitivamente el runtime del agente de la interfaz visual del usuario.

3. Arquitectura y Topología del Protocolo AG-UI

El protocolo AG-UI adopta una topología desacoplada Productor-Consumidor sobre la infraestructura estándar de la web:

• Transporte de Red: Aunque soporta WebSockets y Webhooks, el transporte nativo y más utilizado es HTTP POST con Server-Sent Events (SSE) (encabezado Content-Type: text/event-stream). Esto garantiza compatibilidad inmediata con proxies empresariales, balanceadores de carga, redes CDN y Edge workers sin la complejidad de mantener conexiones TCP duplex permanentes.
• Direccionalidad del Turno: Cada ejecución (denominada Run) es estrictamente unidireccional durante su ciclo de vida activo: el cliente inicia la ejecución enviando el contrato de entrada, y el agente emite una secuencia ininterrumpida de eventos tipados JSON hasta emitir un evento terminal.

// Contrato TypeScript: RunAgentInput (@ag-ui/core)
type RunAgentInput = {
  threadId: string;                 // Identificador único de la conversación / hilo
  runId: string;                    // Identificador unívoco de la ejecución actual
  parentRunId?: string;             // Opcional: ID de la ejecución padre si es subagente
  state?: Record<string, any>;      // Estado global estructurado compartido UI-Agente
  messages: Message[];              // Historial cronológico de mensajes (User, Assistant, Tool)
  tools: Tool[];                    // Herramientas declaradas en el CLIENTE para ejecutar en la UI
  context: Context[];               // Metadatos contextuales (usuario activo, URL actual, tema)
  forwardedProps?: any;             // Propiedades arbitrarias transferidas al framework
  resume?: ResumeEntry[];           // Respuestas de Human-in-the-Loop para reanudar un interrupt
};

Este contrato estandariza todas las variables esenciales que cualquier runtime agéntico necesita para iniciar su ciclo de pensamiento, permitiendo que backends implementados en Python, TypeScript, Go o .NET compartan exactamente la misma interfaz externa.

4. Taxonomía de Eventos: El Sistema de Streaming Reactivo

Cada fragmento transmitido por el agente a través del canal SSE (data: { ... }\n\n) hereda de la estructura universal BaseEvent:

interface BaseEvent {
  type: string;                    // Discriminador (SCREAMING_SNAKE_CASE en el cable)
  timestamp?: number;               // Epoch en milisegundos de la emisión
  rawEvent?: any;                   // Payload nativo original del framework subyacente
  metadata?: Record<string, any>;    // Metadatos abiertos: tokens consumidos, trace ID, latencias
}

La especificación AG-UI 1.0 organiza sus eventos en siete familias funcionales claramente diferenciadas:

5. Human-in-the-Loop: El Patrón Interrupt-and-Resume sin Estado

Uno de los mayores desafíos de ingeniería al integrar agentes en producción es el patrón Human-in-the-Loop (HITL). Supongamos que un agente bancario decide emitir una transferencia de 10.000 € y necesita la confirmación explícita del usuario con su firma digital o botón de aprobación.

En arquitecturas ingenuas basadas en WebSockets o sockets HTTP suspendidos (long-polling), el servidor mantiene el hilo y la conexión TCP abierta esperando que el usuario haga clic. Si el usuario tarda 20 minutos en responder, pierde cobertura Wi-Fi o apaga su portátil para almorzar, la conexión expira con un timeout 504, los recursos de memoria se fugan y el estado del agente queda corrupto o huérfano.

AG-UI resuelve este problema implementando un ciclo Interrupt-and-Resume completamente sin estado (stateless):

// 1. Evento de cierre emitido por el backend del agente (SSE)
{
  "type": "RUN_FINISHED",
  "threadId": "thread_quijotech_982",
  "runId": "run_initial_001",
  "outcome": {
    "type": "interrupt",
    "interrupts": [
      {
        "id": "intr_transfer_approval",
        "reason": "payment_authorization_required",
        "message": "¿Deseas autorizar la transferencia de 10.000 € a la cuenta ES91-2100...?",
        "toolCallId": "call_wire_transfer_77",
        "responseSchema": {
          "type": "object",
          "properties": {
            "authorized": { "type": "boolean" },
            "securityCode": { "type": "string" }
          },
          "required": ["authorized"]
        },
        "expiresAt": "2026-10-23T14:00:00Z"
      }
    ]
  }
}

Una vez recibido este evento, la conexión HTTP finaliza limpiamente. El backend libera la memoria del hilo de inferencia persistiendo el estado en su almacén distribuido (Redis, Postgres o SQLite). La interfaz de usuario interpreta el objeto interrupts y despliega un cuadro de diálogo nativo o modal de autorización.

// 2. Petición POST del cliente para reanudar la ejecución (RunAgentInput)
{
  "threadId": "thread_quijotech_982",
  "runId": "run_resume_002",
  "messages": [],
  "resume": [
    {
      "interruptId": "intr_transfer_approval",
      "status": "resolved",
      "payload": {
        "authorized": true,
        "securityCode": "948201"
      },
      "metadata": {
        "userIp": "192.168.1.50",
        "approvedBy": "dr_villanueva"
      }
    }
  ]
}

El runtime del agente recibe esta entrada, recupera el estado serializado del hilo, valida que el payload satisface el responseSchema, inyecta la aprobación en el punto exacto donde se detuvo el grafo de ejecución y prosigue su tarea streaming en una nueva sesión SSE. Este mecanismo es resiliente a desconexiones, escalable horizontalmente en clústeres Kubernetes y completamente auditable.

6. Generative UI y Herramientas del Lado del Cliente (Frontend Tools)

Una distinción arquitectónica crucial en AG-UI es la coexistencia de dos clases de herramientas:

• Server-Side Tools (Herramientas de Servidor): Ejecutadas por el backend del agente (habitualmente expuestas vía servidores MCP locales o remotos, como consultas a bases de datos o comandos de terminal). El backend ejecuta la acción y emite TOOL_CALL_RESULT directamente en la secuencia SSE.
• Client-Side Tools (Herramientas de Cliente): Declaradas por la interfaz frontend en el arreglo tools de RunAgentInput. Representan capacidades nativas del navegador o del dispositivo del usuario (por ejemplo: abrir una pestaña, renderizar un gráfico interactivo con Chart.js, pedir geolocalización GPS o cambiar el zoom de un mapa).

El flujo de ejecución de una herramienta de cliente sigue una regla inviolable de la especificación AG-UI: el backend del agente NUNCA debe fabricar ni responder automáticamente la llamada a una herramienta de cliente. En su lugar:

1. El agente detecta la necesidad de invocar una herramienta frontend y emite los eventos TOOL_CALL_START, TOOL_CALL_ARGS y TOOL_CALL_END en la secuencia SSE.
2. El agente cierra el paso actual dejando la herramienta pendiente de resolución.
3. El frontend intercepta la llamada, ejecuta la lógica en el DOM o el sistema del cliente (ej. renderizando un componente reactivo).
4. El frontend remite el resultado resultante al agente en el siguiente ciclo o mediante submitToolResult, garantizando que la IA razone basándose en la retroalimentación real del entorno visual.

7. Implementación Práctica: Servidor FastAPI y Cliente React

Veamos una implementación técnica representativa construida con componentes estándar de la industria.

A. Servidor Backend en Python (FastAPI + Pydantic + Streaming SSE)

A continuación se ilustra un microservicio que expone el endpoint AG-UI /api/agent/run emitiendo eventos SSE con tokens, herramientas y deltas de estado JSON Patch:

# backend_agent_agui.py
import asyncio
import json
import time
from typing import Any, Dict, List, Optional
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field

app = FastAPI(title="Quijotech AG-UI Compliant Agent")

class Message(BaseModel):
    role: str
    content: str

class Tool(BaseModel):
    name: str
    description: str
    parameters: Dict[str, Any]

class ResumeEntry(BaseModel):
    interruptId: str
    status: str
    payload: Optional[Dict[str, Any]] = None

class RunAgentInput(BaseModel):
    threadId: str
    runId: str
    parentRunId: Optional[str] = None
    state: Optional[Dict[str, Any]] = Field(default_factory=dict)
    messages: List[Message]
    tools: Optional[List[Tool]] = Field(default_factory=list)
    resume: Optional[List[ResumeEntry]] = None

def format_sse(event_data: dict) -> str:
    """Serializa un evento AG-UI al formato de cable Server-Sent Events."""
    return f"data: {json.dumps(event_data, ensure_ascii=False)}\n\n"

async def agui_event_generator(payload: RunAgentInput):
    # 1. Notificar inicio de ejecución
    yield format_sse({
        "type": "RUN_STARTED",
        "timestamp": int(time.time() * 1000),
        "threadId": payload.threadId,
        "runId": payload.runId
    })
    await asyncio.sleep(0.1)

    # 2. Emitir fase de pensamiento interno (Reasoning)
    yield format_sse({"type": "REASONING_START"})
    thoughts = ["Analizando telemetría de sensores IoT...", "Detectada anomalía térmica en nodo #4."]
    for thought in thoughts:
        yield format_sse({
            "type": "REASONING_MESSAGE_CONTENT",
            "delta": thought + " "
        })
        await asyncio.sleep(0.15)
    yield format_sse({"type": "REASONING_END"})

    # 3. Invocar una herramienta del backend
    tool_call_id = "call_iot_reset_01"
    yield format_sse({
        "type": "TOOL_CALL_START",
        "toolCallId": tool_call_id,
        "toolCallName": "reiniciar_nodo_sensor"
    })
    yield format_sse({
        "type": "TOOL_CALL_ARGS",
        "toolCallId": tool_call_id,
        "delta": json.dumps({"nodo_id": 4, "modo": "soft_reset"})
    })
    yield format_sse({"type": "TOOL_CALL_END", "toolCallId": tool_call_id})

    # Simular ejecución de la herramienta
    await asyncio.sleep(0.2)
    yield format_sse({
        "type": "TOOL_CALL_RESULT",
        "toolCallId": tool_call_id,
        "result": json.dumps({"status": "SUCCESS", "nodo_id": 4, "temperatura_actual": "24.2C"})
    })

    # 4. Sincronizar mutación de estado con JSON Patch (RFC 6902)
    yield format_sse({
        "type": "STATE_DELTA",
        "patch": [
            {"op": "replace", "path": "/nodos/4/estado", "value": "operativo"},
            {"op": "add", "path": "/alertas_activas", "value": []}
        ]
    })

    # 5. Emitir mensaje textual final en streaming
    msg_id = "msg_resp_101"
    yield format_sse({
        "type": "TEXT_MESSAGE_START",
        "messageId": msg_id,
        "role": "assistant"
    })
    response_tokens = ["El ", "nodo ", "sensor #4 ", "ha ", "sido ", "reiniciado ", "con ", "éxito."]
    for token in response_tokens:
        yield format_sse({
            "type": "TEXT_MESSAGE_CONTENT",
            "messageId": msg_id,
            "delta": token
        })
        await asyncio.sleep(0.08)
    yield format_sse({"type": "TEXT_MESSAGE_END", "messageId": msg_id})

    # 6. Finalización exitosa del run
    yield format_sse({
        "type": "RUN_FINISHED",
        "threadId": payload.threadId,
        "runId": payload.runId,
        "outcome": {"type": "success"}
    })

@app.post("/api/agent/run")
async def run_agent_endpoint(payload: RunAgentInput):
    return StreamingResponse(
        agui_event_generator(payload),
        media_type="text/event-stream"
    )

B. Cliente Frontend en TypeScript / React

En el lado del cliente, el consumo de AG-UI puede realizarse directamente a través de un lector nativo de EventSource / fetch con ReadableStream o empleando el SDK oficial @ag-ui/client. El siguiente código ejemplifica el procesamiento reactivo de eventos:

// agui_client_consumer.ts
import { applyPatch } from 'fast-json-patch'; // RFC 6902

export async function executeAgentRun(promptText: string, currentAppState: any) {
  const payload = {
    threadId: "session-crm-55",
    runId: "run-" + crypto.randomUUID(),
    state: currentAppState,
    messages: [{ role: "user", content: promptText }],
    tools: []
  };

  const response = await fetch("/api/agent/run", {
    method: "POST",
    headers: { "Content-Type": "application/json" },
    body: JSON.stringify(payload)
  });

  const reader = response.body?.getReader();
  const decoder = new TextDecoder("utf-8");
  let buffer = "";

  while (reader) {
    const { done, value } = await reader.read();
    if (done) break;

    buffer += decoder.decode(value, { stream: true });
    const lines = buffer.split("\n\n");
    buffer = lines.pop() || ""; // Mantener residuo si el chunk está cortado

    for (const block of lines) {
      if (block.startsWith("data: ")) {
        const rawJson = block.replace("data: ", "").trim();
        if (!rawJson) continue;
        const event = JSON.parse(rawJson);

        switch (event.type) {
          case "RUN_STARTED":
            console.log(`[AG-UI] Ejecución iniciada: ${event.runId}`);
            break;

          case "REASONING_MESSAGE_CONTENT":
            // Renderizar pensamiento en caja plegable
            document.getElementById("thought-bubble")!.innerText += event.delta;
            break;

          case "TEXT_MESSAGE_CONTENT":
            // Renderizar respuesta typewriter en la interfaz
            document.getElementById("agent-response")!.innerText += event.delta;
            break;

          case "STATE_DELTA":
            // Sincronizar estado global mediante parches JSON Patch
            console.log("[AG-UI] Aplicando parches reactivos:", event.patch);
            applyPatch(currentAppState, event.patch);
            actualizarInterfazReactiva(currentAppState);
            break;

          case "RUN_FINISHED":
            if (event.outcome?.type === "interrupt") {
              mostrarModalAprobacionHumana(event.outcome.interrupts[0]);
            } else {
              console.log("[AG-UI] Tarea concluida con éxito.");
            }
            break;
        }
      }
    }
  }
}

8. Matriz Comparativa: MCP vs. A2A vs. AG-UI

Para arquitectos de software e investigadores en sistemas agénticos, comprender la especialización de cada protocolo previene antipatrones de diseño y superposiciones innecesarias: