프런트 엔드 도구는 AG-UI 클라이언트에 의해 선언되고 실행됩니다. 서버는 해당 스키마를 수신하므로 모델에서 스키마를 요청할 수 있지만 해당 구현은 수신하지 않습니다.
프런트 엔드 도구 등록
도구를 만들고 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 는 연속 흐름을 처리합니다.
- 실행 요청을 사용하여 프런트 엔드 도구 선언을 보냅니다.
- 서버에서 모델의 도구 호출을 받습니다.
- 일치하는 함수를 로컬로 실행합니다.
- 결과를 서버로 다시 보냅니다.
- 실행을 계속하고 최종 응답을 스트리밍합니다.
Tip
전체 클라이언트 및 서버에 대한 .NET 프런트 엔드 도구 샘플을 참조하세요.
Warning
신뢰할 수 없는 클라이언트에서 제공하는 도구 선언 및 결과는 신뢰할 수 없는 입력입니다. 서버 쪽 에이전트 실행에 영향을 줄 수 있는 클라이언트 도구에 권한을 부여하고 권한 있는 작업에 사용하기 전에 결과의 유효성을 검사합니다.
일반적인 도구 작성 지침은 에이전트와 함께 함수 도구 사용을 참조하세요.
다음 단계
이 자습서에서는 AG-UI 클라이언트에 프런트 엔드 함수 도구를 추가하는 방법을 보여 줍니다. 프런트 엔드 도구는 클라이언트 쪽에서 실행되는 함수로, AI 에이전트가 사용자의 로컬 환경과 상호 작용하거나 클라이언트별 데이터에 액세스하거나 UI 작업을 수행할 수 있도록 합니다.
사전 요구 사항
시작하기 전에 시작 자습서를 완료하고 다음을 수행했는지 확인합니다.
- Python 3.10 이상
-
httpxHTTP 클라이언트 기능을 위해 설치됨 - AG-UI 클라이언트 설정에 대한 기본 이해
- 구성된 Azure OpenAI 서비스
프런트 엔드 도구란?
프런트 엔드 도구는 다음과 같은 함수 도구입니다.
- 클라이언트에 정의 및 등록됨
- 서버가 아닌 클라이언트의 환경에서 실행
- AI 에이전트가 클라이언트별 리소스와 상호 작용하도록 허용
- 에이전트가 응답에 통합할 수 있도록 결과를 서버에 다시 제공합니다.
일반적인 사용 사례:
- 로컬 센서 데이터 읽기
- 클라이언트 쪽 스토리지 또는 기본 설정에 액세스
- UI 작업 수행
- 디바이스별 기능과 상호 작용
프런트 엔드 도구 만들기
Python의 프런트 엔드 도구는 백 엔드 도구와 유사하게 정의되지만 클라이언트에 등록됩니다.
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}"
프런트 엔드 도구를 사용하여 AG-UI 클라이언트 만들기
프런트 엔드 도구를 사용하는 전체 클라이언트 구현은 다음과 같습니다.
"""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())
프런트 엔드 도구의 작동 방식
프로토콜 흐름
- 클라이언트 등록: 클라이언트가 도구 선언(이름, 설명, 매개 변수)을 서버에 보냅니다.
- 서버 오케스트레이션: AI 에이전트가 사용자 요청에 따라 프런트 엔드 도구를 호출할 시기를 결정합니다.
-
도구 호출 이벤트: 서버는
TOOL_CALL_START,TOOL_CALL_ARGS,TOOL_CALL_END이벤트를 클라이언트로 스트리밍합니다 - 클라이언트 실행: 클라이언트가 로컬로 도구를 실행합니다.
-
결과 이벤트: 도구 결과가 스트림의 이벤트로
TOOL_CALL_RESULT표시됩니다. - 에이전트 처리: 서버는 결과를 통합하고 응답을 계속합니다.
주요 이벤트
-
TOOL_CALL_START/TOOL_CALL_ARGS/TOOL_CALL_END: 서버 요청 및 스트림 도구 호출 세부 정보 -
TOOL_CALL_RESULT: 도구 실행 결과 이벤트
예상 출력
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]
서버 설정
시작 자습서의 표준 AG-UI 서버는 프런트 엔드 도구를 자동으로 지원합니다. 서버 쪽에서 변경이 필요하지 않습니다. 도구 오케스트레이션을 자동으로 처리합니다.
모범 사례
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"
오류 처리
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)}"
비동기 작업
async def capture_photo() -> str:
"""Capture a photo from device camera."""
# Simulate camera access
await asyncio.sleep(1)
return "photo_12345.jpg"
Troubleshooting
도구가 호출되지 않음
- 도구 선언이 서버로 전송되는지 확인
- 도구 설명이 용도를 명확하게 나타내는지 확인
- 서버 로그에서 도구 등록 확인
실행 오류
- 포괄적인 오류 처리 추가
- 처리하기 전에 매개 변수 유효성 검사
- 사용자에게 친숙한 오류 메시지 반환
- 디버깅에 대한 로그 오류
타입 문제
- 복합 형식에 대해 Pydantic 모델 사용
- serialization 전에 모델을 딕셔너리로 변환
- 형식 변환을 명시적으로 처리
다음 단계
- 백 엔드 도구 렌더링: 서버 쪽 도구와 결합
추가 리소스
go AG-UI 서버는 호스트된 에이전트에서 자동 함수 호출을 사용하지 않도록 설정하여 프런트 엔드에 대한 도구 호출을 남길 수 있습니다.
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
전체 실행 가능한 예제는 AG-UI 프런트 엔드 도구 샘플을 참조하세요.