Introdução ao AG-UI

Pré-requisitos

• .NET 8 ou posterior
• Um projeto ASP.NET Core
• Um MAF configurado AIAgent

O exemplo usa o Azure OpenAI, mas MapAGUIServer funciona com qualquer agente MAF.

Criar um servidor AG-UI

Instale o pacote de alojamento:

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

Registe AG-UI alojamento e mapeie o teu 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 aceita pedidos AG-UI RunAgentInput e transmite a resposta do agente como eventos AG-UI através de Server-Sent Events (SSE).

Execute o servidor na URL utilizada pelo exemplo do cliente:

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

Ligue-se a um cliente .NET

O SDK AG-UI .NET fornece AGUIChatClient, que implementa IChatClient e pode ser adaptado para um 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);
    }
}

Também pode ligar-se a qualquer cliente que implemente o protocolo AG-UI.

Continuidade da conversa

AG-UI utiliza threadId e parentRunId para identificar solicitações de continuação. Estes identificadores são dados de protocolo, não credenciais de autorização.

AGUIChatClient é apátrida. Para continuar uma conversa pertencente ao servidor, obtenha os identificadores a partir de RunStartedEvent da primeira interação e, em seguida, inclua o mesmo threadId e o runId anterior como parentRunId no pedido seguinte:

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

Envie apenas as novas mensagens num pedido de continuação. MapAGUIServer utiliza threadId para selecionar a sessão do agente alojado e parentRunId para identificar a execução que está a ser continuada. Sem persistência de sessão alojada, cada pedido recebe uma nova sessão de servidor; O cliente pode, em vez disso, reenviar o histórico das conversas.

Para manter o estado AgentSession pertencente ao servidor entre pedidos, configure a persistência e o isolamento da sessão alojada e, em seguida, mapeie o agente alojado com nome com MapAGUIServer. Para a fronteira de confiança específica de AG-UI, veja Considerações de Produção e segurança.

Passos seguintes

Recursos relacionados

• AG-UI visão geral
• MAF hospedagem
• Documentação do protocolo AG-UI

Pré-requisitos

Antes de começar, certifique-se de ter o seguinte:

• Python 3.10 ou posterior
• Endpoint (ponto de extremidade) e implantação do serviço Azure OpenAI configurados
• CLI do Azure instalada e autenticada
• O usuário tem a Cognitive Services OpenAI Contributor função para o recurso OpenAI do Azure

Etapa 1: Criando um AG-UI Server

O servidor AG-UI hospeda o agente de IA e expõe-no por endpoints HTTP usando FastAPI.

Instalar pacotes necessários

Instale os pacotes necessários para o servidor:

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

Ou usando uv:

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

Isto instalará automaticamente agent-framework-core, fastapi, uvicorn e sse-starlette como dependências.

Código do servidor

Crie um arquivo chamado 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)

Conceitos-chave

• add_agent_framework_fastapi_endpoint: Registra o ponto de extremidade AG-UI com tratamento automático de solicitação/resposta e streaming SSE
• Agent: O agente do Agent Framework que lidará com solicitações de entrada
• Integração FastAPI: usa o suporte assíncrono nativo da FastAPI para respostas de streaming
• Instruções: O agente é criado com instruções padrão, que podem ser substituídas por mensagens do cliente
• Configuração: OpenAIChatCompletionClient aceita entradas explícitas de encaminhamento Azure como model, azure_endpoint, api_version, e credential, e também pode ler a partir de variáveis de ambiente

Configurar e executar o servidor

Defina as variáveis de ambiente necessárias:

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

Execute o servidor:

python server.py

Ou usando uvicorn diretamente:

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

O servidor começará a escutar no http://127.0.0.1:8888.

Etapa 2: Criando um cliente AG-UI

O cliente AG-UI se conecta ao servidor remoto e exibe respostas de streaming.

Instalar pacotes necessários

O pacote AG-UI já está instalado, o que inclui o AGUIChatClient:

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

Código do Cliente

Crie um arquivo chamado 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())

Conceitos-chave

• Server-Sent Events (SSE): O protocolo usa o formato SSE (data: {json}\n\n)
• Tipos de evento: eventos diferentes fornecem metadados e conteúdo (MAIÚSCULAS com sublinhados):
  • RUN_STARTED: O agente começou o processamento
  • TEXT_MESSAGE_START: Início de uma mensagem de texto do agente
  • TEXT_MESSAGE_CONTENT: Texto incremental transmitido do agente (com campo delta)
  • TEXT_MESSAGE_END: Fim de uma mensagem de texto
  • RUN_FINISHED: Conclusão bem-sucedida
  • RUN_ERROR: Informações de erro
• Nomenclatura de campos: Os campos de evento usam camelCase (por exemplo, threadId, , runIdmessageId)
• Gerenciamento de threads: mantém o threadId contexto de conversa entre solicitações
• Instruções do Lado do Cliente: As mensagens do sistema são enviadas do cliente

Configurar e executar o cliente

Opcionalmente, defina uma URL de servidor personalizada:

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

Execute o cliente (em um terminal separado):

python client.py

Etapa 3: Testando o sistema completo

Com o servidor e o cliente em execução, agora você pode testar o sistema completo.

Saída esperada

$ 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

Color-Coded Saída

O cliente exibe diferentes tipos de conteúdo com cores distintas:

• Amarelo: Executar notificações iniciadas
• Cyan: Respostas de texto do agente (transmitidas em tempo real)
• Verde: Executar notificações de conclusão
• Vermelho: Mensagens de erro

Teste com ondulação (Opcional)

Antes de executar o cliente, você pode testar o servidor manualmente usando 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?"}
    ]
  }'

Você deve ver eventos enviados pelo servidor sendo retransmitidos:

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":"..."}

Para um stream inativo, o curl pode também mostrar : keepalive linhas de comentário. Estes são comentários de transporte SSE, não eventos AG-UI.

Como funciona

Fluxo do Lado do Servidor

1. Cliente envia solicitação HTTP POST com mensagens
2. O endpoint FastAPI recebe a requisição
3. AgentFrameworkAgent Envoltório orquestra a execução
4. O agente processa as mensagens usando o Agent Framework
5. AgentFrameworkEventBridge Converte atualizações do agente em eventos AG-UI
6. As respostas são transmitidas de volta como Eventos Enviados pelo Servidor (SSE)
7. A conexão é fechada quando a execução é concluída

Fluxo do lado do cliente

1. Cliente envia solicitação HTTP POST para o ponto de extremidade do servidor
2. O servidor responde com fluxo SSE
3. O cliente analisa as linhas de entrada data: como eventos JSON
4. Cada evento é exibido com base em seu tipo
5. threadId é capturado para continuidade da conversa
6. O fluxo é concluído quando o evento RUN_FINISHED chega

Detalhes do protocolo

O protocolo AG-UI utiliza:

• HTTP POST para envio de solicitações
• Server-Sent Events (SSE) para streaming de respostas
• JSON para serialização de eventos
• IDs de thread para manter o contexto da conversa
• IDs de execução para rastrear execuções individuais
• Nomenclatura do tipo de evento: MAIÚSCULAS com sublinhados (por exemplo, RUN_STARTED, TEXT_MESSAGE_CONTENT)
• Nome do campo: camelCase (por exemplo, threadId, runId, messageId)
• O SSE mantém comentários em vivo a cada 15 segundos enquanto a transmissão está inativa. Os clientes que processam apenas data: linhas ignoram automaticamente estes comentários.

Padrões comuns

Configuração personalizada do servidor

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 deve ser um número positivo ou None.

Vários 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")

Tratamento de erros

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

Ligação recusada

Verifique se o servidor está em execução antes de iniciar o cliente:

# Terminal 1
python server.py

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

Erros de autenticação

Certifique-se de que está autenticado com o Azure:

az login

Verifique se você tem a atribuição de função correta no recurso OpenAI do Azure.

Streaming não funciona

Verifique se o tempo limite do seu cliente é suficiente:

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

Para agentes de longa duração, aumente o tempo limite de forma adequada.

Streams inativos emitem um comentário SSE keepalive a cada 15 segundos por defeito. Se um proxy fechar as ligações ociosas mais cedo, configure um valor positivo keepalive_seconds menor ao registar o endpoint.

Contexto de thread perdido

O cliente gerencia automaticamente a continuidade do thread. Se o contexto for perdido:

1. Verifique se threadId está sendo capturado de RUN_STARTED eventos
2. Verifique se a mesma instância do cliente é usada entre mensagens
3. Verifique se o servidor está a receber thread_id nos pedidos subsequentes

Próximas Etapas

Agora que você entende os conceitos básicos do AG-UI, você pode:

• Adicionar ferramentas de back-end: crie ferramentas de função personalizadas para o seu domínio

Recursos adicionais

• Visão geral do AG-UI
• Documentação do Agent Framework
• Especificação do protocoloAG-UI

Go suporta a AG-UI através de provider/aguiprovider para servidores e 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 quando a sua aplicação Go precisar de chamar um 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{},
)