AG-UI の始め方

前提条件

• .NET 8 以降
• ASP.NET Core プロジェクト
• 設定済みの MAF AIAgent

この例では OpenAI Azure使用しますが、MapAGUIServerは MAF エージェントで動作します。

AG-UI サーバーを作成する

ホスティング パッケージをインストールします。

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

AG-UI ホスティングを登録し、エージェントをマップします。

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 は、AG-UI RunAgentInput 要求を受け入れ、サーバー送信イベント (SSE) 経由の AG-UI イベントとしてエージェントの応答をストリーム配信します。

クライアントで使用される URL でサーバーを実行する例:

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

.NET クライアントとの接続

AG-UI .NET SDK は、IChatClientを実装し、MAF エージェントに適応できるAGUIChatClientを提供します。

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

AG-UI プロトコルを実装する任意のクライアントに接続することもできます。

会話の継続性

AG-UI では、 threadId と parentRunId を使用して継続要求を識別します。 これらの識別子はプロトコル データであり、承認資格情報ではありません。

AGUIChatClient はステートレスです。 サーバー所有の会話を続行するには、最初のターンのRunStartedEventから識別子を取得し、次の要求でparentRunIdと同じthreadIdと前のrunIdを含めます。

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

継続要求で新しいメッセージのみを送信します。 MapAGUIServer は、 threadId を使用してホストされたエージェント セッションを選択し、 parentRunId 実行が継続されていることを識別します。 ホストされたセッションの永続化がない場合、各要求は新しいサーバー セッションを受け取ります。クライアントは代わりに会話履歴を再送信できます。

要求間でサーバー所有の AgentSession 状態を保持するには、 ホストされたセッションの永続化と分離を構成し、名前付きホストエージェントを MapAGUIServerにマップします。 AG UI 固有の信頼境界については、「 運用とセキュリティに関する考慮事項」を参照してください。

次のステップ

関連するリソース

• AG-UI の概要
• MAF ホスティング
• AG-UI プロトコルのドキュメント

前提条件

開始する前に、次のことを確認してください。

• Python 3.10 以降
• Azure OpenAI サービスのエンドポイントとデプロイメントの構成
• Azure CLI のインストール と 認証
• ユーザーが Azure OpenAI リソースの Cognitive Services OpenAI Contributor ロールを持っている

手順 1: AG-UI サーバーの作成

AG-UI サーバーは、AI エージェントをホストし、FastAPI を使用して HTTP エンドポイント経由で公開します。

必要なパッケージをインストールする

サーバーに必要なパッケージをインストールします。

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

または uv を使用します。

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

これにより、依存関係として agent-framework-core、 fastapi、 uvicorn、 sse-starlette が自動的にインストールされます。

サーバー コード

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)

主な概念

• add_agent_framework_fastapi_endpoint: AG-UI エンドポイントを自動要求/応答処理と SSE ストリーミングに登録します。
• Agent: 着信要求を処理する Agent Framework エージェント
• FastAPI 統合: ストリーミング応答に FastAPI のネイティブ非同期サポートを使用します
• 手順: エージェントは、クライアント メッセージによってオーバーライドできる既定の手順で作成されます
• 構成: OpenAIChatCompletionClient は、 model、 azure_endpoint、 api_version、 credentialなどの明示的な Azure ルーティング入力を受け入れ、環境変数から読み取ることもできます

サーバーの構成と実行

必要な環境変数を設定します。

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

サーバーを実行します。

python server.py

または、uvicorn を直接使用します。

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

サーバーが http://127.0.0.1:8888でリッスンを開始します。

手順 2: AG-UI クライアントの作成

AG-UI クライアントはリモート サーバーに接続し、ストリーミング応答を表示します。

必要なパッケージをインストールする

AG-UI パッケージは既にインストールされています。これには、 AGUIChatClientが含まれています。

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

クライアント コード

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

主な概念

• Server-Sent イベント (SSE):プロトコルは SSE 形式 (data: {json}\n\n) を使用します
• イベントの種類: さまざまなイベントがメタデータとコンテンツを提供します (大文字とアンダースコア)。
  • RUN_STARTED: エージェントが処理を開始しました
  • TEXT_MESSAGE_START: エージェントからのテキスト メッセージの開始
  • TEXT_MESSAGE_CONTENT: エージェントからストリーミングされた増分テキスト ( delta フィールド付き)
  • TEXT_MESSAGE_END: テキスト メッセージの末尾
  • RUN_FINISHED: 正常に完了しました
  • RUN_ERROR: エラー情報
• フィールドの名前付け: イベント フィールドは camelCase を使用します (例: threadId、 runId、 messageId)
• スレッド管理: threadId は要求間で会話コンテキストを維持します
• Client-Side 手順: システム メッセージはクライアントから送信されます

クライアントの構成と実行

必要に応じて、カスタム サーバー URL を設定します。

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

(別のターミナルで) クライアントを実行します。

python client.py

手順 3: 完全なシステムのテスト

サーバーとクライアントの両方が実行されている状態で、システム全体をテストできるようになりました。

予想される出力

$ 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

色分けされた出力

クライアントは、異なる色を持つさまざまなコンテンツ タイプを表示します。

• 黄色: 開始された通知を実行する
• シアン: エージェントのテキスト応答 (リアルタイムでストリーミング)
• 緑: 実行完了通知
• 赤: エラー メッセージ

curl を使用したテスト (省略可能)

クライアントを実行する前に、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?"}
    ]
  }'

Server-Sent イベントがストリーミングバックされていることがわかります。

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

アイドル状態のストリームの場合、curl は : keepalive コメント行を表示することもあります。 これらは SSE トランスポート コメントであり、AG-UI イベントではありません。

しくみ

サーバーサイドフロー

1. クライアントがメッセージを含む HTTP POST 要求を送信する
2. FastAPI エンドポイントが要求を受信する
3. AgentFrameworkAgent ラッパーが実行を調整する
4. エージェントは、エージェント フレームワークを使用してメッセージを処理します
5. AgentFrameworkEventBridge エージェントの更新を AG-UI イベントに変換する
6. 応答は、Server-Sent イベント (SSE) としてストリーミングバックされます
7. 実行が完了すると接続が閉じる

クライアントサイドフロー

1. クライアントが HTTP POST 要求をサーバー エンドポイントに送信する
2. サーバーが SSE ストリームで応答する
3. クライアントが受信 data: 行を JSON イベントとして解析する
4. 各イベントは、その種類に基づいて表示されます
5. threadId は、会話の継続性のためにキャプチャされます
6. RUN_FINISHED イベントが到着するとストリームが完了する

プロトコルの詳細

AG-UI プロトコルでは、次のものが使用されます。

• 要求を送信するための HTTP POST
• ストリーミング応答用の Server-Sent イベント (SSE)
• イベントのシリアル化用の JSON
• 会話コンテキストを維持するためのスレッド ID
• 個々の実行を追跡するための ID の実行
• イベントの種類の名前付け: アンダースコア付きの大文字 (例: RUN_STARTED、 TEXT_MESSAGE_CONTENT)
• フィールドの名前付け: camelCase (例: threadId、 runId、 messageId)
• ストリームがアイドル状態の間、SSE のキープアライブコメントは 15 秒ごとに送信されます。 data:行のみを処理するクライアントは、これらのコメントを自動的に無視します。

一般的なパターン

カスタム サーバー構成

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 は正の数または Noneである必要があります。

複数のエージェント

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

エラー処理

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

接続が拒否されました

クライアントを起動する前に、サーバーが実行されていることを確認します。

# Terminal 1
python server.py

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

認証エラー

Azure で認証されていることを確認します。

az login

Azure OpenAI リソースに対する正しいロールの割り当てがあることを確認します。

ストリーミングが機能しない

クライアントのタイムアウトで十分であることを確認します。

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

実行時間の長いエージェントの場合は、それに応じてタイムアウトを増やします。

アイドル ストリームは、既定で 15 秒ごとに SSE キープアライブ コメントを出力します。 プロキシがアイドル状態の接続を早く閉じる場合は、エンドポイントを登録するときに、より小さい正の keepalive_seconds 値を構成します。

スレッド コンテキストが失われた

クライアントは、スレッドの継続性を自動的に管理します。 コンテキストが失われた場合:

1. threadIdがイベントからキャプチャされていることを確認RUN_STARTED
2. メッセージ間で同じクライアント インスタンスが使用されていることを確認する
3. サーバーが後続の要求で thread_id を受信することを確認する

次のステップ

AG-UI の基本を理解したら、次のことができます。

• バックエンド ツールの追加: ドメインのカスタム関数ツールを作成する

その他のリソース

• AG-UI の概要
• Agent Framework のドキュメント
• AG-UI プロトコルの仕様

Go では、サーバーとクライアントの両方の provider/aguiprovider を介した AG-UI がサポートされます。

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

Go アプリで AG-UI サーバーをエージェントとして呼び出す必要がある場合は、 aguiprovider.NewAgent を使用します。

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