Function Calling

from vision_agents.core.llm import LLM

class MyLLM(LLM):
    def __init__(self):
        super().__init__()
        # self.function_registry is automatically created

from vision_agents.llm import openai

llm = openai.LLM(model="gpt-4o")

@llm.register_function(
    name="get_weather",  # Optional, defaults to function name
    description="Get current weather for a city",  # Optional, from docstring
)
async def get_weather(city: str, units: str = "fahrenheit") -> dict:
    """Fetch weather data for a city.
    
    Args:
        city: City name
        units: Temperature units (fahrenheit or celsius)
    
    Returns:
        Weather data including temperature and conditions
    """
    # Call weather API
    response = await weather_api.get(city, units)
    return {
        "temperature": response.temp,
        "conditions": response.conditions,
        "humidity": response.humidity,
    }

# ✅ Correct
@llm.register_function()
async def fetch_data(query: str) -> dict:
    return await api.get(query)

# ❌ Wrong - will raise ValueError
@llm.register_function()
def fetch_data(query: str) -> dict:
    return api.get_sync(query)

from typing import List, Optional

@llm.register_function()
async def search_products(
    query: str,
    max_results: int = 10,
    categories: Optional[List[str]] = None,
) -> List[dict]:
    """Search for products."""
    # Type hints are converted to JSON schema:
    # query: {"type": "string"}
    # max_results: {"type": "integer"}
    # categories: {"type": "array", "items": {"type": "string"}}
    pass

# Internal LLM execution flow
async def _execute_tools(
    self,
    calls: List[NormalizedToolCallItem],
    max_concurrency: int = 8,  # Run up to 8 functions in parallel
    timeout_s: float = 30,     # 30 second timeout per function
):
    """Execute multiple tool calls concurrently."""
    sem = asyncio.Semaphore(max_concurrency)
    
    async def _guarded(tc):
        async with sem:
            return await self._run_one_tool(tc, timeout_s)
    
    return await asyncio.gather(*[_guarded(tc) for tc in calls])

def _tc_key(self, tc: NormalizedToolCallItem) -> Tuple:
    """Generate unique key for deduplication."""
    return (
        tc.get("id"),
        tc["name"],
        json.dumps(tc.get("arguments_json", {}), sort_keys=True),
    )

# Same function with same args is only called once
await self._dedup_and_execute(calls, seen=seen_set)

try:
    result = await asyncio.wait_for(
        function(**args),
        timeout=30.0,  # Configurable per execution
    )
except asyncio.TimeoutError:
    return {"error": "Function timed out"}

from vision_agents.core.llm.events import ToolStartEvent

@agent.subscribe
async def on_tool_start(event: ToolStartEvent):
    print(f"Calling {event.tool_name} with {event.arguments}")
    # event.tool_call_id - Unique ID for this call

from vision_agents.core.llm.events import ToolEndEvent

@agent.subscribe
async def on_tool_end(event: ToolEndEvent):
    if event.success:
        print(f"{event.tool_name} succeeded: {event.result}")
    else:
        print(f"{event.tool_name} failed: {event.error}")
    
    print(f"Execution time: {event.execution_time_ms}ms")

@llm.register_function()
async def risky_operation(param: str) -> dict:
    if not param:
        raise ValueError("param is required")
    
    response = await external_api.call(param)
    return response

# When error occurs:
# {"error": "ValueError: param is required"}
# This is sent back to the LLM to handle gracefully

def _sanitize_tool_output(self, value: Any, max_chars: int = 60_000) -> str:
    """Prevent oversized function outputs."""
    if isinstance(value, str):
        s = value
    elif isinstance(value, Exception):
        s = f"Error: {type(value).__name__}: {value}"
    else:
        s = json.dumps(value)
    
    return (s[:max_chars] + "…") if len(s) > max_chars else s

from vision_agents.mcp import FileSystemServer, WebSearchServer
from vision_agents import Agent

agent = Agent(
    # ... other config
    mcp_servers=[
        FileSystemServer(root_path="/data"),
        WebSearchServer(api_key="..."),
    ],
)

# MCP tools are automatically registered with the LLM
# The LLM can now call file operations and web search

from vision_agents.core.llm.function_registry import FunctionRegistry

registry = FunctionRegistry()

@registry.register(
    name="mcp_tool",
    description="Tool from MCP server",
    parameters_schema={  # Explicit JSON schema
        "type": "object",
        "properties": {
            "query": {"type": "string"},
            "limit": {"type": "integer"},
        },
        "required": ["query"],
    },
)
async def mcp_tool(query: str, limit: int = 10):
    # Tool implementation
    pass

schemas = llm.get_available_functions()

# Returns List[ToolSchema]
for schema in schemas:
    print(schema.name)          # Function name
    print(schema.description)   # Description
    print(schema.parameters_schema)  # JSON schema

from dataclasses import dataclass
from typing import Dict, Any

@dataclass
class ToolSchema:
    name: str
    description: str
    parameters_schema: Dict[str, Any]  # JSON schema

class LLM:
    def _convert_tools_to_provider_format(
        self,
        tools: List[ToolSchema],
    ) -> List[Dict[str, Any]]:
        """Convert to provider-specific format.
        
        Override this in each LLM implementation.
        """
        # OpenAI format:
        return [
            {
                "type": "function",
                "function": {
                    "name": tool.name,
                    "description": tool.description,
                    "parameters": tool.parameters_schema,
                },
            }
            for tool in tools
        ]

from vision_agents import Agent
from vision_agents.llm import openai
from vision_agents.edge import getstream
from typing import List, Optional
import httpx

# Create LLM
llm = openai.LLM(model="gpt-4o")

# Register weather function
@llm.register_function(
    description="Get current weather for a location"
)
async def get_weather(location: str, units: str = "celsius") -> dict:
    """Fetch weather data.
    
    Args:
        location: City name or coordinates
        units: Temperature units (celsius or fahrenheit)
    
    Returns:
        Weather data with temperature and conditions
    """
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://api.weather.com/current",
            params={"location": location, "units": units}
        )
        data = response.json()
        
        return {
            "temperature": data["temp"],
            "conditions": data["conditions"],
            "humidity": data["humidity"],
            "wind_speed": data["wind"],
        }

# Register stock price function
@llm.register_function(
    description="Get current stock price"
)
async def get_stock_price(symbol: str) -> dict:
    """Fetch current stock price.
    
    Args:
        symbol: Stock ticker symbol (e.g., AAPL, GOOGL)
    
    Returns:
        Stock price and change information
    """
    async with httpx.AsyncClient() as client:
        response = await client.get(
            f"https://api.stocks.com/quote/{symbol}"
        )
        data = response.json()
        
        return {
            "symbol": symbol,
            "price": data["current_price"],
            "change": data["change"],
            "change_percent": data["change_percent"],
        }

# Register database query function
@llm.register_function(
    description="Search customer database"
)
async def search_customers(
    query: str,
    limit: int = 10,
    filters: Optional[List[str]] = None,
) -> List[dict]:
    """Search for customers in database.
    
    Args:
        query: Search query
        limit: Maximum number of results
        filters: Optional filters to apply
    
    Returns:
        List of matching customer records
    """
    # Query database
    results = await db.query(
        "SELECT * FROM customers WHERE name LIKE ?",
        f"%{query}%",
        limit=limit,
    )
    
    return [
        {
            "id": r.id,
            "name": r.name,
            "email": r.email,
            "status": r.status,
        }
        for r in results
    ]

# Create agent with functions
agent = Agent(
    edge=getstream.Edge(),
    llm=llm,
    agent_user=User(id="agent-1", name="Assistant"),
    instructions="""
        You are a helpful assistant with access to weather, stock prices, 
        and customer data. Use the available functions to answer questions.
    """,
)

# Subscribe to function events
@agent.subscribe
async def on_tool_start(event: ToolStartEvent):
    print(f"🔧 Calling {event.tool_name}...")

@agent.subscribe
async def on_tool_end(event: ToolEndEvent):
    if event.success:
        print(f"✅ {event.tool_name} completed in {event.execution_time_ms}ms")
    else:
        print(f"❌ {event.tool_name} failed: {event.error}")

# Use agent
async with agent.join(call):
    # User: "What's the weather in San Francisco?"
    # Agent will call get_weather(location="San Francisco")
    
    # User: "How's Apple stock doing?"
    # Agent will call get_stock_price(symbol="AAPL")
    
    # User: "Find customers named John"
    # Agent will call search_customers(query="John")
    
    await agent.finish()