Skip to main content
Function calling lets your AI agent execute Python functions and access external services during conversations. This guide covers registering functions, connecting MCP servers, and building agents with tool access.
For a conceptual overview of MCP, see Model Context Protocol.

Registering Functions

Use @llm.register_function() to make any Python function callable by the LLM:
from vision_agents.plugins import openai

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

@llm.register_function(description="Get current weather for a location")
def get_weather(location: str) -> dict:
    return {"location": location, "temperature": "22°C", "condition": "Sunny"}

@llm.register_function(description="Calculate the sum of two numbers")
def calculate_sum(a: int, b: int) -> int:
    return a + b
The LLM automatically calls these functions when relevant:
response = await llm.simple_response("What's the weather in London?")
# Calls get_weather("London") and incorporates result

Parameters and Types

Functions support any combination of required and optional parameters:
@llm.register_function(description="Search for products")
def search_products(
    query: str,                    # Required
    category: str = "all",         # Optional with default
    max_price: float = 1000.0,     # Optional with default
    in_stock: bool = True          # Optional with default
) -> list:
    return [{"name": "Product 1", "price": 29.99}]

Custom Function Names

Override the function name exposed to the LLM:
@llm.register_function(
    name="check_permissions",
    description="Check if a user has specific permissions"
)
def verify_user_access(user_id: str, permission: str) -> bool:
    return True

MCP Servers

MCP servers provide your agent with access to external tools and services. Vision Agents supports both local and remote servers.

Local Servers

Run on your machine via stdio:
from vision_agents.core.mcp import MCPServerLocal

local_server = MCPServerLocal(
    command="python my_mcp_server.py",
    session_timeout=300.0
)

Remote Servers

Connect over HTTP:
from vision_agents.core.mcp import MCPServerRemote

github_server = MCPServerRemote(
    url="https://api.githubcopilot.com/mcp/",
    headers={"Authorization": f"Bearer {token}"},
    timeout=10.0,
    session_timeout=300.0
)

Connecting to Agent

Pass MCP servers to your agent — tools are automatically discovered and registered:
from vision_agents.core.agents import Agent
from vision_agents.core.mcp import MCPServerRemote
from vision_agents.plugins import openai, getstream
from vision_agents.core.edge.types import User

agent = Agent(
    edge=getstream.Edge(),
    llm=openai.LLM(model="gpt-4o-mini"),
    agent_user=User(name="Assistant", id="agent"),
    instructions="You have access to GitHub tools.",
    mcp_servers=[github_server]
)

Multiple Servers

Connect multiple MCP servers for access to different services:
agent = Agent(
    edge=getstream.Edge(),
    llm=llm,
    agent_user=User(name="Multi-Tool Assistant", id="agent"),
    mcp_servers=[github_server, weather_server, database_server]
)

Complete Example

Combining registered functions with MCP servers:
import asyncio
import os
from vision_agents.core.agents import Agent
from vision_agents.core.mcp import MCPServerRemote
from vision_agents.plugins import openai, getstream
from vision_agents.core.edge.types import User

async def main():
    # MCP server for GitHub access
    github_server = MCPServerRemote(
        url="https://api.githubcopilot.com/mcp/",
        headers={"Authorization": f"Bearer {os.getenv('GITHUB_PAT')}"},
        timeout=10.0
    )
    
    # LLM with custom function
    llm = openai.LLM(model="gpt-4o-mini")
    
    @llm.register_function(description="Get current time")
    def get_current_time() -> str:
        from datetime import datetime
        return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    
    # Agent with both MCP tools and custom functions
    agent = Agent(
        edge=getstream.Edge(),
        llm=llm,
        agent_user=User(name="GitHub Assistant", id="agent"),
        instructions="You can access GitHub and tell the time.",
        mcp_servers=[github_server]
    )
    
    await agent.create_user()
    call = agent.edge.client.video.call("default", "mcp-demo")
    
    async with agent.join(call):
        await agent.finish()

if __name__ == "__main__":
    asyncio.run(main())

Implementing Function Calling in Custom LLMs

When building a custom LLM plugin, you inherit function calling support from the base LLM class. The base class provides a FunctionRegistry that stores registered functions and handles execution.

How the Function Registry Works

The base LLM class (source) provides:
# Already available in your custom LLM:
self.function_registry        # FunctionRegistry instance
self.get_available_functions() # Returns List[ToolSchema]
self.call_function(name, args) # Executes a registered function
self._execute_tools(calls)     # Runs multiple tools concurrently
self._dedup_and_execute(calls) # Deduplicates and executes

Key Types

The function registry uses these types from vision_agents.core.llm.llm_types:
TypeFieldsDescription
ToolSchemaname, description, parameters_schemaFunction definition for LLMs
NormalizedToolCallItemtype, name, arguments_json, idStandardized tool call format

Required Overrides

To enable function calling, override these methods in your custom LLM:
from typing import Any, Dict, List
from vision_agents.core.llm.llm import LLM
from vision_agents.core.llm.llm_types import ToolSchema, NormalizedToolCallItem

class MyCustomLLM(LLM):
    
    def _convert_tools_to_provider_format(
        self, tools: List[ToolSchema]
    ) -> List[Dict[str, Any]]:
        """Convert ToolSchema to your provider's format."""
        return [
            {
                "name": tool["name"],
                "description": tool.get("description", ""),
                "parameters": tool.get("parameters_schema", {})
            }
            for tool in tools
        ]
    
    def _extract_tool_calls_from_response(
        self, response: Any
    ) -> List[NormalizedToolCallItem]:
        """Extract tool calls from your provider's response."""
        calls = []
        for tool_call in response.get("tool_calls", []):
            calls.append({
                "type": "tool_call",
                "id": tool_call["id"],
                "name": tool_call["function"]["name"],
                "arguments_json": tool_call["function"]["arguments"]
            })
        return calls

Using Tools in Your LLM

In your simple_response() or equivalent method:
async def simple_response(self, text: str, **kwargs):
    # 1. Get tools in provider format
    tools = self.get_available_functions()
    provider_tools = self._convert_tools_to_provider_format(tools)
    
    # 2. Call your provider with tools
    response = await self.client.chat(
        messages=[{"role": "user", "content": text}],
        tools=provider_tools
    )
    
    # 3. Extract and execute tool calls
    tool_calls = self._extract_tool_calls_from_response(response)
    if tool_calls:
        # _dedup_and_execute runs tools concurrently with deduplication
        triples, seen = await self._dedup_and_execute(tool_calls)
        # triples = [(tool_call_dict, result, error), ...]
        
        # 4. Build tool result messages and get follow-up response
        tool_results = self._create_tool_result_message(
            [t[0] for t in triples],
            [t[1] for t in triples]
        )
        # Send results back to provider...
    
    return response

Tool Execution Events

The base class automatically emits events when tools execute:
EventWhenFields
ToolStartEventBefore tool runstool_name, arguments, tool_call_id
ToolEndEventAfter tool completestool_name, success, result/error, execution_time_ms

Multi-Round Tool Calling

The OpenAI plugin supports multiple tool-calling rounds. If the model needs to call more tools after seeing results, it can continue for up to max_tool_rounds (default: 3):
llm = openai.LLM(model="gpt-4o-mini", max_tool_rounds=5)
See the OpenAI LLM implementation for a complete reference.