Perenderan Alat Frontend dengan AG-UI

Alat frontend dideklarasikan dan dijalankan oleh klien AG-UI. Server menerima skema mereka sehingga model dapat memintanya, tetapi tidak menerima implementasinya.

Mendaftarkan alat frontend

Buat alat dan teruskan ke agen yang didukung oleh 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 menangani alur kelanjutan:

Mengirim deklarasi alat frontend dengan permintaan eksekusi.
Menerima panggilan alat model dari server.
Menjalankan fungsi pencocokan secara lokal.
Mengirim hasilnya kembali ke server.
Melanjutkan eksekusi dan mengalirkan respons akhir.

Tip

Lihat sampel .NET alat frontend untuk klien dan server lengkap.

Warning

Deklarasi alat dan hasil yang disediakan oleh klien yang tidak tepercaya adalah input yang tidak tepercaya. Otorisasi alat klien mana yang dapat memengaruhi eksekusi agen sisi server, dan memvalidasi hasil sebelum menggunakannya untuk operasi istimewa.

Untuk panduan penulisan alat umum, lihat Menggunakan alat fungsi dengan agen.

Langkah berikutnya

Menggunakan persetujuan manusia dengan AG-UI

Tutorial ini menunjukkan kepada Anda cara menambahkan alat fungsi frontend ke klien AG-UI Anda. Alat frontend adalah fungsi yang dijalankan di sisi klien, memungkinkan agen AI berinteraksi dengan lingkungan lokal pengguna, mengakses data khusus klien, atau melakukan operasi UI.

Prasyarat

Sebelum memulai, pastikan Anda telah menyelesaikan tutorial Memulai dan memiliki:

Python 3.10 atau yang lebih baru
httpx diinstal untuk fungsionalitas klien HTTP
Pemahaman dasar tentang penyiapan klien AG-UI
Layanan Azure OpenAI dikonfigurasi

Apa itu Alat Frontend?

Alat frontend adalah alat fungsi yang:

Didefinisikan dan didaftarkan pada klien
Jalankan di lingkungan klien (bukan di server)
Mengizinkan agen AI berinteraksi dengan sumber daya khusus klien
Kirim kembali hasil ke server agar agen dapat mengintegrasikan ke dalam respons.

Kasus penggunaan umum:

Membaca data sensor lokal
Mengakses penyimpanan atau preferensi sisi klien
Melakukan operasi UI
Berinteraksi dengan fitur khusus perangkat

Membuat Alat Frontend

Alat frontend di Python didefinisikan mirip dengan alat backend tetapi terdaftar di klien:

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

Membuat Klien AG-UI dengan Alat Frontend

Berikut adalah implementasi klien lengkap dengan alat frontend:

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

Cara Kerja Alat Frontend

Sistem Aliran Protokol

Pendaftaran Klien: Klien mengirim deklarasi alat (nama, deskripsi, parameter) ke server
Orkestrasi Server: Agen AI memutuskan kapan harus memanggil alat frontend berdasarkan permintaan pengguna
Peristiwa Pemanggilan Alat: Server mengalirkan event TOOL_CALL_START, TOOL_CALL_ARGS, dan TOOL_CALL_END ke klien
Eksekusi Klien: Klien menjalankan alat secara lokal
Peristiwa Hasil: Hasil alat direpresentasikan sebagai TOOL_CALL_RESULT peristiwa dalam aliran
Pemrosesan Agen: Server menggabungkan hasil dan melanjutkan respons

Peristiwa Utama

TOOL_CALL_START / TOOL_CALL_ARGS / TOOL_CALL_END: Permintaan server dan detail panggilan alat streaming
TOOL_CALL_RESULT: Peristiwa hasil eksekusi alat

Output yang Diharapkan

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]

Penyetelan Server

Server AG-UI standar dari tutorial Memulai secara otomatis mendukung alat frontend. Tidak ada perubahan yang diperlukan di sisi server - ia menangani orkestrasi alat secara otomatis.

Praktik Terbaik

Keamanan

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"

Penanganan Kesalahan

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

Operasi Asinkron

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

Troubleshooting

Alat Tidak Dipanggil

Pastikan bahwa deklarasi alat dikirim ke server
Memastikan deskripsi alat menunjukkan tujuan dengan jelas
Periksa log server untuk pendaftaran alat

Kesalahan Eksekusi

Menambahkan penanganan kesalahan komprehensif
Memvalidasi parameter sebelum memproses
Mengembalikan pesan kesalahan yang mudah digunakan
Kesalahan pencatatan untuk debugging

Masalah Tipe

Menggunakan model Pydantic untuk jenis kompleks
Mengonversi model menjadi dict sebelum serialisasi
Menangani konversi tipe secara eksplisit

Langkah Selanjutnya

Penyajian Alat Backend: Gabungkan dengan alat sisi server

Sumber Daya Tambahan

Go AG-UI server dapat meninggalkan panggilan alat untuk frontend dengan menonaktifkan panggilan fungsi otomatis pada agen yang dihosting.

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

Lihat sampel alat frontendAG-UI untuk contoh lengkap yang dapat dijalankan.

Saran dan Komentar

Apakah halaman ini membantu?

Last updated on 2026-08-14