Frontend-Tool-Rendering mit AG-UI

Frontend-Tools werden vom AG-UI-Client deklariert und ausgeführt. Der Server empfängt seine Schemas, damit das Modell sie anfordern kann, aber er empfängt seine Implementierungen nicht.

Registrieren eines Frontend-Tools

Erstellen Sie das Tool und übergeben Sie es an den Agenten, der von AGUIChatClient unterstützt wird:

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 behandelt den Fortsetzungsfluss:

1. Sendet die Front-End-Tool-Deklaration mit der Ausführungsanforderung.
2. Empfängt den Toolaufruf des Modells vom Server.
3. Führt die übereinstimmende Funktion lokal aus.
4. Sendet das Ergebnis zurück an den Server.
5. Setzt die Ausführung fort und streamt die endgültige Antwort.

Allgemeine Anleitungen zur Toolerstellung finden Sie unter Verwenden von Funktionstools mit einem Agent.

Nächste Schritte

In diesem Lernprogramm erfahren Sie, wie Sie Ihren AG-UI Clients Frontend-Funktionstools hinzufügen. Frontend-Tools sind Funktionen, die auf clientseitiger Seite ausgeführt werden, sodass der KI-Agent mit der lokalen Umgebung des Benutzers interagieren kann, auf clientspezifische Daten zugreifen oder UI-Vorgänge ausführt.

Voraussetzungen

Bevor Sie beginnen, stellen Sie sicher, dass Sie das Lernprogramm " Erste Schritte " abgeschlossen haben und folgendes haben:

• Python 3.10 oder höher
• httpx installiert für HTTP-Clientfunktionen
• Grundlegendes Verständnis des AG-UI Clientsetups
• Azure OpenAI-Dienst konfiguriert

Was sind Frontend-Tools?

Frontend-Tools sind Funktionstools, die:

• Werden definiert und auf dem Client registriert
• Ausführen in der Umgebung des Clients (nicht auf dem Server)
• Zulassen, dass der KI-Agent mit clientspezifischen Ressourcen interagiert
• Übermittlung von Ergebnissen zurück an den Server, damit der Agent die Ergebnisse in Antworten integrieren kann

Häufige Anwendungsfälle:

• Lesen lokaler Sensordaten
• Zugreifen auf clientseitigen Speicher oder Einstellungen
• Ausführen von UI-Vorgängen
• Interaktion mit gerätespezifischen Features

Erstellen von Frontend-Tools

Frontend-Tools in Python werden ähnlich wie Back-End-Tools definiert, aber beim Client registriert:

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

Erstellen eines AG-UI Clients mit Frontend-Tools

Hier ist eine vollständige Clientimplementierung mit Frontend-Tools:

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

Funktionsweise von Front-End-Tools

Protokollfluss

1. Clientregistrierung: Der Client sendet Tooldeklarationen (Namen, Beschreibungen, Parameter) an den Server.
2. Server-Orchestrierung: KI-Agent entscheidet, wann Frontend-Tools basierend auf Benutzeranforderung aufgerufen werden sollen.
3. Tool-Aufrufereignisse: Der Server streamt TOOL_CALL_START, TOOL_CALL_ARGS und TOOL_CALL_END-Ereignisse an den Client
4. Clientausführung: Client führt das Tool lokal aus
5. Ergebnisereignisse: Toolergebnisse werden als TOOL_CALL_RESULT Ereignisse im Datenstrom dargestellt.
6. Agent-Verarbeitung: Der Server verarbeitet Ergebnisse und setzt die Antwort fort.

Wichtige Ereignisse

• TOOL_CALL_START / TOOL_CALL_ARGS / TOOL_CALL_END: Der Server fordert Details zu Tool-Aufrufen an und streamt sie
• TOOL_CALL_RESULT: Ergebnisereignis der Toolausführung

Erwartete Ausgabe

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]

Server-Einrichtung

Der Standardserver AG-UI server aus dem Lernprogramm "Erste Schritte" unterstützt automatisch Frontend-Tools. Auf der Serverseite sind keine Änderungen erforderlich – sie verarbeitet die Tool-Orchestrierung automatisch.

Bewährte Methoden

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"

Fehlerbehandlung

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

Asynchrone Vorgänge

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

Troubleshooting

Nicht aufgerufene Tools

1. Sicherstellen, dass Tooldeklarationen an den Server gesendet werden
2. Überprüfen von Toolbeschreibungen, die eindeutig den Zweck angeben
3. Überprüfen von Serverprotokollen für die Toolregistrierung

Ausführungsfehler

1. Umfassende Fehlerbehandlung hinzufügen
2. Überprüfen von Parametern vor der Verarbeitung
3. Zurückgeben von benutzerfreundlichen Fehlermeldungen
4. Protokollfehler für das Debuggen

Typprobleme

1. Verwenden von Pydantischen Modellen für komplexe Typen
2. Konvertieren von Modellen in Diktaten vor der Serialisierung
3. Behandeln Sie Typkonvertierungen explizit.

Nächste Schritte

• Back-End-Toolrendering: Kombinieren mit serverseitigen Tools

Zusätzliche Ressourcen

• AG-UI Übersicht
• Einführungstutorial
• Agent Framework-Dokumentation

Go-AG-UI-Server können Toolaufrufe dem Frontend überlassen, indem das automatische Aufrufen von Funktionen für den gehosteten Agenten deaktiviert wird.

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