使用 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 處理後續流程：

1. 會傳送前端工具聲明並附上執行請求。
2. 接收來自伺服器的模型工具呼叫。
3. 在本地執行匹配函式。
4. 會把結果回傳給伺服器。
5. 繼續執行並串流傳輸最終回應。

關於一般工具編寫指引，請參見「 使用函式工具搭配代理」

下一步

本教學課程說明如何將前端函式工具新增至 AG-UI 用戶端。 前端工具是在客戶端執行的功能，允許 AI 代理與使用者的本地環境互動、存取特定於客戶端的資料或執行 UI 操作。

Prerequisites

開始之前，請確定您已完成快速 入門 教學課程，並具備：

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

前端工具的工作原理

通訊協定流程

1. 客戶端註冊：客戶端將工具聲明（名稱、描述、參數）發送到服務器
2. 服務器編排： AI 代理根據用戶請求決定何時調用前端工具
3. 工具呼叫事件：伺服器向用戶端串流傳送 TOOL_CALL_START、TOOL_CALL_ARGS 和 TOOL_CALL_END 事件
4. 客戶端執行：客戶端在本機執行工具
5. 結果事件：工具結果以事件形式呈現 TOOL_CALL_RESULT 在串流中
6. 代理程式處理：伺服器合併結果並繼續回應

關鍵事件

• 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

未呼叫工具

1. 確保工具宣告已傳送至伺服器
2. 驗證工具描述是否清楚指出用途
3. 檢查伺服器日誌以確認工具註冊情況

執行錯誤

1. 新增全面的錯誤處理
2. 在處理之前驗證參數
3. 傳回使用者友善的錯誤訊息
4. 紀錄錯誤以進行偵錯

類型問題

1. 將 Pydantic 模型用於複雜類型
2. 在序列化之前將模型轉換為字典格式
3. 明確處理類型轉換

後續步驟

• 後端工具渲染：與伺服器端工具結合

其他資源

• AG-UI 概覽
• 入門指南
• 代理程式架構文件

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