使用 AG-UI 進行前端工具渲染

本教學課程說明如何將前端函式工具新增至 AG-UI 用戶端。前端工具是在客戶端執行的功能，允許 AI 代理與使用者的本地環境互動、存取特定於客戶端的資料或執行 UI 操作。伺服器會協調何時呼叫這些工具，但執行完全發生在用戶端上。

先決條件

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

.NET 8.0 或更新版本
Microsoft.Agents.AI.AGUI 已安裝套件
Microsoft.Agents.AI 已安裝套件
對 AG-UI 客戶端設定的基本了解

什麼是前端工具？

前端工具是功能工具，具有以下功能：

在用戶端上定義並註冊
在用戶端環境中執行（而不是在伺服器上）
允許 AI 代理與客戶特定資源互動
將結果提供回伺服器，讓代理程式合併到回應中
實現個人化、情境感知體驗

常見用例：

讀取本地感測器資料（GPS、溫度等）
存取用戶端儲存或偏好設定
執行 UI 操作（變更主題、顯示通知）
與裝置特定功能（相機、麥克風）互動

在用戶端上註冊前端工具

與快速入門教學課程相比，主要差異在於將工具註冊到用戶端代理程式。以下是變化：

// Define a frontend function tool
[Description("Get the user's current location from GPS.")]
static string GetUserLocation()
{
    // Access client-side GPS
    return "Amsterdam, Netherlands (52.37°N, 4.90°E)";
}

// Create frontend tools
AITool[] frontendTools = [AIFunctionFactory.Create(GetUserLocation)];

// Pass tools when creating the agent
AIAgent agent = chatClient.AsAIAgent(
    name: "agui-client",
    description: "AG-UI Client Agent",
    tools: frontendTools);

客戶端程式碼的其餘部分將與快速入門指南中顯示的相同。

如何將工具傳送至伺服器

當您向 AsAIAgent() 註冊工具時，AGUIChatClient 會自動執行以下操作：

擷取工具定義（名稱、描述、參數結構描述）
將工具隨著每個請求一同傳送至伺服器代理程式，該代理程式會將它們對應至 ChatAgentRunOptions.ChatOptions.Tools

伺服器接收用戶端工具宣告，AI 模型可以決定何時呼叫它們。

使用中介軟體檢查和修改工具

您可以使用代理程式中介軟體來檢查或修改代理程式執行，包括存取工具：

// Create agent with middleware that inspects tools
AIAgent inspectableAgent = baseAgent
    .AsBuilder()
    .Use(runFunc: null, runStreamingFunc: InspectToolsMiddleware)
    .Build();

static async IAsyncEnumerable<AgentResponseUpdate> InspectToolsMiddleware(
    IEnumerable<ChatMessage> messages,
    AgentSession? session,
    AgentRunOptions? options,
    AIAgent innerAgent,
    CancellationToken cancellationToken)
{
    // Access the tools from ChatClientAgentRunOptions
    if (options is ChatClientAgentRunOptions chatOptions)
    {
        IList<AITool>? tools = chatOptions.ChatOptions?.Tools;
        if (tools != null)
        {
            Console.WriteLine($"Tools available for this run: {tools.Count}");
            foreach (AITool tool in tools)
            {
                if (tool is AIFunction function)
                {
                    Console.WriteLine($"  - {function.Metadata.Name}: {function.Metadata.Description}");
                }
            }
        }
    }

    await foreach (AgentResponseUpdate update in innerAgent.RunStreamingAsync(messages, session, options, cancellationToken))
    {
        yield return update;
    }
}

此中介軟體模式可讓您：

在執行之前驗證工具定義

重要概念

以下是前端工具的新概念：

用戶端註冊：工具會在用戶端上註冊，然後透過 AIFunctionFactory.Create() 傳遞給 AsAIAgent()
自動捕獲：工具會被自動捕獲，並通過ChatAgentRunOptions.ChatOptions.Tools

前端工具的工作原理

伺服器端流程

伺服器不知道前端工具的實作細節。它只知道：

工具名稱和描述（來自客戶註冊）
參數結構描述
何時要求工具執行

當 AI 代理決定呼叫前端工具時：

伺服器透過 SSE 向用戶端傳送工具呼叫請求
伺服器等待客戶端執行工具並傳回結果
伺服器會將結果併入代理程式的內容中
代理程序繼續使用工具結果進行處理

用戶端流程

客戶端處理前端工具執行：

從伺服器接收 FunctionCallContent 指示工具呼叫請求
將工具名稱與本機註冊的函數進行比對
從請求中反序列化參數
在本地執行功能
序列化結果
傳 FunctionResultContent 回伺服器
繼續接收客服專員回應

前端工具的預期輸出

當代理呼叫前端工具時，您會看到工具呼叫並產生串流輸出：

User (:q or quit to exit): Where am I located?

[Client Tool Call - Name: GetUserLocation]
[Client Tool Result: Amsterdam, Netherlands (52.37°N, 4.90°E)]

You are currently in Amsterdam, Netherlands, at coordinates 52.37°N, 4.90°E.

前端工具的伺服器設定

伺服器不需要特殊配置即可支援前端工具。使用快速入門教學課程中的標準 AG-UI 伺服器 — 它會自動：

在客戶端連接期間接收前端工具聲明
當 AI 代理需要時請求工具的執行
等待用戶端的結果
將結果納入客服專員的決策中

後續步驟

現在您已經了解了前端工具，您可以：

與後端工具結合：同時使用前端和後端工具

其他資源

本教學課程說明如何將前端函式工具新增至 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)

                            # Handle tool call requests from server
                            if event.get("type") == "TOOL_CALL_REQUEST":
                                await self._handle_tool_call(event, client)
                            else:
                                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")

            # Send result back to server
            await client.post(
                f"{self.server_url}/tool_result",
                json={
                    "tool_call_id": tool_call_id,
                    "result": result,
                },
            )

        except Exception as e:
            print(f"\033[91m[Tool Error: {e}]\033[0m")
            # Send error back to server
            await client.post(
                f"{self.server_url}/tool_result",
                json={
                    "tool_call_id": tool_call_id,
                    "error": str(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 代理根據用戶請求決定何時調用前端工具
工具呼叫請求：伺服器透過 SSE 將事件傳送 TOOL_CALL_REQUEST 給用戶端
客戶端執行：客戶端在本機執行工具
結果提交：客戶端通過 POST 請求將結果發送回服務器
代理程式處理：伺服器合併結果並繼續回應

關鍵事件

TOOL_CALL_REQUEST：伺服器要求前端工具執行
TOOL_CALL_RESULT：客戶端提交執行結果（透過HTTP POST）

預期輸出

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"

故障排除

未呼叫工具

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

執行錯誤

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

類型問題

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

後續步驟

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

其他資源

意見反應

此頁面對您有幫助嗎？

Last updated on 2025-11-11

共用方式為

使用 AG-UI 進行前端工具渲染

先決條件

什麼是前端工具？

在用戶端上註冊前端工具

如何將工具傳送至伺服器

使用中介軟體檢查和修改工具

重要概念

前端工具的工作原理

伺服器端流程

用戶端流程

前端工具的預期輸出

前端工具的伺服器設定

後續步驟

其他資源

先決條件

什麼是前端工具？

創建前端工具

使用前端工具建立 AG-UI 客戶端

前端工具的工作原理

通訊協定流程

關鍵事件

預期輸出

伺服器設定

最佳做法

安全性

錯誤處理

異步操作

故障排除

未呼叫工具

執行錯誤

類型問題

後續步驟

其他資源

意見反應

其他資源