Function calling lets your AI agent execute Python functions and access external services during conversations.
Vision Agents requires a Stream account for real-time transport.
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")
async def get_weather(location: str) -> dict:
return {"location": location, "temperature": "22°C", "condition": "Sunny"}
@llm.register_function(description="Calculate the sum of two numbers")
async def calculate_sum(a: int, b: int) -> int:
return a + b
Only async functions can be registered. Passing a synchronous function to @register_function() raises a ValueError.
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 required and optional parameters:
@llm.register_function(description="Search for products")
async def search_products(
query: str, # Required
category: str = "all", # Optional with default
max_price: float = 1000.0,
in_stock: bool = True
) -> 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"
)
async 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.
Local Servers
Run on your machine via stdio:
from vision_agents.core.mcp import MCPServerLocal
local_server = MCPServerLocal(
command="uv run 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 import Agent, User
from vision_agents.plugins import openai, getstream
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
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
import asyncio
import os
from vision_agents.core import Agent, User
from vision_agents.core.mcp import MCPServerRemote
from vision_agents.plugins import openai, getstream
async def main():
github_server = MCPServerRemote(
url="https://api.githubcopilot.com/mcp/",
headers={"Authorization": f"Bearer {os.getenv('GITHUB_PAT')}"},
timeout=10.0
)
llm = openai.LLM(model="gpt-4o-mini")
@llm.register_function(description="Get current time")
async def get_current_time() -> str:
from datetime import datetime
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
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())
The framework emits events when tools execute:
| Event | When | Fields |
|---|
ToolStartEvent | Before tool runs | tool_name, arguments, tool_call_id |
ToolEndEvent | After tool completes | tool_name, success, result/error, execution_time_ms |
The OpenAI plugin supports multiple tool-calling rounds. If the model needs to call more tools after seeing results:
llm = openai.LLM(model="gpt-4o-mini", max_tool_rounds=5)
Next Steps