Introdução ao AG-UI

Este tutorial demonstra como criar aplicativos cliente e servidor usando o protocolo AG-UI com o Agent Framework. Você aprenderá a hospedar um agente em um endpoint AG-UI e conectar um cliente para conversas interativas.

O que você construirá

Ao final deste tutorial, você terá:

  • Um servidor AG-UI que hospeda um agente de IA acessível via HTTP
  • Um aplicativo cliente que se conecta ao servidor e transmite respostas
  • Noções básicas de como o protocolo AG-UI funciona com o Agent Framework

Prerequisites

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

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

Criar um servidor AG-UI

Instale o pacote de hospedagem:

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

Registre AG-UI hospedagem e mapeie seu 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 solicitações RunAgentInput de AG-UI e transmite a resposta do agente como eventos AG-UI usando eventos enviados pelo servidor (SSE).

Execute o servidor na URL usada pelo exemplo do cliente:

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

Tip

Consulte o exemplo de primeiros passos do .NET para ver um servidor e um cliente de console completos.

Conectar-se com um cliente .NET

O SDK do 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);
    }
}

Você também pode se conectar a qualquer cliente que implemente o protocolo AG-UI.

Continuidade da conversa

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

AGUIChatClient é sem estado. Para continuar uma conversa pertencente ao servidor, obtenha os identificadores RunStartedEvent do primeiro turno. Depois, inclua o mesmo threadId e o runId anterior como parentRunId na próxima solicitação:

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 em uma solicitação de continuação. MapAGUIServer usa threadId para selecionar a sessão do agente hospedado e parentRunId identificar a execução que está sendo continuada. Sem persistência de sessão hospedada, cada solicitação recebe uma nova sessão de servidor; em vez disso, o cliente pode reenviar o histórico de conversas.

Para manter o estado pertencente ao servidor AgentSession entre solicitações, configure persistência e isolamento de sessão hospedada e, depois, mapeie o agente hospedado nomeado com MapAGUIServer. Para obter o limite de confiança específico do AG-UI, consulte considerações sobre produção e segurança.

Próximas Etapas 

Prerequisites

Antes de começar, verifique se você tem o seguinte:

Note

Esses exemplos usam modelos do Azure OpenAI. Para obter mais informações, confira como implantar modelos do Azure OpenAI com o Foundry.

Note

Esses exemplos usam DefaultAzureCredential para autenticação. Verifique se você está autenticado com o Azure (por exemplo, via az login). Para obter mais informações, consulte a documentação da Identidade do Azure.

Aviso

O protocolo AG-UI ainda está em desenvolvimento e está sujeito a alterações. Manteremos essas amostras atualizadas à medida que o protocolo evoluir.

Etapa 1: Criando um servidor AG-UI

O servidor AG-UI hospeda seu agente de IA e o disponibiliza através de 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

Isso 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 endpoint AG-UI com gerenciamento automático de solicitações/respostas e streaming de SSE
  • Agent: o agente do Agent Framework que lidará com solicitações de entrada
  • Integração do FastAPI: usa o suporte assíncrono nativo do FastAPI para respostas de streaming
  • Instruções: O agente é criado com instruções padrão, que podem ser substituídas por mensagens de cliente
  • Configuração: OpenAIChatCompletionClient aceita entradas explícitas de roteamento do Azure, como model, azure_endpoint, api_version, e credential, e também pode ler de variáveis do 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 o uvicorn diretamente:

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

O servidor começará a escutar em 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.

AGUIChatClient reutiliza um cliente HTTP de propriedade interna que não persiste cookies de resposta.

Se o servidor exigir cookies para autenticação, sessões ou afinidade do balanceador de carga, passe um httpx.AsyncClient pertencente ao chamador usando http_client=. Restrinja o cliente a uma entidade autenticada e feche-o no seu aplicativo. Uma ID de thread AG-UI é um identificador de correlação, não um limite de autenticação.

Instalar pacotes necessários

O pacote AG-UI já está instalado, o que inclui :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())

Enviar mensagens multimodal

AGUIChatClient preserva a ordem de partes de texto e mídia em mensagens de usuário. Use um tipo MIME para identificar o conteúdo de URI ou de imagem embutida, áudio, vídeo e documento:

from pathlib import Path

from agent_framework import Content, Message

message = Message(
    role="user",
    contents=[
        Content.from_text("Compare this image with the attached brief."),
        Content.from_uri(
            "https://example.com/product.png",
            media_type="image/png",
        ),
        Content.from_data(
            Path("brief.pdf").read_bytes(),
            media_type="application/pdf",
        ),
    ],
)

await agent.run(message, session=thread)

Novas URIs de mídia de entrada devem usar o esquema http, data ou https. Também há suporte para dados binários embutidos. Resolva e autorize referências file, FTP, relativas, de esquema personalizado e id ou file no nível do protocolo no código do aplicativo antes de enviá-las como HTTPS, URI de dados ou conteúdo embutido.

Quando um Instantâneo de Thread persistente contém uma referência de mídia sem suporte mais antiga, o adaptador de Python o substitui por texto de espaço reservado inerte antes da reprodução do modelo e remove a referência na próxima vez em que salva o instantâneo.

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 (UPPERCASE com sublinhados):
    • RUN_STARTED: o agente iniciou o processamento
    • TEXT_MESSAGE_START: início de uma mensagem de texto do agente
    • TEXT_MESSAGE_CONTENT: texto incremental transmitido do agente (com o 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 campo: os campos de evento usam camelCase (por exemplo, , threadId, runId, messageId)
  • Gerenciamento de threads: o threadId mantém o contexto da 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

Saída com código de cores

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

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

Testando com curl (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ê deverá ver os Eventos Enviados pelo Servidor (SSE) sendo transmitidos de volta:

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 fluxo inativo, curl também pode exibir linhas de comentário : keepalive. Esses 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 do FastAPI recebe a solicitação
  3. AgentFrameworkAgent wrapper orquestra a execução
  4. O agente processa as mensagens usando o Agent Framework
  5. AgentFrameworkEventBridge converte atualizações de agente em eventos de AG-UI
  6. As respostas são transmitidas como Eventos Enviados pelo Servidor (SSE)
  7. A conexão é fechada quando a execução é concluída

Fluxo de Client-Side

  1. O cliente envia solicitação HTTP POST para o ponto de extremidade do servidor
  2. O servidor responde com o 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 usa:

  • HTTP POST para enviar solicitações
  • Server-Sent Events (SSE) para respostas de streaming
  • JSON para serialização de eventos
  • IDs de thread para manter o contexto de conversação
  • IDs de execução para rastrear execuções individuais
  • Nomenclatura de tipo de evento: UPPERCASE com sublinhados (por exemplo, RUN_STARTED, TEXT_MESSAGE_CONTENT)
  • Nomenclatura de campo: camelCase (por exemplo, threadId, runId, messageId)
  • Comentários de keepalive SSE a cada 15 segundos enquanto um fluxo estiver ocioso. Os clientes que processam apenas data: linhas ignoram esses comentários automaticamente.

Padrões comuns

Adicionar superfícies interativas A2UI

O A2UI permite que um agente gere superfícies interativas que um cliente AG-UI compatível renderiza como fluxos de resposta. Instale a dependência opcional de A2UI:

pip install "agent-framework-ag-ui[a2ui]" --pre

Para uv, execute uv pip install "agent-framework-ag-ui[a2ui]" --prerelease=allow.

Para habilitar o A2UI por padrão para um endpoint, passe um a2ui_config com a opção de adesão do backend:

from agent_framework.ag_ui import add_agent_framework_fastapi_endpoint

add_agent_framework_fastapi_endpoint(
    app=app,
    agent=agent,
    path="/a2ui",
    a2ui_config={"inject_a2ui_tool": True},
)

O front-end precisa do middleware AG-UI A2UI ou comportamento equivalente para:

  • Forneça seu catálogo de componentes e as orientações para geração no contexto da solicitação AG-UI.
  • Renderize os argumentos da ferramenta transmitida render_a2ui como atualizações de superfície.
  • Defina forwardedProps.injectA2UITool quando a A2UI está habilitada para uma solicitação. Um valor explícito false substitui a opção de adesão do backend para essa solicitação.

O adaptador apresenta catálogos de componentes fornecidos pelo cliente e diretrizes de geração para o modelo como dados de usuário não confiáveis. Mantenha a política confiável e as instruções de renderização nas instruções ou na configuração do agente de propriedade do servidor.

Use OpenAIChatCompletionClient para o agente que atende o endpoint A2UI. Ele transmite deltas de argumento de chamada de ferramenta para que o cliente possa pintar a superfície progressivamente. O cliente de API de Respostas não dá suporte a esse fluxo de renderização progressivo.

O adaptador valida árvores de componentes concluídas em relação ao catálogo fornecido. Se a validação falhar, ela adicionará os erros de validação ao prompt de geração e tentará novamente de acordo com a recovery configuração. Se a recuperação estiver esgotada, o cliente receberá um envelope de falha em vez de uma exceção.

Para implementações completas, consulte os agentes A2UI em Python e a configuração do endpoint A2UI no repositório do Agent Framework.

Configuração do 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 deve ser um número positivo ou None.

Continuar execuções finitas depois de desconectar

Por padrão, desconectar o cliente SSE pode cancelar o gerador de resposta e interromper a execução. Defina detached_runs=True quando um agente finito ou uma execução de fluxo de trabalho deve continuar por tempo suficiente para finalizar instantâneos, pontos de verificação ou estado de aprovação:

add_agent_framework_fastapi_endpoint(
    app,
    agent,
    "/agent",
    detached_runs=True,
    max_detached_runs=32,
    detached_run_timeout_seconds=3600,
)

Ao configurar a persistência de snapshot, também forneça um snapshot_scope_resolver confiável. Um cliente que se reconecta recupera o Snapshot de Thread confirmado mais recente; a execução desanexada não reproduz eventos SSE emitidos enquanto nenhum cliente está conectado.

Cada registro de endpoint retém no máximo max_detached_runs produtores e retorna HTTP 503 quando essa capacidade é atingida. Ele retorna HTTP 409 para uma outra solicitação de modificação na mesma (Snapshot Scope, threadId) enquanto uma execução em segundo plano está ativa. Essa coordenação é local de processo, portanto, aplicativos com vários trabalhadores devem fornecer serialização distribuída.

As dependências do FastAPI e outros recursos com escopo da solicitação podem ser liberados após o encerramento da resposta devido à desconexão. Resolva a autorização, o escopo do snapshot e outros valores duráveis antes de a execução começar e não retenha clientes ou sessões pertencentes à solicitação em ferramentas e provedores desanexados.

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}")

AGUIChatClient pode executar uma chamada de ferramenta de cliente local completa sem um evento TOOL_CALL_END. Se RUN_ERROR chegar depois que o cliente expõe uma chamada de ferramenta local, o cliente gerará ResponseInvalidatedException e descartará a chamada em vez de executar um efeito colateral de uma resposta inválida.

Troubleshooting

Conexã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

Verifique se você está autenticado com o Azure:

az login

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

Streaming não funcionando

Verifique se o tempo limite configurado para o cliente é adequado.

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

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

Fluxos ociosos emitem um comentário de keepalive SSE a cada 15 segundos por padrão. Se um proxy fechar conexões ociosas mais cedo, configure um valor keepalive_seconds positivo menor ao registrar o ponto de extremidade.

Contexto da 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á recebendo o thread_id em solicitações subsequentes

Próximas etapas

Agora que você entende as noções básicas do AG-UI, você pode:

Recursos adicionais

Go oferece suporte à AG-UI por meio de provider/aguiprovider para ambos os 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 seu aplicativo Go precisar chamar um servidor AG-UI como um 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{},
)

Tip

Consulte os exemplos de servidor de primeiros passos do AG-UI e de cliente para ver exemplos completos e executáveis.