Frontend-toolweergave met AG-UI

Front-endhulpprogramma's worden gedeclareerd en uitgevoerd door de AG-UI-client. De server ontvangt hun schema's, zodat het model deze kan aanvragen, maar de implementaties niet ontvangen.

Een front-endhulpprogramma registreren

Maak het hulpprogramma en geef het door aan de agent die wordt ondersteund door 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 verwerkt de vervolgstroom:

1. Verzendt de declaratie van de frontend-tool met het uitvoeringsverzoek.
2. Ontvangt de toolaanroep van het model van de server.
3. Hiermee wordt de overeenkomende functie lokaal uitgevoerd.
4. Hiermee wordt het resultaat teruggestuurd naar de server.
5. Zet de uitvoering voort en streamt het uiteindelijke antwoord.

Zie Functiehulpprogramma's gebruiken met een agent voor algemene richtlijnen voor het ontwerpen van hulpprogramma's.

Volgende stappen 

In deze zelfstudie leert u hoe u hulpprogramma's voor front-endfuncties toevoegt aan uw AG-UI-clients. Front-endhulpprogramma's zijn functies die aan de clientzijde worden uitgevoerd, zodat de AI-agent kan communiceren met de lokale omgeving van de gebruiker, toegang heeft tot clientspecifieke gegevens of ui-bewerkingen kan uitvoeren.

Prerequisites

Voordat u begint, moet u ervoor zorgen dat u de zelfstudie Aan de slag hebt voltooid en dat u het volgende hebt gedaan:

• Python 3.10 of hoger
• httpx geïnstalleerd voor HTTP-clientfunctionaliteit
• Basiskennis van het instellen van AG-UI-client
• Azure OpenAI-service geconfigureerd

Wat zijn front-endhulpprogramma's?

Front-endhulpprogramma's zijn functiehulpprogramma's die:

• Zijn gedefinieerd en geregistreerd op de client
• Uitvoeren in de clientomgeving (niet op de server)
• Toestaan dat de AI-agent communiceert met clientspecifieke resources
• Resultaten teruggeven aan de server zodat de agent ze kan opnemen in antwoorden

Veelvoorkomende gebruiksvoorbeelden:

• Lokale sensorgegevens lezen
• Toegang tot opslag of voorkeuren aan de clientzijde
• UI-bewerkingen uitvoeren
• Interactie met apparaatspecifieke functies

Front-endhulpprogramma's maken

Front-endhulpprogramma's in Python worden op dezelfde manier gedefinieerd als back-endhulpprogramma's, maar zijn geregistreerd bij de client:

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

Een AG-UI-client maken met frontendhulpprogramma's

Hier volgt een volledige client-implementatie met front-endhulpprogramma's:

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

Hoe front-endhulpprogramma's werken

Protocolstroom

1. Clientregistratie: Client verzendt hulpprogrammadeclaraties (namen, beschrijvingen, parameters) naar de server
2. Serverindeling: AI-agent bepaalt wanneer front-endhulpprogramma's moeten worden aangeroepen op basis van gebruikersaanvraag
3. Toolaanroepgebeurtenissen: De server streamt TOOL_CALL_START, TOOL_CALL_ARGS en TOOL_CALL_END-gebeurtenissen naar de client
4. Clientuitvoering: Client voert het hulpprogramma lokaal uit
5. Resultaatevenementen: Toolresultaten worden weergegeven als TOOL_CALL_RESULT gebeurtenissen in de stream
6. Agentverwerking: Server verwerkt resultaat en zet de reactie voort

Belangrijke gebeurtenissen

• TOOL_CALL_START / TOOL_CALL_ARGS / TOOL_CALL_END: Serververzoeken en streamt details van hulpprogramma-aanroepen
• TOOL_CALL_RESULT: Resultaatgebeurtenis van de uitvoering van een hulpprogramma

Verwachte uitvoer

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]

Serverinstellingen

De standaardserver AG-UI uit de zelfstudie Aan de slag ondersteunt automatisch front-endhulpprogramma's. Er zijn geen wijzigingen nodig aan de serverzijde. Hiermee wordt de indeling van hulpprogramma's automatisch verwerkt.

Beste praktijken

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"

Foutafhandeling

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 bewerkingen

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

Probleemoplossing

Hulpprogramma's die niet worden aangeroepen

1. Controleren of hulpprogrammadeclaraties naar de server worden verzonden
2. Zorg ervoor dat beschrijvingen van hulpprogramma's duidelijk maken wat het doel ervan is.
3. Serverlogboeken controleren op registratie van hulpprogramma's

Uitvoeringsfouten

1. Uitgebreide foutafhandeling toevoegen
2. Parameters valideren voordat ze worden verwerkt
3. Gebruiksvriendelijke foutberichten retourneren
4. Fouten loggen voor foutopsporing

Typefouten

1. Pydantic-modellen gebruiken voor complexe typen
2. Modellen converteren naar dicteren vóór serialisatie
3. Typeconversies expliciet verwerken

Volgende stappen

• Rendering van back-endhulpprogramma's: combineren met hulpprogramma's aan de serverzijde

Aanvullende bronnen

• overzicht vanAG-UI
• Aan de slag met de handleiding
• Documentatie voor Agent Framework

Go AG-UI-servers kunnen toolaanroepen aan de frontend overlaten door automatische functieaanroepen op de gehoste agent uit te schakelen.

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