Representación de herramientas de front-end con AG-UI

El cliente de AG-UI declara y ejecuta las herramientas de front-end. El servidor recibe sus esquemas para que el modelo pueda solicitarlos, pero no recibe sus implementaciones.

Registro de una herramienta de front-end

Cree la herramienta y pásela al agente respaldado por 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 gestiona el flujo de continuación:

1. Envía la declaración de la herramienta de front-end con la solicitud de ejecución.
2. Recibe la invocación de herramienta del modelo desde el servidor.
3. Ejecuta la función coincidente localmente.
4. Devuelve el resultado al servidor.
5. Continúa la ejecución y transmite la respuesta final.

Tip

Consulte el ejemplo de herramientas de front-end de .NET para obtener un cliente y un servidor completos.

Warning

Las declaraciones de herramientas y los resultados proporcionados por un cliente no fiable son entradas no fiables. Autorice qué herramientas de cliente pueden influir en la ejecución del agente del lado servidor y valide los resultados antes de usarlos para las operaciones con privilegios.

Para obtener instrucciones generales sobre la creación de herramientas, consulte Uso de herramientas de funciones con un agente.

Pasos siguientes

Uso de la aprobación humana con AG-UI

En este tutorial se muestra cómo agregar herramientas de función de front-end a los clientes de AG-UI. Las herramientas de front-end son funciones que se ejecutan en el lado cliente, lo que permite que el agente de IA interactúe con el entorno local del usuario, acceda a datos específicos del cliente o realice operaciones de interfaz de usuario.

Prerequisites

Antes de comenzar, asegúrese de que ha completado el tutorial de introducción y que tiene:

• Python 3.10 o posterior
• httpx instalado para la funcionalidad del cliente HTTP
• Conocimientos básicos de la configuración del cliente de AG-UI
• Servicio OpenAI de Azure configurado

¿Qué son las herramientas de front-end?

Las herramientas de front-end son herramientas de función que:

• Se definen y registran en el cliente
• Ejecutar en el entorno del cliente (no en el servidor)
• Permitir que el agente de IA interactúe con recursos específicos del cliente
• Proporcionar los resultados al servidor para que el agente los incorpore en las respuestas.

Casos de uso comunes:

• Lectura de datos del sensor local
• Acceso al almacenamiento o preferencias del lado cliente
• Realización de operaciones de interfaz de usuario
• Interacción con características específicas del dispositivo

Creación de herramientas de front-end

Las herramientas de front-end de Python se definen de forma similar a las herramientas de back-end, pero se registran con el cliente:

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

Creación de un cliente de AG-UI con herramientas de front-end

Esta es una implementación de cliente completa con herramientas de front-end:

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

Cómo funcionan las herramientas de front-end

Flujo de protocolo

1. Registro de cliente: el cliente envía declaraciones de herramientas (nombres, descripciones, parámetros) al servidor
2. Orquestación del servidor: el agente de IA decide cuándo llamar a las herramientas de front-end en función de la solicitud de usuario
3. Eventos de llamada a herramientas: El servidor transmite al cliente los eventos TOOL_CALL_START, TOOL_CALL_ARGS y TOOL_CALL_END
4. Ejecución del cliente: el cliente ejecuta la herramienta localmente.
5. Eventos de resultados: los resultados de la herramienta se representan como TOOL_CALL_RESULT eventos en la secuencia
6. Procesamiento del agente: el servidor incorpora el resultado y continúa la respuesta

Eventos clave

• TOOL_CALL_START / TOOL_CALL_ARGS / TOOL_CALL_END: solicitudes del servidor y transmisión de detalles de la llamada de herramienta
• TOOL_CALL_RESULT: evento de resultado de la ejecución de una herramienta

Resultado esperado

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]

Configuración del servidor

El servidor AG-UI estándar del tutorial de introducción admite automáticamente las herramientas de frontend. No se necesitan cambios en el lado servidor: controla automáticamente la orquestación de herramientas.

Prácticas recomendadas

Security

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"

Tratamiento de errores

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

Operaciones asincrónicas

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

Troubleshooting

Herramientas a las que no se llama

1. Asegurarse de que las declaraciones de herramientas se envían al servidor
2. Verificar que las descripciones de la herramienta indiquen claramente el propósito
3. Comprobación de los registros del servidor para el registro de herramientas

Errores de ejecución

1. Incluir una gestión de errores completa.
2. Validación de parámetros antes del procesamiento
3. Devolver mensajes de error fáciles de entender
4. Registro de errores para la depuración

Problemas de tipo

1. Uso de modelos Pydantic para tipos complejos
2. Conversión de modelos en dicts antes de la serialización
3. Maneje las conversiones de tipo explícitamente

Pasos siguientes

• Representación de herramientas de back-end: combinación con herramientas del lado servidor

Recursos adicionales

• Información general deAG-UI
• Tutorial de introducción
• Documentación de Agent Framework

Los servidores Go AG-UI pueden dejar las llamadas a herramientas para el frontend desactivando la invocación automática de funciones en el agente alojado.

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

Consulte el ejemplo de herramientas de front-end deAG-UI para ver un ejemplo completo de ejecución.