Introduction: Function calling transforms LLMs from text generators into action-taking agents. Instead of just describing what to do, the model can invoke actual functions with structured arguments. This enables powerful integrations: querying databases, calling APIs, executing code, and orchestrating complex workflows. But function calling requires careful design—poorly defined functions confuse the model, missing validation causes runtime errors, and uncontrolled execution creates security risks. This guide covers practical function calling patterns: defining clear function schemas, validating and executing calls safely, handling multi-step tool use, and building robust systems that let LLMs take real actions.
Function Definition
from dataclasses import dataclass, field
from typing import Any, Optional, Callable, get_type_hints
import json
import inspect
@dataclass
class FunctionParameter:
"""A function parameter definition."""
name: str
type: str
description: str
required: bool = True
enum: list[str] = None
default: Any = None
@dataclass
class FunctionDefinition:
"""A function definition for LLM tool use."""
name: str
description: str
parameters: list[FunctionParameter]
handler: Callable = None
def to_openai_schema(self) -> dict:
"""Convert to OpenAI function schema."""
properties = {}
required = []
for param in self.parameters:
prop = {
"type": param.type,
"description": param.description
}
if param.enum:
prop["enum"] = param.enum
properties[param.name] = prop
if param.required:
required.append(param.name)
return {
"type": "function",
"function": {
"name": self.name,
"description": self.description,
"parameters": {
"type": "object",
"properties": properties,
"required": required
}
}
}
def function_tool(description: str = None):
"""Decorator to create function definition from Python function."""
def decorator(func: Callable) -> FunctionDefinition:
hints = get_type_hints(func)
sig = inspect.signature(func)
doc = func.__doc__ or ""
# Extract parameter descriptions from docstring
param_docs = {}
for line in doc.split('\n'):
line = line.strip()
if line.startswith(':param'):
parts = line.split(':', 2)
if len(parts) >= 3:
param_name = parts[1].replace('param ', '').strip()
param_desc = parts[2].strip()
param_docs[param_name] = param_desc
parameters = []
for name, param in sig.parameters.items():
if name == 'self':
continue
# Get type
type_hint = hints.get(name, str)
json_type = _python_type_to_json(type_hint)
# Get description
param_desc = param_docs.get(name, f"The {name} parameter")
# Check if required
required = param.default == inspect.Parameter.empty
default = None if required else param.default
parameters.append(FunctionParameter(
name=name,
type=json_type,
description=param_desc,
required=required,
default=default
))
return FunctionDefinition(
name=func.__name__,
description=description or doc.split('\n')[0],
parameters=parameters,
handler=func
)
return decorator
def _python_type_to_json(python_type) -> str:
"""Convert Python type to JSON schema type."""
type_map = {
str: "string",
int: "integer",
float: "number",
bool: "boolean",
list: "array",
dict: "object"
}
return type_map.get(python_type, "string")
# Example function definitions
@function_tool("Get the current weather for a location")
def get_weather(
location: str,
unit: str = "celsius"
) -> dict:
"""
Get current weather information.
:param location: The city and country, e.g., "London, UK"
:param unit: Temperature unit, either "celsius" or "fahrenheit"
"""
# Implementation would call weather API
return {"temperature": 20, "condition": "sunny", "unit": unit}
@function_tool("Search for information in a knowledge base")
def search_knowledge(
query: str,
max_results: int = 5
) -> list:
"""
Search the knowledge base for relevant information.
:param query: The search query
:param max_results: Maximum number of results to return
"""
# Implementation would search vector database
return [{"title": "Result 1", "content": "..."}]Function Execution
from dataclasses import dataclass
from typing import Any, Optional, Callable
import json
import asyncio
@dataclass
class FunctionCall:
"""A function call from the LLM."""
name: str
arguments: dict
call_id: str = None
@dataclass
class FunctionResult:
"""Result of function execution."""
call_id: str
name: str
result: Any
success: bool
error: str = None
class FunctionRegistry:
"""Registry of available functions."""
def __init__(self):
self._functions: dict[str, FunctionDefinition] = {}
def register(self, func_def: FunctionDefinition):
"""Register a function."""
self._functions[func_def.name] = func_def
def get(self, name: str) -> Optional[FunctionDefinition]:
"""Get a function by name."""
return self._functions.get(name)
def get_schemas(self) -> list[dict]:
"""Get OpenAI schemas for all functions."""
return [f.to_openai_schema() for f in self._functions.values()]
def list_functions(self) -> list[str]:
"""List all registered function names."""
return list(self._functions.keys())
class FunctionExecutor:
"""Execute function calls safely."""
def __init__(self, registry: FunctionRegistry):
self.registry = registry
async def execute(self, call: FunctionCall) -> FunctionResult:
"""Execute a function call."""
# Get function definition
func_def = self.registry.get(call.name)
if func_def is None:
return FunctionResult(
call_id=call.call_id,
name=call.name,
result=None,
success=False,
error=f"Unknown function: {call.name}"
)
# Validate arguments
validation_error = self._validate_arguments(func_def, call.arguments)
if validation_error:
return FunctionResult(
call_id=call.call_id,
name=call.name,
result=None,
success=False,
error=validation_error
)
# Execute
try:
if asyncio.iscoroutinefunction(func_def.handler):
result = await func_def.handler(**call.arguments)
else:
result = func_def.handler(**call.arguments)
return FunctionResult(
call_id=call.call_id,
name=call.name,
result=result,
success=True
)
except Exception as e:
return FunctionResult(
call_id=call.call_id,
name=call.name,
result=None,
success=False,
error=str(e)
)
def _validate_arguments(
self,
func_def: FunctionDefinition,
arguments: dict
) -> Optional[str]:
"""Validate function arguments."""
# Check required parameters
for param in func_def.parameters:
if param.required and param.name not in arguments:
return f"Missing required parameter: {param.name}"
# Check for unknown parameters
known_params = {p.name for p in func_def.parameters}
for arg_name in arguments:
if arg_name not in known_params:
return f"Unknown parameter: {arg_name}"
# Type validation could be added here
return None
class SafeExecutor(FunctionExecutor):
"""Executor with additional safety measures."""
def __init__(
self,
registry: FunctionRegistry,
allowed_functions: list[str] = None,
timeout_seconds: float = 30.0
):
super().__init__(registry)
self.allowed_functions = set(allowed_functions) if allowed_functions else None
self.timeout_seconds = timeout_seconds
async def execute(self, call: FunctionCall) -> FunctionResult:
"""Execute with safety checks."""
# Check if function is allowed
if self.allowed_functions and call.name not in self.allowed_functions:
return FunctionResult(
call_id=call.call_id,
name=call.name,
result=None,
success=False,
error=f"Function not allowed: {call.name}"
)
# Execute with timeout
try:
result = await asyncio.wait_for(
super().execute(call),
timeout=self.timeout_seconds
)
return result
except asyncio.TimeoutError:
return FunctionResult(
call_id=call.call_id,
name=call.name,
result=None,
success=False,
error=f"Function execution timed out after {self.timeout_seconds}s"
)Tool Use Loop
from dataclasses import dataclass
from typing import Any, Optional
import json
@dataclass
class ToolUseResult:
"""Result of tool use conversation."""
final_response: str
tool_calls: list[FunctionCall]
tool_results: list[FunctionResult]
iterations: int
class ToolUseAgent:
"""Agent that uses tools to complete tasks."""
def __init__(
self,
client: Any,
executor: FunctionExecutor,
model: str = "gpt-4o",
max_iterations: int = 10
):
self.client = client
self.executor = executor
self.model = model
self.max_iterations = max_iterations
async def run(
self,
messages: list[dict],
tools: list[dict] = None
) -> ToolUseResult:
"""Run tool use loop until completion."""
tools = tools or self.executor.registry.get_schemas()
all_tool_calls = []
all_tool_results = []
iterations = 0
while iterations < self.max_iterations:
iterations += 1
# Call LLM
response = await self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=tools,
tool_choice="auto"
)
message = response.choices[0].message
# Check if done (no tool calls)
if not message.tool_calls:
return ToolUseResult(
final_response=message.content or "",
tool_calls=all_tool_calls,
tool_results=all_tool_results,
iterations=iterations
)
# Add assistant message to history
messages.append({
"role": "assistant",
"content": message.content,
"tool_calls": [
{
"id": tc.id,
"type": "function",
"function": {
"name": tc.function.name,
"arguments": tc.function.arguments
}
}
for tc in message.tool_calls
]
})
# Execute tool calls
for tool_call in message.tool_calls:
call = FunctionCall(
name=tool_call.function.name,
arguments=json.loads(tool_call.function.arguments),
call_id=tool_call.id
)
all_tool_calls.append(call)
result = await self.executor.execute(call)
all_tool_results.append(result)
# Add tool result to messages
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result.result) if result.success else result.error
})
# Max iterations reached
return ToolUseResult(
final_response="Maximum iterations reached without completion",
tool_calls=all_tool_calls,
tool_results=all_tool_results,
iterations=iterations
)
class ParallelToolAgent(ToolUseAgent):
"""Agent that executes multiple tool calls in parallel."""
async def run(
self,
messages: list[dict],
tools: list[dict] = None
) -> ToolUseResult:
"""Run with parallel tool execution."""
tools = tools or self.executor.registry.get_schemas()
all_tool_calls = []
all_tool_results = []
iterations = 0
while iterations < self.max_iterations:
iterations += 1
response = await self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=tools,
tool_choice="auto"
)
message = response.choices[0].message
if not message.tool_calls:
return ToolUseResult(
final_response=message.content or "",
tool_calls=all_tool_calls,
tool_results=all_tool_results,
iterations=iterations
)
messages.append({
"role": "assistant",
"content": message.content,
"tool_calls": [
{
"id": tc.id,
"type": "function",
"function": {
"name": tc.function.name,
"arguments": tc.function.arguments
}
}
for tc in message.tool_calls
]
})
# Execute all tool calls in parallel
calls = [
FunctionCall(
name=tc.function.name,
arguments=json.loads(tc.function.arguments),
call_id=tc.id
)
for tc in message.tool_calls
]
all_tool_calls.extend(calls)
results = await asyncio.gather(*[
self.executor.execute(call)
for call in calls
])
all_tool_results.extend(results)
# Add all results to messages
for tool_call, result in zip(message.tool_calls, results):
messages.append({
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result.result) if result.success else result.error
})
return ToolUseResult(
final_response="Maximum iterations reached",
tool_calls=all_tool_calls,
tool_results=all_tool_results,
iterations=iterations
)Structured Output
from dataclasses import dataclass
from typing import Any, Optional, Type
from pydantic import BaseModel
import json
class StructuredOutputExtractor:
"""Extract structured output using function calling."""
def __init__(self, client: Any, model: str = "gpt-4o"):
self.client = client
self.model = model
async def extract(
self,
text: str,
schema: Type[BaseModel],
instructions: str = None
) -> BaseModel:
"""Extract structured data from text."""
# Create function schema from Pydantic model
json_schema = schema.model_json_schema()
function = {
"type": "function",
"function": {
"name": "extract_data",
"description": instructions or f"Extract {schema.__name__} from the text",
"parameters": json_schema
}
}
messages = [
{
"role": "system",
"content": "Extract the requested information from the provided text. Call the extract_data function with the extracted values."
},
{
"role": "user",
"content": text
}
]
response = await self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=[function],
tool_choice={"type": "function", "function": {"name": "extract_data"}}
)
# Parse the function call arguments
tool_call = response.choices[0].message.tool_calls[0]
arguments = json.loads(tool_call.function.arguments)
return schema(**arguments)
# Example usage
class ContactInfo(BaseModel):
name: str
email: str
phone: Optional[str] = None
company: Optional[str] = None
class OrderDetails(BaseModel):
product_name: str
quantity: int
unit_price: float
shipping_address: str
# Extract contact info from text
async def extract_contact(text: str) -> ContactInfo:
extractor = StructuredOutputExtractor(client)
return await extractor.extract(
text,
ContactInfo,
"Extract contact information from the text"
)
class MultiSchemaExtractor:
"""Extract multiple schemas from text."""
def __init__(self, client: Any, model: str = "gpt-4o"):
self.client = client
self.model = model
async def extract_all(
self,
text: str,
schemas: dict[str, Type[BaseModel]]
) -> dict[str, BaseModel]:
"""Extract multiple schemas from text."""
# Create functions for each schema
functions = []
for name, schema in schemas.items():
functions.append({
"type": "function",
"function": {
"name": f"extract_{name}",
"description": f"Extract {name} information",
"parameters": schema.model_json_schema()
}
})
messages = [
{
"role": "system",
"content": "Extract all requested information from the text. Call the appropriate extraction functions."
},
{
"role": "user",
"content": text
}
]
response = await self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=functions,
tool_choice="auto"
)
results = {}
for tool_call in response.choices[0].message.tool_calls or []:
func_name = tool_call.function.name
schema_name = func_name.replace("extract_", "")
if schema_name in schemas:
arguments = json.loads(tool_call.function.arguments)
results[schema_name] = schemas[schema_name](**arguments)
return resultsProduction Function Service
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional, Any
app = FastAPI()
# Initialize components
registry = FunctionRegistry()
executor = SafeExecutor(registry)
agent = ParallelToolAgent(client=None, executor=executor)
# Register example functions
registry.register(get_weather)
registry.register(search_knowledge)
class ExecuteRequest(BaseModel):
function_name: str
arguments: dict
class AgentRequest(BaseModel):
messages: list[dict]
max_iterations: int = 10
class ExtractRequest(BaseModel):
text: str
schema_name: str
@app.post("/v1/functions/execute")
async def execute_function(request: ExecuteRequest):
"""Execute a single function."""
call = FunctionCall(
name=request.function_name,
arguments=request.arguments,
call_id="direct"
)
result = await executor.execute(call)
if not result.success:
raise HTTPException(status_code=400, detail=result.error)
return {
"result": result.result,
"function": result.name
}
@app.post("/v1/agent/run")
async def run_agent(request: AgentRequest):
"""Run agent with tool use."""
agent.max_iterations = request.max_iterations
result = await agent.run(request.messages)
return {
"response": result.final_response,
"iterations": result.iterations,
"tool_calls": [
{"name": tc.name, "arguments": tc.arguments}
for tc in result.tool_calls
],
"tool_results": [
{"name": tr.name, "success": tr.success, "result": tr.result}
for tr in result.tool_results
]
}
@app.get("/v1/functions")
async def list_functions():
"""List available functions."""
return {
"functions": [
{
"name": func.name,
"description": func.description,
"parameters": [
{
"name": p.name,
"type": p.type,
"description": p.description,
"required": p.required
}
for p in func.parameters
]
}
for func in registry._functions.values()
]
}
@app.get("/v1/functions/{name}/schema")
async def get_function_schema(name: str):
"""Get OpenAI schema for a function."""
func = registry.get(name)
if func is None:
raise HTTPException(status_code=404, detail="Function not found")
return func.to_openai_schema()
@app.get("/health")
async def health():
return {"status": "healthy"}References
- OpenAI Function Calling: https://platform.openai.com/docs/guides/function-calling
- Anthropic Tool Use: https://docs.anthropic.com/claude/docs/tool-use
- LangChain Tools: https://python.langchain.com/docs/modules/tools/
- JSON Schema: https://json-schema.org/
Conclusion
Function calling unlocks the full potential of LLMs by letting them take real actions. Start with clear function definitions—good descriptions and parameter documentation help the model choose the right function and extract correct arguments. Use a registry to manage available functions and generate schemas automatically. Implement safe execution with validation, timeouts, and allowlists to prevent misuse. Build a tool use loop that handles multi-step interactions where the model calls functions, receives results, and decides what to do next. Execute independent function calls in parallel to reduce latency. Use function calling for structured extraction when you need reliable data formats. The key insight is that function calling is about giving LLMs agency—the ability to affect the world, not just describe it. Design your functions carefully, validate inputs thoroughly, and monitor execution closely. Well-designed function calling transforms chatbots into capable agents.
Discover more from Code, Cloud & Context
Subscribe to get the latest posts sent to your email.