AG-UI Integration¶

from llmling_agent.agent import AGUIAgent

async def main():
    # Connect to a remote AG-UI agent
    async with AGUIAgent(
        endpoint="http://localhost:8000/agent/run",
        name="remote-assistant",
        description="External AG-UI agent",
    ) as agent:
        # Use like any other agent
        result = await agent.run("Hello, world!")
        print(result.content)

        # Stream responses
        async for event in agent.run_stream("Tell me a story"):
            print(event)

from llmling_agent import AgentPool

async def main():
    async with AgentPool("agents.yml") as pool:
        agent = pool.get_agent("remote_assistant")
        result = await agent.run("What can you do?")

from llmling_agent.agent import AGUIAgent

def user_confirmation(action: str, importance: str = "medium") -> bool:
    """Ask user to confirm an action."""
    print(f"[{importance.upper()}] Confirm: {action}")
    response = input("Approve? (y/n): ")
    return response.lower() == "y"

def fetch_user_data(user_id: str) -> dict:
    """Retrieve user data from local database."""
    # Access local database
    return {"id": user_id, "name": "John Doe", "role": "admin"}

async def main():
    async with AGUIAgent(
        endpoint="http://localhost:8000/agent/run",
        name="remote-agent",
        tools=[user_confirmation, fetch_user_data],
    ) as agent:
        # Agent can now request these tools
        result = await agent.run("Deploy to production")
        # Will prompt for user confirmation

async with AGUIAgent(
    endpoint="http://localhost:8000/agent/run",
    name="agent",
) as agent:
    # Register tools dynamically
    @agent.register_tool
    def new_tool(x: int) -> int:
        """Process a number."""
        return x * 2

    # Tool is now available
    result = await agent.run("Use the new tool on 5")

def risky_operation() -> str:
    """An operation that might fail."""
    raise ValueError("Something went wrong")

async with AGUIAgent(
    endpoint="http://localhost:8000/agent/run",
    tools=[risky_operation],
) as agent:
    # Error is caught and sent to remote agent as tool error
    result = await agent.run("Try the risky operation")
    # Agent receives error message and can adapt

def confirm_deployment(
    environment: str,
    changes: list[str],
    risk_level: str = "medium"
) -> bool:
    """Ask user to confirm deployment."""
    print(f"\n{'='*50}")
    print(f"DEPLOYMENT TO {environment.upper()}")
    print(f"Risk Level: {risk_level}")
    print(f"\nChanges:")
    for change in changes:
        print(f"  - {change}")
    print('='*50)

    response = input("\nProceed with deployment? (yes/no): ")
    return response.lower() in ("yes", "y")

import sqlite3

def query_database(query: str, params: dict | None = None) -> list[dict]:
    """Execute a SQL query against local database."""
    conn = sqlite3.connect("local.db")
    cursor = conn.cursor()

    if params:
        cursor.execute(query, params)
    else:
        cursor.execute(query)

    columns = [desc[0] for desc in cursor.description]
    results = [dict(zip(columns, row)) for row in cursor.fetchall()]

    conn.close()
    return results

from typing import Literal

def navigate_to(
    page: Literal["home", "dashboard", "settings", "profile"],
    params: dict | None = None
) -> str:
    """Navigate to a different page in the application."""
    # In a real app, this would trigger navigation
    print(f"Navigating to {page} with params: {params}")
    return f"Navigated to {page}"

async with AgentPool("agents.yml") as pool:
    # Get individual agents
    local = pool.get_agent("local_assistant")
    remote = pool.get_agent("remote_specialist")

    # Or use the team
    team = pool.get_team("hybrid_team")
    result = await team.run("Process this request")

from llmling_agent.agent.events import ToolCallStartEvent

async def tool_call_handler(ctx, event):
    if isinstance(event, ToolCallStartEvent):
        print(f"Tool called: {event.tool_name}")

async with AGUIAgent(
    endpoint="http://localhost:8000/agent/run",
    event_handlers=[tool_call_handler],
) as agent:
    result = await agent.run("Use a tool")

import httpx

async with AGUIAgent(
    endpoint="http://localhost:8000/agent/run",
    timeout=120.0,  # Increase timeout
    headers={"Connection": "keep-alive"},
) as agent:
    result = await agent.run("prompt")

tools = await agent.tools.get_tools()
for tool in tools:
    print(f"Tool: {tool.name}")
    print(f"Schema: {tool.schema}")

import logging
logging.basicConfig(level=logging.DEBUG)

async with AGUIAgent(...) as agent:
    # Will log SSE parsing details
    result = await agent.run("prompt")