Klientdelsverktygsrendering med AG-UI

Frontendverktyg deklareras och körs av AG-UI-klienten. Servern tar emot sina scheman så att modellen kan begära dem, men den tar inte emot deras implementeringar.

Registrera ett frontendverktyg

Skapa verktyget och skicka det till agenten som backas upp av 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 hanterar fortsättningsflödet:

1. Skickar frontend-verktygets deklaration med körningsbegäran.
2. Tar emot modellens verktygsanrop från servern.
3. Kör matchningsfunktionen lokalt.
4. Skickar tillbaka resultatet till servern.
5. Fortsätter körningen och streamar det slutliga svaret.

Allmän vägledning för verktygsredigering finns i Använda funktionsverktyg med en agent.

Nästa steg

Den här handledningen visar hur du lägger till frontend-funktionsverktyg för dina AG-UI-klienter. Klientdelsverktyg är funktioner som körs på klientsidan, vilket gör att AI-agenten kan interagera med användarens lokala miljö, komma åt klientspecifika data eller utföra användargränssnittsåtgärder.

Förutsättningar

Innan du börjar kontrollerar du att du har slutfört handledningen Komma igång och har:

• Python 3.10 eller senare
• httpx installerat för HTTP-klientfunktioner
• Grundläggande förståelse för AG-UI klientkonfiguration
• Azure OpenAI-tjänsten har konfigurerats

Vad är Frontend-verktyg?

Frontend-verktyg är funktionella verktyg som

• Definieras och registreras på klienten
• Kör i klientens miljö (inte på servern)
• Tillåt ATT AI-agenten interagerar med klientspecifika resurser
• Ange resultat tillbaka till servern så att agenten kan införliva i svar

Vanliga användningsfall:

• Läsa lokala sensordata
• Åtkomst till lagring på klientsidan eller inställningar
• Utföra användargränssnittsåtgärder
• Interagera med enhetsspecifika funktioner

Skapa frontendverktyg

Klientdelsverktyg i Python definieras på samma sätt som serverdelsverktyg men registreras med klienten:

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

Skapa en AG-UI-klient med frontend-verktyg

Här är en komplett klientimplementering med frontend-verktyg:

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

Så här fungerar klientdelsverktyg

Protokollflöde

1. Klientregistrering: Klienten skickar verktygsdeklarationer (namn, beskrivningar, parametrar) till servern
2. Serverorkestrering: AI-agenten bestämmer när klientdelsverktyg ska anropas baserat på användarbegäran
3. Verktygsanropshändelser: Servern strömmar TOOL_CALL_START, TOOL_CALL_ARGS och TOOL_CALL_END-händelser till klienten
4. Klientkörning: Klienten kör verktyget lokalt
5. Resultathändelser: Verktygsresultat representeras som TOOL_CALL_RESULT händelser i strömmen
6. Agentbearbetning: Servern inkorporerar resultatet och fortsätter med att svara

Viktiga händelser

• TOOL_CALL_START / TOOL_CALL_ARGS / TOOL_CALL_END: Serverförfrågningar och strömning av information om verktygsanrop
• TOOL_CALL_RESULT: Resultathändelse för verktygsexekvering

Förväntade utdata

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]

Serverkonfiguration

Standardservern AG-UI från kom igång-självstudien stöder automatiskt gränssnittsverktyg. Inga ändringar behövs på serversidan – den hanterar verktygsorkestrering automatiskt.

Metodtips

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"

Felhantering

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

Asynkrona åtgärder

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

Troubleshooting

Verktyg som inte anropas

1. Se till att verktygsdeklarationer skickas till servern
2. Kontrollera verktygsbeskrivningar som tydligt anger syftet
3. Kontrollera om det finns verktygsregistrering i serverloggarna

Körningsfel

1. Lägga till omfattande felhantering
2. Verifiera parametrar före bearbetning
3. Returnera användarvänliga felmeddelanden
4. Loggfel för felsökning

Typproblem

1. Använda pydantiska modeller för komplexa typer
2. Konvertera modeller till ordlistor före serialisering
3. Hantera typkonverteringar explicit.

Nästa steg

• Rendering av serverdelsverktyg: Kombinera med verktyg på serversidan

Ytterligare resurser

• översikt överAG-UI
• Kom igång-guide
• Dokumentation om Agent Framework

Go AG-UI-servrar kan lämna verktygsanrop till frontenden genom att inaktivera automatisk funktionsanropning på den hostade agenten.

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