使用 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"

Troubleshooting

未调用的工具

确保将工具声明发送到服务器
验证工具说明明确指示用途
检查服务器日志以获取工具注册

执行错误

添加全面的错误处理
在处理之前验证参数
返回用户友好的错误消息
记录错误以进行调试

类型问题

使用 Pydantic 模型来处理复杂类型
在序列化之前将模型转换为字典
显式处理类型转换

后续步骤

后端工具渲染：与服务器端工具组合使用

其他资源

反馈

此页面是否有帮助？

Last updated on 2025-11-11

通过

使用 AG-UI 呈现前端工具

先决条件

什么是前端工具？

在客户端上注册前端工具

如何将工具发送到服务器

使用中间件检查和修改工具

关键概念

前端工具的工作原理

服务器端流程

客户端流

前端工具的预期输出

前端工具的服务器设置

后续步骤

其他资源

先决条件

什么是前端工具？

创建前端工具

使用前端工具创建 AG-UI 客户端

前端工具的工作原理

协议流

关键事件

预期输出

服务器设置

最佳做法

安全性

错误处理

异步操作

Troubleshooting

未调用的工具

执行错误

类型问题

后续步骤

其他资源

反馈

其他资源