Komma igång med AG-UI

Förutsättningar

• .NET 8 eller senare
• Ett ASP.NET Core projekt
• En konfigurerad MAF AIAgent

Exemplet använder Azure OpenAI, men MapAGUIServer fungerar med alla MAF-agenter.

Skapa en AG-UI-server

Installera värdpaketet:

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

Registrera AG-UI värd och mappa din agent:

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 tar emot AG-UI-RunAgentInputbegäranden och skickar agentens svar som en ström av AG-UI-händelser över server-sent events (SSE).

Kör servern på den URL som används av klientexemplet:

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

Ansluta med en .NET-klient

AG-UI .NET SDK tillhandahåller AGUIChatClient, som implementerar IChatClient och kan anpassas till en MAF-agent:

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

Du kan också ansluta med alla klienter som implementerar AG-UI protokollet.

Konversationskontinuitet

AG-UI använder threadId och parentRunId för att identifiera fortsättningsbegäranden. Dessa identifieringsuppgifter är protokolldata, inte behörighetsuppgifter för auktorisering.

AGUIChatClient är tillståndslös. Om du vill fortsätta en serverägd konversation hämtar du identifierarna från den första interaktionens RunStartedEvent och inkluderar sedan samma threadId och föregående runId som parentRunId i nästa förfrågan:

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

Skicka endast de nya meddelandena i en fortsättningsbegäran. MapAGUIServer använder threadId för att välja sessionen för den värdbaserade agenten och parentRunId för att identifiera den körning som fortsätts. Utan värdbaserad sessionspersistence får varje begäran en ny serversession. klienten kan i stället skicka om konversationshistoriken.

Om du vill behålla serverägt AgentSession tillstånd mellan begäranden konfigurerar du värdbaserad sessionspersistens och isolering och mappar sedan den namngivna värdbaserade agenten med MapAGUIServer. Information om den AG-UI-specifika förtroendegränsen finns i Överväganden för produktion och säkerhet.

Nästa steg

Relaterade resurser

• AG-UI översikt
• MAF-värdtjänst
• dokumentation omAG-UI protokoll

Förutsättningar

Kontrollera att du har följande innan du börjar:

• Python 3.10 eller senare
• Azure OpenAI-tjänstslutpunkt och distribution konfigurerad
• Azure CLI installerat och autentiserat
• Användaren har Cognitive Services OpenAI Contributor rollen för Azure OpenAI-resursen

Steg 1: Skapa en AG-UI Server

AG-UI-servern är värd för din AI-agent och exponerar den via HTTP-slutpunkter med FastAPI.

Installera nödvändiga paket

Installera nödvändiga paket för servern:

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

Eller med uv:

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

Detta installerar automatiskt agent-framework-core, fastapi, uvicorn och sse-starlette som beroenden.

Serverkod

Skapa en fil med namnet 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)

Viktiga begrepp

• add_agent_framework_fastapi_endpoint: Registrerar AG-UI slutpunkten med automatisk hantering av begäran/svar och SSE-strömning
• Agent: Agent Framework-agenten som hanterar inkommande begäranden
• FastAPI-integrering: Använder FastAPI:s interna asynkrona stöd för strömningssvar
• Instruktioner: Agenten skapas med standardinstruktioner som kan åsidosättas av klientmeddelanden
• Konfiguration: OpenAIChatCompletionClient accepterar explicita Azure-routningsindata som model, azure_endpoint, api_versionoch och credentialkan också läsa från miljövariabler

Konfigurera och köra servern

Ange nödvändiga miljövariabler:

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

Kör servern:

python server.py

Eller använda uvicorn direkt:

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

Servern börjar lyssna på http://127.0.0.1:8888.

Steg 2: Skapa en AG-UI-klient

Den AG-UI klienten ansluter till fjärrservern och visar strömmande svar.

AGUIChatClient återanvänder en internt ägd HTTP-klient som inte bevarar svarscookies.

Om servern kräver kakor för autentisering, sessioner eller lastbalanserarens affinitet skickar du med ett anroparägt httpx.AsyncClient via http_client=. Begränsa klienten till ett autentiserat säkerhetsobjekt och stäng sedan klienten i programmet. Ett AG-UI tråd-ID är en korrelationsidentifierare, inte en autentiseringsgräns.

Installera de paket som krävs

Det AG-UI paketet är redan installerat, som innehåller AGUIChatClient:

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

Klientkod

Skapa en fil med namnet 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())

Skicka multimodala meddelanden

AGUIChatClient bevarar ordningen på text- och mediedelar i användarmeddelanden. Använd en MIME-typ för att identifiera URI eller infogat bild-, ljud-, video- och dokumentinnehåll:

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)

Viktiga begrepp

• Server-Sent Events (SSE): Protokollet använder SSE-format (data: {json}\n\n)
• Händelsetyper: Olika händelser tillhandahåller metadata och innehåll (VERSALER med understreck):
  • RUN_STARTED: Agenten har börjat bearbeta
  • TEXT_MESSAGE_START: Start av ett textmeddelande från agenten
  • TEXT_MESSAGE_CONTENT: Inkrementell text som strömmas från agenten, med fältet delta
  • TEXT_MESSAGE_END: Slutet på ett textmeddelande
  • RUN_FINISHED: Slutfört
  • RUN_ERROR: Felinformation
• Fältnamngivning: Händelsefält använder camelCase (t.ex. threadId, runId, messageId)
• Trådhantering: Upprätthåller threadId konversationskontexten mellan begäranden
• Client-Side instruktioner: Systemmeddelanden skickas från klienten

Konfigurera och köra klienten

Du kan också ange en anpassad server-URL:

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

Kör klienten (i en separat terminal):

python client.py

Steg 3: Testa det fullständiga systemet

När både servern och klienten körs kan du nu testa hela systemet.

Förväntade utdata

$ 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

Färgkodad utdata

Klienten visar olika innehållstyper med distinkta färger:

• Gul: Meddelanden om start av körning
• Cyan: Agenttextsvar (strömmas i realtid)
• Grön: Kör aviseringar om slutförande
• Röd: Felmeddelanden

Testa med curl (valfritt)

Innan du kör klienten kan du testa servern manuellt med 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?"}
    ]
  }'

Du bör se Server-Sent Events som strömmas tillbaka:

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

För en inaktiv dataström kan curl också visa kommentarrader : keepalive. Det här är SSE-transportkommenteringar, inte AG-UI händelser.

Så här fungerar det

Flöde på Serversidan

1. Klienten skickar HTTP POST-begäran med meddelanden
2. FastAPI-slutpunkten tar emot begäran
3. AgentFrameworkAgent wrapper orkestrerar körningen
4. Agenten bearbetar meddelandena med Agent Framework
5. AgentFrameworkEventBridge konverterar agentuppdateringar till AG-UI händelser
6. Svar strömmas tillbaka som Server-Sent Events (SSE)
7. Anslutningen stängs när processen är klar

Klientsideflöde

1. Klienten skickar HTTP POST-begäran till serverslutpunkten
2. Servern svarar med SSE-ström
3. Klienten parsar inkommande data: rader som JSON-händelser
4. Varje händelse visas baserat på dess typ
5. threadId samlas in för konversationskontinuitet
6. Flödet kompletteras när RUN_FINISHED händelsen anländer

Protokollinformation

Protokollet AG-UI använder:

• HTTP POST för att skicka begäranden
• Server-Sent Events (SSE) för strömningssvar
• JSON för händelse serialisering
• Tråd-ID:t för att upprätthålla konversationskontext
• Kör ID:t för att spåra enskilda körningar
• Namngivning av händelsetyp: VERSALER med understreck (t.ex. RUN_STARTED, TEXT_MESSAGE_CONTENT)
• Namn på fält: camelCase (t.ex. threadId, runId, messageId)
• SSE keepalive-kommentarer var 15:e sekund när strömmen är inaktiv. Klienter som endast bearbetar data: rader ignorerar dessa kommentarer automatiskt.

Vanliga mönster

Lägga till interaktiva A2UI-ytor

A2UI låter en agent generera interaktiva ytor som en kompatibel AG-UI-klient renderar allteftersom svaret strömmas. Installera det valfria A2UI-beroendet:

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

För uv pip install "agent-framework-ag-ui[a2ui]" --prerelease=allow, kör uv.

Om du vill aktivera A2UI som standard för en slutpunkt skickar du ett a2ui_config med backendens opt-in-inställning:

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

Klientdelen behöver AG-UI mellanprogram för A2UI eller motsvarande beteende för att:

• Ange dess komponentkatalog och genereringsvägledning i AG-UI begärandekontext.
• Rendera de strömmade verktygsargumenten render_a2ui som ytuppdateringar.
• Ange forwardedProps.injectA2UITool när A2UI är aktiverat för en begäran. Ett uttryckligt false-värde åsidosätter backendens aktivering för den begäran.

Används OpenAIChatCompletionClient för agenten som hanterar A2UI-slutpunkten. Den strömmar argumentdelta för verktygsanrop så att klienten kan måla ytan progressivt. Api-klienten för svar stöder inte det här progressiva återgivningsflödet.

Adaptern validerar slutförda komponentträd mot den angivna katalogen. Om verifieringen misslyckas lägger den till valideringsfelen i generationsprompten och försöker igen enligt konfigurationen recovery . Om återställningsförsöken har uttömts tar klienten emot ett felmeddelande i ett kuvert i stället för ett undantag.

Fullständiga implementeringar finns i Python A2UI-agenter och A2UI-slutpunktskonfiguration på Agent Framework-lagringsplatsen.

Anpassad serverkonfiguration

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 måste vara ett positivt tal eller None.

Fortsätt ändliga körningar efter frånkoppling

Som standard kan frånkoppling av SSE-klienten avbryta svarsgeneratorn och stoppa körningen. Ange detached_runs=True när en begränsad agent eller arbetsflödeskörning måste fortsätta tillräckligt länge för att slutföra ögonblicksbilder, kontrollpunkter eller godkännandetillstånd:

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

När du konfigurerar beständighet för ögonblicksbilder anger du även en betrodd snapshot_scope_resolver. En klient som återansluter återställer den senaste sparade Thread Snapshot; frikopplad körning återger inte SSE-händelser som emitteras medan ingen klient är ansluten.

Varje slutpunktsregistrering behåller som mest max_detached_runs producenter och returnerar HTTP 503 när kapaciteten är full. Det returnerar HTTP 409 för en annan ändringsbegäran på samma (Snapshot Scope, threadId) medan en frikopplad körning är aktiv. Den här samordningen är processlokal, så program med flera arbetare måste tillhandahålla distribuerad serialisering.

FastAPI-beroenden och andra resurser med begäransomfång kan släppas när det frånkopplade svaret har upphört. Fastställ auktorisering, snapshot-omfång och andra beständiga värden innan körningen startar, och behåll inte klienter eller sessioner som ägs av begäran i fristående verktyg och leverantörer.

Flera agenter

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

Felhantering

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

Anslutningen nekades

Kontrollera att servern körs innan du startar klienten:

# Terminal 1
python server.py

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

Autentiseringsfel

Kontrollera att du är autentiserad med Azure:

az login

Kontrollera att du har rätt rolltilldelning för Azure OpenAI-resursen.

Strömning fungerar inte

Kontrollera att klientens timeout är tillräcklig:

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

För långkörande agenter ökar du tidsgränsen därefter.

Inaktiva strömmar genererar en SSE keepalive-kommentar var 15:e sekund som standard. Om en proxy stänger inaktiva anslutningar tidigare konfigurerar du ett mindre positivt keepalive_seconds värde när du registrerar slutpunkten.

Trådkontext förlorad

Klienten hanterar automatiskt trådkontinuitet. Om kontexten går förlorad:

1. Kontrollera att threadId samlas in från RUN_STARTED händelser
2. Kontrollera att samma klientinstans används mellan meddelanden
3. Kontrollera att servern tar emot thread_id i efterföljande begäranden

Nästa steg

Nu när du förstår grunderna i AG-UI kan du:

• Lägg till serverdelsverktyg: Skapa anpassade funktionsverktyg för din domän

Ytterligare resurser

• översikt överAG-UI
• Dokumentation om Agent Framework
• AG-UI protokollspecifikation

Go stöder AG-UI via provider/aguiprovider för både servrar och klienter.

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

Använd aguiprovider.NewAgent när din Go-app behöver anropa en AG-UI-server som en agent:

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