AG-UI を使用したフロントエンドツールのレンダリング

フロントエンドツールは、AG-UI クライアントによって宣言および実行されます。サーバーはスキーマを受け取るので、モデルはスキーマを要求できますが、実装は受け取りません。

フロントエンドツールを登録する

ツールを作成し、 AGUIChatClientによってサポートされているエージェントに渡します。

using System.ComponentModel;
using AGUI.Client;
using Microsoft.Agents.AI;
using Microsoft.Extensions.AI;

[Description("Get the user's current location from the client device.")]
static string GetUserLocation() => "Amsterdam, Netherlands";

AITool locationTool = AIFunctionFactory.Create(
    GetUserLocation,
    name: "get_user_location");

using HttpClient httpClient = new() { BaseAddress = new Uri("http://localhost:8888") };
AGUIChatClient chatClient = new(new AGUIChatClientOptions(httpClient, "/"));
AIAgent agent = chatClient.AsAIAgent(tools: [locationTool]);

AGUIChatClient は継続フローを処理します。

実行要求を使用してフロントエンドツールの宣言を送信します。
サーバーからモデルのツール呼び出しを受け取ります。
一致する関数をローカルで実行します。
結果をサーバーに返します。
実行を続行し、最終的な応答をストリーム配信します。

Tip

完全なクライアントとサーバーについては、.NETフロントエンドツールのサンプルを参照してください。

Warning

信頼されていないクライアントによって提供されるツールの宣言と結果は、信頼されていない入力です。サーバー側エージェントの実行に影響を与える可能性のあるクライアントツールを承認し、特権操作に使用する前に結果を検証します。

ツール作成の一般的なガイダンスについては、エージェントでの関数ツールの使用に関するページを参照してください。

次のステップ

AG-UI で人間の承認を使用する

このチュートリアルでは、フロントエンド関数ツールを AG-UI クライアントに追加する方法について説明します。フロントエンドツールは、クライアント側で実行される関数であり、AI エージェントがユーザーのローカル環境と対話したり、クライアント固有のデータにアクセスしたり、UI 操作を実行したりできます。

前提条件

開始する前に、作業の開始に関するチュートリアルを完了し、次のことを行っていることを確認します。

Python 3.10 以降
httpx HTTP クライアント機能用にインストール済み
AG-UI クライアントセットアップの基本的な理解
構成された Azure OpenAI サービス

フロントエンドツールとは

フロントエンドツールは、次の機能ツールです。

クライアントで定義および登録されている
(サーバーではなく) クライアントの環境で実行する
AI エージェントがクライアント固有のリソースと対話できるようにする
応答にエージェントを組み込むための結果をサーバーに返す

一般的なユースケース:

ローカルセンサーデータの読み取り
クライアント側のストレージまたはユーザー設定へのアクセス
UI 操作の実行
デバイス固有の機能とのインタラクション

フロントエンドツールの作成

Python のフロントエンドツールはバックエンドツールと同様に定義されますが、クライアントに登録されます。

from typing import Annotated
from pydantic import BaseModel, Field


class SensorReading(BaseModel):
    """Sensor reading from client device."""
    temperature: float
    humidity: float
    air_quality_index: int


def read_climate_sensors(
    include_temperature: Annotated[bool, Field(description="Include temperature reading")] = True,
    include_humidity: Annotated[bool, Field(description="Include humidity reading")] = True,
) -> SensorReading:
    """Read climate sensor data from the client device."""
    # Simulate reading from local sensors
    return SensorReading(
        temperature=22.5 if include_temperature else 0.0,
        humidity=45.0 if include_humidity else 0.0,
        air_quality_index=75,
    )


def change_background_color(color: Annotated[str, Field(description="Color name")] = "blue") -> str:
    """Change the console background color."""
    # Simulate UI change
    print(f"\n🎨 Background color changed to {color}")
    return f"Background changed to {color}"

フロントエンドツールを使用した AG-UI クライアントの作成

フロントエンドツールを使用した完全なクライアント実装を次に示します。

"""AG-UI client with frontend tools."""

import asyncio
import json
import os
from typing import Annotated, AsyncIterator

import httpx
from pydantic import BaseModel, Field


class SensorReading(BaseModel):
    """Sensor reading from client device."""
    temperature: float
    humidity: float
    air_quality_index: int


# Define frontend tools
def read_climate_sensors(
    include_temperature: Annotated[bool, Field(description="Include temperature")] = True,
    include_humidity: Annotated[bool, Field(description="Include humidity")] = True,
) -> SensorReading:
    """Read climate sensor data from the client device."""
    return SensorReading(
        temperature=22.5 if include_temperature else 0.0,
        humidity=45.0 if include_humidity else 0.0,
        air_quality_index=75,
    )


def get_user_location() -> dict:
    """Get the user's current GPS location."""
    # Simulate GPS reading
    return {
        "latitude": 52.3676,
        "longitude": 4.9041,
        "accuracy": 10.0,
        "city": "Amsterdam",
    }


# Tool registry maps tool names to functions
FRONTEND_TOOLS = {
    "read_climate_sensors": read_climate_sensors,
    "get_user_location": get_user_location,
}


class AGUIClientWithTools:
    """AG-UI client with frontend tool support."""

    def __init__(self, server_url: str, tools: dict):
        self.server_url = server_url
        self.tools = tools
        self.thread_id: str | None = None

    async def send_message(self, message: str) -> AsyncIterator[dict]:
        """Send a message and handle streaming response with tool execution."""
        # Prepare tool declarations for the server
        tool_declarations = []
        for name, func in self.tools.items():
            tool_declarations.append({
                "name": name,
                "description": func.__doc__ or "",
                # Add parameter schema from function signature
            })

        request_data = {
            "messages": [
                {"role": "system", "content": "You are a helpful assistant with access to client tools."},
                {"role": "user", "content": message},
            ],
            "tools": tool_declarations,  # Send tool declarations to server
        }

        if self.thread_id:
            request_data["thread_id"] = self.thread_id

        async with httpx.AsyncClient(timeout=60.0) as client:
            async with client.stream(
                "POST",
                self.server_url,
                json=request_data,
                headers={"Accept": "text/event-stream"},
            ) as response:
                response.raise_for_status()

                async for line in response.aiter_lines():
                    if line.startswith("data: "):
                        data = line[6:]
                        try:
                            event = json.loads(data)

                            # Tool calls arrive as TOOL_CALL_START/ARGS/END events
                            # and results are streamed back as TOOL_CALL_RESULT events.
                            yield event

                            # Capture thread_id
                            if event.get("type") == "RUN_STARTED" and not self.thread_id:
                                self.thread_id = event.get("threadId")

                        except json.JSONDecodeError:
                            continue

    async def _handle_tool_call(self, event: dict, client: httpx.AsyncClient):
        """Execute frontend tool and send result back to server."""
        tool_name = event.get("toolName")
        tool_call_id = event.get("toolCallId")
        arguments = event.get("arguments", {})

        print(f"\n\033[95m[Client Tool Call: {tool_name}]\033[0m")
        print(f"  Arguments: {arguments}")

        try:
            # Execute the tool
            tool_func = self.tools.get(tool_name)
            if not tool_func:
                raise ValueError(f"Unknown tool: {tool_name}")

            result = tool_func(**arguments)

            # Convert Pydantic models to dict
            if hasattr(result, "model_dump"):
                result = result.model_dump()

            print(f"\033[94m[Client Tool Result: {result}]\033[0m")

            # In current Python AG-UI, frontend tool declarations are sent with
            # the run request. Tool-call lifecycle events are streamed back over SSE.
            print(f"Tool result for {tool_call_id}: {result}")

        except Exception as e:
            print(f"\033[91m[Tool Error: {e}]\033[0m")
            print(f"Tool error for {tool_call_id}: {e}")


async def main():
    """Main client loop with frontend tools."""
    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")

    client = AGUIClientWithTools(server_url, FRONTEND_TOOLS)

    try:
        while True:
            message = input("\nUser (:q or quit to exit): ")
            if not message.strip():
                continue

            if message.lower() in (":q", "quit"):
                break

            print()
            async for event in client.send_message(message):
                event_type = event.get("type", "")

                if event_type == "RUN_STARTED":
                    print(f"\033[93m[Run Started]\033[0m")

                elif event_type == "TEXT_MESSAGE_CONTENT":
                    print(f"\033[96m{event.get('delta', '')}\033[0m", end="", flush=True)

                elif event_type == "RUN_FINISHED":
                    print(f"\n\033[92m[Run Finished]\033[0m")

                elif event_type == "RUN_ERROR":
                    error_msg = event.get("message", "Unknown error")
                    print(f"\n\033[91m[Error: {error_msg}]\033[0m")

            print()

    except KeyboardInterrupt:
        print("\n\nExiting...")
    except Exception as e:
        print(f"\n\033[91mError: {e}\033[0m")


if __name__ == "__main__":
    asyncio.run(main())

フロントエンドツールのしくみ

プロトコルフロー

クライアント登録: クライアントがツール宣言 (名前、説明、パラメーター) をサーバーに送信する
サーバーオーケストレーション: AI エージェントは、ユーザー要求に基づいてフロントエンドツールを呼び出すタイミングを決定します
ツール呼び出しイベント: サーバーは、 TOOL_CALL_START、 TOOL_CALL_ARGS、および TOOL_CALL_END イベントをクライアントにストリーム配信します
クライアントの実行: クライアントがツールをローカルで実行する
結果イベント: ツールの結果は、ストリーム内の TOOL_CALL_RESULT イベントとして表されます
エージェント処理: サーバーに結果が組み込まれており、応答を続行する

キーイベント

TOOL_CALL_START / TOOL_CALL_ARGS / TOOL_CALL_END: サーバー要求とストリームツール呼び出しの詳細
TOOL_CALL_RESULT: ツールの実行結果イベント

予想される出力

User (:q or quit to exit): What's the temperature reading from my sensors?

[Run Started]

[Client Tool Call: read_climate_sensors]
  Arguments: {'include_temperature': True, 'include_humidity': True}
[Client Tool Result: {'temperature': 22.5, 'humidity': 45.0, 'air_quality_index': 75}]

Based on your sensor readings, the current temperature is 22.5°C and the 
humidity is at 45%. These are comfortable conditions!
[Run Finished]

サーバーのセットアップ

概要チュートリアルの標準 AG-UI サーバーでは、フロントエンドツールが自動的にサポートされます。サーバー側で変更は必要ありません。ツールオーケストレーションは自動的に処理されます。

ベストプラクティス

セキュリティ

def access_sensitive_data() -> str:
    """Access user's sensitive data."""
    # Always check permissions first
    if not has_permission():
        return "Error: Permission denied"

    try:
        # Access data
        return "Data retrieved"
    except Exception as e:
        # Don't expose internal errors
        return "Unable to access data"

エラー処理

def read_file(path: str) -> str:
    """Read a local file."""
    try:
        with open(path, "r") as f:
            return f.read()
    except FileNotFoundError:
        return f"Error: File not found: {path}"
    except PermissionError:
        return f"Error: Permission denied: {path}"
    except Exception as e:
        return f"Error reading file: {str(e)}"

非同期操作

async def capture_photo() -> str:
    """Capture a photo from device camera."""
    # Simulate camera access
    await asyncio.sleep(1)
    return "photo_12345.jpg"

Troubleshooting

ツールが呼び出されない

ツール宣言がサーバーに送信されることを確認する
ツールの説明が目的を明確に示すかどうかを確認する
ツールの登録に関するサーバーログを確認する

実行エラー

包括的なエラー処理を追加する
処理前にパラメーターを検証する
わかりやすいエラーメッセージを返す
デバッグのエラーをログに記録する

タイプの問題

複合型に Pydantic モデルを使用する
シリアル化の前にモデルをディクテーションに変換する
型変換を明示的に処理する

次のステップ

バックエンドツールレンダリング: サーバー側ツールとの組み合わせ

その他のリソース

Go AG-UI サーバーは、ホストされているエージェントで自動関数呼び出しを無効にすることで、フロントエンドのツール呼び出しを残すことができます。

a := foundryprovider.NewAgent(endpoint, token, foundryprovider.ModelDeployment(model), foundryprovider.AgentConfig{
    Instructions: "You are a helpful assistant.",
    Config: agent.Config{
        Name:                "AGUIAssistant",
        DisableFuncAutoCall: true,
    },
})

mux := http.NewServeMux()
mux.Handle("/", aguiprovider.NewJSONHTTPHandler(a, aguiprovider.HandlerConfig{}))

Tip

実行可能な完全な例については、 AG-UI フロントエンドツールのサンプルを参照してください。

フィードバック

このページはお役に立ちましたか?

Last updated on 2026-08-14

AG-UI を使用したフロントエンド ツールのレンダリング

フロントエンド ツールを登録する

次のステップ

前提条件

フロントエンド ツールとは

フロントエンド ツールの作成

フロントエンド ツールを使用した AG-UI クライアントの作成

フロントエンド ツールのしくみ

プロトコル フロー

キー イベント

予想される出力

サーバーのセットアップ

ベスト プラクティス

セキュリティ

エラー処理

非同期操作

Troubleshooting

ツールが呼び出されない

実行エラー

タイプの問題

次のステップ

その他のリソース

フィードバック

その他のリソース

AG-UI を使用したフロントエンドツールのレンダリング

フロントエンドツールを登録する

フロントエンドツールとは

フロントエンドツールの作成

フロントエンドツールを使用した AG-UI クライアントの作成

フロントエンドツールのしくみ

プロトコルフロー

キーイベント

ベストプラクティス