Začínáme s AG-UI

Tento kurz ukazuje, jak sestavovat serverové a klientské aplikace pomocí protokolu AG-UI s rozhraním Agent Framework. Dozvíte se, jak hostovat agenta za koncovým bodem AG-UI a připojit klienta k interaktivním konverzacím.

Co budete vytvářet

Na konci tohoto kurzu budete mít:

  • Server AG-UI hostující agenta AI přístupný prostřednictvím protokolu HTTP
  • Klientská aplikace, která se připojuje k serveru a streamuje odpovědi
  • Principy fungování protokolu AG-UI s rozhraním Agent Framework

Prerequisites

  • .NET 8 nebo novější
  • Projekt ASP.NET Core
  • Nakonfigurovaný MAF AIAgent

Příklad používá Azure OpenAI, ale MapAGUIServer pracuje s jakýmkoli agentem MAF.

Vytvoření serveru AG-UI

Nainstalujte hostitelský balíček:

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

Zaregistrujte AG-UI hostování a namapujte agenta:

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 přijímá AG-UI RunAgentInput požadavky a streamuje odpověď agenta jako události AG-UI přes události odeslané serverem (SSE).

Spusťte server na adrese URL použité v příkladu klienta:

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

Tip

Podívejte se na ukázku Začínáme s platformou .NET, která obsahuje kompletní server a konzolového klienta.

Připojení pomocí klienta .NET

Sada SDK AG-UI pro .NET poskytuje AGUIChatClient, která implementuje IChatClient a kterou lze přizpůsobit pro agenta 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);
    }
}

Můžete se také připojit pomocí libovolného klienta, který implementuje protokol AG-UI.

Kontinuita konverzací

AG-UI používá threadId a parentRunId identifikuje žádosti o pokračování. Tyto identifikátory jsou data protokolu, nikoli autorizační přihlašovací údaje.

AGUIChatClient je bezestavová. Pokud chcete pokračovat v konverzaci spravované serverem, získejte identifikátory z prvního tahu RunStartedEvent a poté v dalším požadavku uveďte stejné threadId a předchozí runId jako parentRunId:

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

Odešlete pouze nové zprávy v žádosti o pokračování. MapAGUIServer používá threadId k výběru hostované relace agenta a parentRunId k identifikaci běhu, ve kterém se pokračuje. Bez trvalosti hostované relace obdrží každý požadavek novou relaci serveru; klient může místo toho znovu odeslat historii konverzací.

Chcete-li zachovat stav spravovaný serverem AgentSession mezi požadavky, nakonfigurujte trvalost a izolaci hostované relace a potom namapujte pojmenovaného hostovaného agenta pomocí MapAGUIServer. Pro hranici důvěry specifickou pro AG-UI viz Provozní aspekty a aspekty zabezpečení.

Další kroky

Prerequisites

Než začnete, ujistěte se, že máte následující:

Poznámka:

Tyto ukázky používají modely Azure OpenAI. Další informace najdete v tématu nasazení modelů Azure OpenAI s Foundry.

Poznámka:

Tyto ukázky používají DefaultAzureCredential k autentizaci. Ujistěte se, že jste ověřeni pomocí Azure (např. prostřednictvím az login). Další informace najdete v dokumentaci k identitě Azure.

Warning

Protokol AG-UI je stále ve vývoji a může se změnit. Tyto ukázky budeme udržovat aktualizované, jak se protokol vyvíjí.

Krok 1: Vytvoření serveru AG-UI

Server AG-UI hostuje vašeho agenta AI a zveřejňuje ho prostřednictvím koncových bodů HTTP pomocí FastAPI.

Instalace požadovaných balíčků

Nainstalujte potřebné balíčky pro server:

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

Nebo použijte uv:

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

Tím se automaticky nainstaluje agent-framework-core, fastapi, uvicorna sse-starlette jako závislosti.

Kód serveru

Vytvořte soubor s názvem 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)

Klíčové koncepty

  • add_agent_framework_fastapi_endpoint: Zaregistruje koncový bod AG-UI pomocí automatického zpracování požadavků a odpovědí a streamování SSE.
  • Agent: Agent frameworku, který bude zpracovávat příchozí požadavky
  • Integrace FastAPI: Používá nativní asynchronní podporu FastAPI pro streamování odpovědí.
  • Pokyny: Agent se vytvoří s výchozími pokyny, které je možné přepsat pomocí zpráv klienta.
  • Konfigurace: OpenAIChatCompletionClient přijímá explicitní vstupy směrování Azure, jako model, azure_endpoint, api_version a credential, a může také číst z proměnných prostředí.

Konfigurace a spuštění serveru

Nastavte požadované proměnné prostředí:

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

Spusťte server:

python server.py

Nebo použijte uvicorn přímo:

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

Server začne naslouchat na http://127.0.0.1:8888.

Krok 2: Vytvoření klienta AG-UI

Klient AG-UI se připojí ke vzdálenému serveru a zobrazí streamové odpovědi.

AGUIChatClient používá interně vlastněného klienta HTTP, který neuchovává soubory cookie odpovědí.

Pokud server vyžaduje soubory cookie pro ověřování, relace nebo afinitu nástroje pro vyrovnávání zatížení, předejte volajícím vlastněný httpx.AsyncClient prostřednictvím http_client=. Omezte klienta na jednu ověřenou identitu a v aplikaci jej uzavřete. ID vlákna AG-UI je identifikátor korelace, nikoli hranice ověřování.

Instalace požadovaných balíčků

Balíček AG-UI je již nainstalován, který zahrnuje AGUIChatClient:

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

Klientský kód

Vytvořte soubor s názvem 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())

Odesílání vícemodálních zpráv

AGUIChatClient zachovává pořadí textových a mediálních částí v uživatelských zprávách. Pomocí typu MIME identifikujte identifikátor URI nebo vložený obrázek, zvuk, video a obsah dokumentu:

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)

Nové identifikátory URI příchozích médií musí používat schéma http, https nebo data. Podporují se také vložená binární data. Vyřešte a autorizujte file, FTP, relativní, vlastní schéma a odkazy na úrovni protokolu id nebo file v kódu aplikace před jejich odesláním jako HTTPS, identifikátor URI dat nebo vložený obsah.

Pokud trvalý snímek vlákna obsahuje starší nepodporovaný odkaz na médium, adaptér Python ho před přehráním modelu nahradí inertním zástupným textem a odebere odkaz při příštím uložení snímku.

Klíčové koncepty

  • Server-Sent Events (SSE):Protokol používá formát SSE (data: {json}\n\n)
  • Typy událostí: Různé události poskytují metadata a obsah (VELKÁ PÍSMENA s podtržítky):
    • RUN_STARTED: Agent začal zpracovávat.
    • TEXT_MESSAGE_START: Začátek textové zprávy od agenta
    • TEXT_MESSAGE_CONTENT: Inkrementální text streamovaný od agenta (s polem delta)
    • TEXT_MESSAGE_END: Konec textové zprávy
    • RUN_FINISHED: Úspěšné dokončení
    • RUN_ERROR: Informace o chybě
  • Pojmenování polí: Pole událostí používají camelCase (např. threadId, runId, messageId)
  • Správa vláken: threadId udržuje kontext konverzace napříč požadavky.
  • Client-Side Pokyny: Systémové zprávy se odesílají z klienta

Konfigurace a spuštění klienta

Volitelně můžete nastavit vlastní adresu URL serveru:

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

Spusťte klienta (v samostatném terminálu):

python client.py

Krok 3: Testování kompletního systému

S běžícím serverem i klientem teď můžete otestovat celý systém.

Očekávaný výstup

$ 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

Barevně kódovaný výstup

Klient zobrazí různé typy obsahu s odlišnými barvami:

  • Žlutá: Oznámení o zahájení spuštění
  • Azurová: Textové odpovědi agenta (streamované v reálném čase)
  • Zelená: Oznámení o dokončení spuštění
  • Červená: Chybové zprávy

Testování pomocí curl (volitelné)

Před spuštěním klienta můžete server otestovat ručně pomocí nástroje 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?"}
    ]
  }'

Měli byste vidět streamované události odesílané serverem zpět.

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

U nečinného streamu může curl také zobrazovat : keepalive řádky komentářů. Jedná se o dopravní komentáře SSE, nikoli AG-UI události.

Jak to funguje

Server-Side proces

  1. Klient odešle požadavek HTTP POST se zprávami.
  2. Koncový bod FastAPI obdrží požadavek.
  3. AgentFrameworkAgent Wrapper řídí provádění.
  4. Agent zpracovává zprávy pomocí rozhraní Agent Framework
  5. AgentFrameworkEventBridge převádí aktualizace agenta na události AG-UI.
  6. Odpovědi se streamují zpět jako události Server-Sent (SSE)
  7. Připojení se uzavře po dokončení spuštění.

Klientský proces

  1. Klient odešle požadavek HTTP POST do koncového bodu serveru.
  2. Server reaguje streamem SSE
  3. Klient analyzuje příchozí data: řádky jako události JSON.
  4. Každá událost se zobrazí na základě jejího typu.
  5. threadId je zachycen pro kontinuitu konverzace.
  6. Stream se dokončí, když přijde událost RUN_FINISHED.

Podrobnosti protokolu

Protokol AG-UI používá:

  • HTTP POST pro odesílání požadavků
  • Události Server-Sent (SSE) pro streamování odpovědí
  • JSON pro serializaci událostí
  • ID vláken pro zachování kontextu konverzace
  • Identifikátory běhu pro sledování jednotlivých spouštění
  • Pojmenování typu události: VELKÁ PÍSMENA s podtržítky (např. RUN_STARTED, TEXT_MESSAGE_CONTENT)
  • Pojmenování polí: camelCase (např. threadId, runId, messageId)
  • SSE udržuje komentáře každých 15 sekund, když je stream nečinný. Klienti, kteří zpracovávají jenom data: řádky, ignorují tyto komentáře automaticky.

Běžné vzory

Přidejte interaktivní plochy A2UI

A2UI umožňuje agentu generovat interaktivní povrchy, které kompatibilní AG-UI klient vykresluje jako streamy odpovědí. Nainstalujte volitelnou závislost A2UI:

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

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

Pokud chcete ve výchozím nastavení povolit A2UI pro koncový bod, předejte a2ui_config se souhlasem backendu:

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

Frontend potřebuje middleware AG-UI A2UI, nebo ekvivalentní chování, aby:

  • Uveďte pokyny k katalogu komponent a generování v kontextu požadavku AG-UI.
  • Vykreslujte streamované argumenty nástroje render_a2ui formou aktualizací rozhraní.
  • Nastavte forwardedProps.injectA2UITool , když je pro požadavek povolená funkce A2UI. Explicitní hodnota false má přednost před přihlášením v backendu pro daný požadavek.

Adaptér představuje katalogy komponent dodané klientem a pokyny ke generování předkládané modelu jako nedůvěryhodná uživatelská data. Udržujte schválené zásady a pokyny pro vykreslování v instrukcích agenta spravovaných serverem nebo v jeho konfiguraci.

Používá se OpenAIChatCompletionClient pro agenta, který obsluhuje koncový bod A2UI. Průběžně přenáší dílčí změny argumentů volání nástroje, aby klient mohl rozhraní postupně vykreslovat. Klient rozhraní API pro odpovědi nepodporuje tento progresivní tok vykreslování.

Adaptér validuje dokončené stromy komponent vůči dodanému katalogu. Pokud validace selže, přidá chyby validace do promptu pro generování a pokus zopakuje podle konfigurace recovery. Pokud jsou možnosti obnovy vyčerpány, klient místo výjimky obdrží chybovou obálku.

Úplné implementace najdete v repozitáři Agent Framework v agentech Python A2UI a konfiguraci endpointu A2UI.

Vlastní konfigurace serveru

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 musí být kladné číslo nebo None.

Pokračování konečných spuštění po odpojení

Ve výchozím nastavení může odpojení klienta SSE zrušit generátor odpovědí a zastavit spuštění. Nastavte detached_runs=True, když agent s omezenou dobou běhu nebo běh pracovního postupu musí pokračovat dostatečně dlouho, aby se dokončily snímky, kontrolní body nebo stav schválení:

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

Když nakonfigurujete trvalé uchování snímků, zadejte také důvěryhodný snapshot_scope_resolver. Klient po opětovném připojení obnoví nejnovější uložený snímek vlákna; běh bez připojeného klienta nepřehrává události SSE vygenerované v době, kdy není připojen žádný klient.

Každá registrace koncového bodu může mít nanejvýš max_detached_runs producentů a po naplnění této kapacity vrací HTTP 503. Vrátí HTTP 409 u dalšího požadavku měnícího stav na stejný (Snapshot Scope, threadId), zatímco je aktivní odpojené spuštění. Tato koordinace je lokální v rámci procesu, takže aplikace s více workery musí zajistit distribuovanou serializaci.

Závislosti FastAPI a další prostředky s oborem požadavků můžou být uvolněny po ukončení odpojené odpovědi. Před zahájením běhu vyřešte autorizaci, rozsah snímku a další trvalé hodnoty a neuchovávejte klienty ani relace náležející požadavku v oddělených nástrojích a poskytovatelích.

Více agentů

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

Zpracování chyb

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 může spustit úplné volání místního klientského nástroje bez události TOOL_CALL_END. Pokud RUN_ERROR přijde poté, co klient zpřístupní volání místního nástroje, klient vyvolá ResponseInvalidatedException a zahodí volání místo spuštění vedlejšího efektu z neplatné odpovědi.

Troubleshooting

Připojení odmítnuto

Před spuštěním klienta se ujistěte, že je server spuštěný:

# Terminal 1
python server.py

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

Chyby ověřování

Ujistěte se, že jste ověřeni pomocí Azure:

az login

Ověřte, že u prostředku Azure OpenAI máte správné přiřazení role.

Nefunguje streamování

Zkontrolujte, jestli je dostatečný časový limit pro klienta:

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

U dlouhotrvajících agentů odpovídajícím způsobem zvyšte časový limit.

Nečinné streamy ve výchozím nastavení generují komentář SSE keepalive každých 15 sekund. Pokud proxy server ukončí nečinná připojení dříve, nakonfigurujte při registraci koncového bodu menší kladnou keepalive_seconds hodnotu.

Ztráta kontextu vlákna

Klient automaticky spravuje kontinuitu vláken. Pokud dojde ke ztrátě kontextu:

  1. Zkontrolujte, že threadId se zaznamenává z událostí RUN_STARTED.
  2. Ujistěte se, že se stejná instance klienta používá napříč zprávami.
  3. Ověřte, že server přijímá thread_id v následujících požadavcích.

Další kroky

Teď, když rozumíte základům AG-UI, můžete:

Další zdroje

Go podporuje prostřednictvím provider/aguiprovider AG-UI pro servery i klienty.

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

Použijte aguiprovider.NewAgent, když vaše aplikace v Go potřebuje volat server AG-UI v roli agenta:

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

Podívejte se na ukázky serveru AG-UI pro začátečníky a klienta, kde najdete kompletní spustitelné příklady.