Introducción a AG-UI

Prerequisites

• .NET 8 o posterior
• Un proyecto de ASP.NET Core
• Un MAF configurado AIAgent

En el ejemplo se usa Azure OpenAI, pero MapAGUIServer funciona con cualquier agente de MAF.

Creación de un servidor AG-UI

Instale el paquete de hospedaje:

dotnet add package Microsoft.Agents.AI.Hosting.AGUI.AspNetCore --prerelease

Registrar el alojamiento de AG-UI y asignar su agente:

using Microsoft.Agents.AI;
using Microsoft.Agents.AI.Hosting.AGUI.AspNetCore;

WebApplicationBuilder builder = WebApplication.CreateBuilder(args);
builder.Services.AddAGUIServer();

AIAgent agent = CreateAgent();

WebApplication app = builder.Build();
app.MapAGUIServer("/", agent);
await app.RunAsync();

MapAGUIServer acepta solicitudes AG-UI RunAgentInput y transmite en flujo la respuesta del agente como eventos AG-UI mediante eventos enviados por el servidor (SSE).

Ejecute el servidor en la dirección URL usada por el ejemplo de cliente:

dotnet run --urls http://localhost:8888

Conexión con un cliente de .NET

El SDK de AG-UI para .NET incluye AGUIChatClient, que implementa IChatClient y puede adaptarse a un agente MAF:

dotnet add package AGUI.Client --prerelease
dotnet add package Microsoft.Agents.AI --prerelease

using AGUI.Abstractions;
using AGUI.Client;
using Microsoft.Agents.AI;
using Microsoft.Extensions.AI;

using HttpClient httpClient = new() { BaseAddress = new Uri("http://localhost:8888") };
AGUIChatClient chatClient = new(new AGUIChatClientOptions(httpClient, "/"));
AIAgent remoteAgent = chatClient.AsAIAgent();
AgentSession session = await remoteAgent.CreateSessionAsync();

List<AgentResponseUpdate> firstTurnUpdates = [];
await foreach (AgentResponseUpdate update in
    remoteAgent.RunStreamingAsync("Hello", session))
{
    firstTurnUpdates.Add(update);

    foreach (TextContent text in update.Contents.OfType<TextContent>())
    {
        Console.Write(text.Text);
    }
}

También puede conectarse con cualquier cliente que implemente el protocolo AG-UI.

Continuidad de la conversación

AG-UI usa threadId y parentRunId para identificar las solicitudes de continuación. Estos identificadores son datos de protocolo, no credenciales de autorización.

AGUIChatClient no tiene estado. Para continuar una conversación controlada por el servidor, obtenga los identificadores del RunStartedEvent del primer turno y, a continuación, incluya el mismo threadId y el runId anterior como parentRunId en la siguiente solicitud:

RunStartedEvent started = firstTurnUpdates
    .Select(update => update.AsChatResponseUpdate().RawRepresentation)
    .OfType<RunStartedEvent>()
    .FirstOrDefault()
    ?? throw new InvalidOperationException("The server didn't return a run-started event.");

ChatMessage nextMessage = new(ChatRole.User, "What did I just say?");
ChatClientAgentRunOptions continuationOptions = new()
{
    ChatOptions = new ChatOptions
    {
        RawRepresentationFactory = _ => new RunAgentInput
        {
            ThreadId = started.ThreadId,
            ParentRunId = started.RunId,
            Messages = new[] { nextMessage }.AsAGUIMessages().ToList(),
        },
    },
};

await foreach (AgentResponseUpdate update in
    remoteAgent.RunStreamingAsync([nextMessage], session, continuationOptions))
{
    // Process the continued response.
}

Envíe solo los nuevos mensajes en una solicitud de continuación. MapAGUIServer usa threadId para seleccionar la sesión del agente hospedado e parentRunId identificar la ejecución que se está realizando. Sin persistencia de sesión hospedada, cada solicitud recibe una nueva sesión de servidor; En su lugar, el cliente puede volver a enviar el historial de conversaciones.

Para conservar el estado del servidor AgentSession de una solicitud a otra, configure la persistencia y el aislamiento de la sesión alojada y, a continuación, asigne el agente alojado especificado por nombre con MapAGUIServer. Para conocer el límite de confianza específico de AG-UI, consulte Consideraciones de producción y seguridad.

Pasos siguientes

Recursos relacionados

• Información general deAG-UI
• Hospedaje de MAF
• Documentación del protocoloAG-UI

Prerequisites

Antes de comenzar, asegúrese de que dispone de lo siguiente:

• Python 3.10 o posterior
• Configuración del punto final del servicio Azure OpenAI y su implementación
• CLI de Azure instalada y autenticada
• El usuario tiene el Cognitive Services OpenAI Contributor rol para el recurso de Azure OpenAI.

Paso 1: Crear un servidor de AG-UI

El servidor AG-UI hospeda el agente de IA y lo expone a través de puntos de conexión HTTP mediante FastAPI.

Instalar paquetes necesarios

Instale los paquetes necesarios para el servidor:

pip install agent-framework-ag-ui --pre

O bien, usando uv:

uv pip install agent-framework-ag-ui --prerelease=allow

Esto instalará automáticamente agent-framework-core, fastapi, uvicorn y sse-starlette como dependencias.

Código de servidor

Cree un archivo denominado server.py:

"""AG-UI server example."""

import os

from agent_framework import Agent
from agent_framework.openai import OpenAIChatCompletionClient
from agent_framework_ag_ui import add_agent_framework_fastapi_endpoint
from azure.identity import AzureCliCredential
from fastapi import FastAPI

# Read required configuration
endpoint = os.environ.get("AZURE_OPENAI_ENDPOINT")
deployment_name = os.environ.get("AZURE_OPENAI_CHAT_COMPLETION_MODEL")

if not endpoint:
    raise ValueError("AZURE_OPENAI_ENDPOINT environment variable is required")
if not deployment_name:
    raise ValueError("AZURE_OPENAI_CHAT_COMPLETION_MODEL environment variable is required")

chat_client = OpenAIChatCompletionClient(
    model=deployment_name,
    azure_endpoint=endpoint,
    api_version=os.getenv("AZURE_OPENAI_API_VERSION"),
    credential=AzureCliCredential(),
)

# Create the AI agent
agent = Agent(
    name="AGUIAssistant",
    instructions="You are a helpful assistant.",
    client=chat_client,
)

# Create FastAPI app
app = FastAPI(title="AG-UI Server")

# Register the AG-UI endpoint
add_agent_framework_fastapi_endpoint(app, agent, "/")

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="127.0.0.1", port=8888)

Conceptos clave

• add_agent_framework_fastapi_endpoint: registra el punto de conexión de AG-UI con el control automático de solicitudes y respuestas y el streaming de SSE.
• Agent: agente de Agent Framework que controlará las solicitudes entrantes.
• Integración con FastAPI: usa la compatibilidad asincrónica nativa de FastAPI con las respuestas de streaming.
• Instrucciones: el agente se crea con instrucciones predeterminadas, que los mensajes de cliente pueden invalidar.
• Configuración: OpenAIChatCompletionClient acepta entradas explícitas de enrutamiento de Azure, como model, azure_endpoint, api_versiony credential, y también puede leer desde variables de entorno.

Configurar y ejecutar el servidor

Establezca las variables de entorno necesarias:

export AZURE_OPENAI_ENDPOINT="https://your-resource.openai.azure.com/"
export AZURE_OPENAI_CHAT_COMPLETION_MODEL="gpt-4o-mini"

Ejecute el servidor:

python server.py

O bien mediante uvicorn directamente:

uvicorn server:app --host 127.0.0.1 --port 8888

El servidor comenzará a escuchar en http://127.0.0.1:8888.

Paso 2: Crear un cliente de AG-UI

El cliente AG-UI se conecta al servidor remoto y muestra las respuestas de streaming.

Instalar paquetes necesarios

El paquete AG-UI ya está instalado, que incluye :AGUIChatClient

# Already installed with agent-framework-ag-ui
pip install agent-framework-ag-ui --pre

Código de cliente

Cree un archivo denominado client.py:

"""AG-UI client example."""

import asyncio
import os

from agent_framework import Agent
from agent_framework_ag_ui import AGUIChatClient


async def main():
    """Main client loop."""
    # Get server URL from environment or use default
    server_url = os.environ.get("AGUI_SERVER_URL", "http://127.0.0.1:8888/")
    print(f"Connecting to AG-UI server at: {server_url}\n")

    # Create AG-UI chat client
    chat_client = AGUIChatClient(endpoint=server_url)

    # Create agent with the chat client
    agent = Agent(
        name="ClientAgent",
        client=chat_client,
        instructions="You are a helpful assistant.",
    )

    # Get a thread for conversation continuity
    thread = agent.create_session()

    try:
        while True:
            # Get user input
            message = input("\nUser (:q or quit to exit): ")
            if not message.strip():
                print("Request cannot be empty.")
                continue

            if message.lower() in (":q", "quit"):
                break

            # Stream the agent response
            print("\nAssistant: ", end="", flush=True)
            async for update in agent.run(message, session=thread, stream=True):
                # Print text content as it streams
                if update.text:
                    print(f"\033[96m{update.text}\033[0m", end="", flush=True)

            print("\n")

    except KeyboardInterrupt:
        print("\n\nExiting...")
    except Exception as e:
        print(f"\n\033[91mAn error occurred: {e}\033[0m")


if __name__ == "__main__":
    asyncio.run(main())

Conceptos clave

• Eventos Enviados por el Servidor (SSE): el protocolo utiliza el formato SSE (data: {json}\n\n)
• Tipos de eventos: los distintos eventos proporcionan metadatos y contenido (MAYÚSCULAS con caracteres de subrayado):
  • RUN_STARTED: el agente ha iniciado el procesamiento
  • TEXT_MESSAGE_START: inicio de un mensaje de texto del agente
  • TEXT_MESSAGE_CONTENT: texto incremental transmitido desde el agente (con el campo delta)
  • TEXT_MESSAGE_END: final de un mensaje de texto
  • RUN_FINISHED: finalización correcta
  • RUN_ERROR: información de error
• Nomenclatura de campos: los campos de evento usan camelCase (por ejemplo, threadId, runId, messageId)
• Administración de subprocesos: El threadId mantiene el contexto de conversación entre las solicitudes.
• Instrucciones del Cliente: Los mensajes del sistema se envían desde el cliente

Configurar y ejecutar el cliente

Opcionalmente, establezca una dirección URL de servidor personalizada:

export AGUI_SERVER_URL="http://127.0.0.1:8888/"

Ejecute el cliente (en un terminal independiente):

python client.py

Paso 3: Probar el sistema completo

Con el servidor y el cliente en ejecución, ahora puede probar el sistema completo.

Resultado esperado

$ python client.py
Connecting to AG-UI server at: http://127.0.0.1:8888/

User (:q or quit to exit): What is 2 + 2?

[Run Started - Thread: abc123, Run: xyz789]
2 + 2 equals 4.
[Run Finished - Thread: abc123, Run: xyz789]

User (:q or quit to exit): Tell me a fun fact about space

[Run Started - Thread: abc123, Run: def456]
Here's a fun fact: A day on Venus is longer than its year! Venus takes
about 243 Earth days to rotate once on its axis, but only about 225 Earth
days to orbit the Sun.
[Run Finished - Thread: abc123, Run: def456]

User (:q or quit to exit): :q

Salida codificada por colores

El cliente muestra diferentes tipos de contenido con colores distintos:

• Amarillo: ejecutar notificaciones iniciadas
• Cian: respuestas de texto del agente (transmitidas en tiempo real)
• Verde: Ejecutar notificaciones de finalización
• Rojo: mensajes de error

Pruebas con curl (opcional)

Antes de ejecutar el cliente, puede probar el servidor manualmente mediante curl:

curl -N http://127.0.0.1:8888/ \
  -H "Content-Type: application/json" \
  -H "Accept: text/event-stream" \
  -d '{
    "messages": [
      {"role": "user", "content": "What is 2 + 2?"}
    ]
  }'

Debería ver un flujo de eventos enviados por el servidor.

data: {"type":"RUN_STARTED","threadId":"...","runId":"..."}

data: {"type":"TEXT_MESSAGE_START","messageId":"...","role":"assistant"}

data: {"type":"TEXT_MESSAGE_CONTENT","messageId":"...","delta":"The"}

data: {"type":"TEXT_MESSAGE_CONTENT","messageId":"...","delta":" answer"}

...

data: {"type":"TEXT_MESSAGE_END","messageId":"..."}

data: {"type":"RUN_FINISHED","threadId":"...","runId":"..."}

En un flujo inactivo, curl también puede mostrar líneas de comentario : keepalive. Estos son comentarios de transporte de SSE, no eventos de AG-UI.

Funcionamiento

flujo del lado del servidor

1. El cliente envía una solicitud HTTP POST con mensajes
2. El punto de conexión de FastAPI recibe la solicitud.
3. AgentFrameworkAgent el envoltorio orquesta la ejecución
4. El agente procesa los mensajes mediante Agent Framework
5. AgentFrameworkEventBridge convierte las actualizaciones del agente en eventos de AG-UI
6. Las respuestas se transmiten como Server-Sent Events (SSE)
7. La conexión se cierra cuando se completa la ejecución

Flujo del lado del cliente

1. El cliente envía la solicitud HTTP POST al punto de conexión del servidor
2. El servidor responde con secuencia SSE
3. El cliente analiza las líneas entrantes data: como eventos JSON
4. Cada evento se muestra en función de su tipo.
5. threadId se captura para la continuidad de la conversación
6. El stream se completa cuando llega el evento RUN_FINISHED

Detalles del protocolo

El protocolo AG-UI usa:

• HTTP POST para enviar solicitudes
• Eventos Server-Sent (SSE) para respuestas en streaming
• JSON para la serialización de eventos
• Identificadores de hilos para mantener el contexto de conversación
• Identificadores de ejecución para seguimiento de ejecuciones individuales
• Nomenclatura de tipo de evento: UPPERCASE con caracteres de subrayado (por ejemplo, RUN_STARTED, TEXT_MESSAGE_CONTENT)
• Nomenclatura de campos: camelCase (por ejemplo, threadId, runId, messageId)
• Comentarios keepalive de SSE cada 15 segundos mientras una secuencia está inactiva. Los clientes que procesan solo líneas data: ignoran estos comentarios automáticamente.

Patrones comunes

Configuración del servidor personalizado

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# Add CORS for web clients
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

add_agent_framework_fastapi_endpoint(
    app,
    agent,
    "/agent",
    keepalive_seconds=30,  # Defaults to 15; set to None to disable
)

keepalive_seconds debe ser un número positivo o None.

Múltiples agentes

app = FastAPI()

weather_agent = Agent(name="weather", ...)
finance_agent = Agent(name="finance", ...)

add_agent_framework_fastapi_endpoint(app, weather_agent, "/weather")
add_agent_framework_fastapi_endpoint(app, finance_agent, "/finance")

Tratamiento de errores

try:
    async for event in client.send_message(message):
        if event.get("type") == "RUN_ERROR":
            error_msg = event.get("message", "Unknown error")
            print(f"Error: {error_msg}")
            # Handle error appropriately
except httpx.HTTPError as e:
    print(f"HTTP error: {e}")
except Exception as e:
    print(f"Unexpected error: {e}")

Troubleshooting

Conexión rechazada

Asegúrese de que el servidor se está ejecutando antes de iniciar el cliente:

# Terminal 1
python server.py

# Terminal 2 (after server starts)
python client.py

Errores de autenticación

Asegúrese de que está autenticado con Azure:

az login

Compruebe que tiene la asignación de roles correcta en el recurso de Azure OpenAI.

Streaming no funciona

Compruebe que el tiempo de espera del cliente es suficiente:

httpx.AsyncClient(timeout=60.0)  # 60 seconds should be enough

En el caso de los agentes de larga duración, aumente el tiempo de espera en consecuencia.

Los flujos inactivos emiten un comentario keepalive de SSE cada 15 segundos de forma predeterminada. Si un proxy cierra las conexiones inactivas antes, configure un valor positivo keepalive_seconds más pequeño al registrar el punto de conexión.

Contexto de hilo perdido

El cliente controla automáticamente la continuidad del hilo. Si se pierde el contexto:

1. Comprobación de que threadId se está capturando de RUN_STARTED eventos
2. Asegúrese de que se usa la misma instancia de cliente en todos los mensajes.
3. Compruebe que el servidor recibe el thread_id en las solicitudes posteriores.

Pasos siguientes

Ahora que comprende los conceptos básicos de AG-UI, puede hacer lo siguiente:

• Agregar herramientas de backend: crear herramientas de función personalizadas para tu dominio

Recursos adicionales

• Información general deAG-UI
• Documentación de Agent Framework
• Especificación del protocoloAG-UI

Go admite AG-UI a través de provider/aguiprovider tanto para servidores como para clientes.

import "github.com/microsoft/agent-framework-go/provider/aguiprovider"

mux := http.NewServeMux()
mux.Handle("/", aguiprovider.NewJSONHTTPHandler(myAgent, aguiprovider.HandlerConfig{}))

if err := http.ListenAndServe(":8888", mux); err != nil {
    log.Fatal(err)
}

Use aguiprovider.NewAgent cuando la aplicación Go necesite llamar a un servidor AG-UI como agente:

import aguiSSEClient "github.com/ag-ui-protocol/ag-ui/sdks/community/go/pkg/client/sse"

a := aguiprovider.NewAgent(
    aguiSSEClient.NewClient(aguiSSEClient.Config{Endpoint: serverURL}),
    aguiprovider.AgentConfig{},
)