Function calling (or tool use) is how LLMs interact with the real world—executing code, querying databases, calling APIs. Getting this right in production requires careful design. Reliability, security, and error handling matter.
Here are function calling patterns that work in production.
Function Calling Basics
How It Works
function_calling_flow:
1_define: "Declare available functions with schemas"
2_prompt: "Send user message with function definitions"
3_decision: "Model decides if/which function to call"
4_execute: "Your code executes the function"
5_respond: "Send result back to model"
6_continue: "Model generates final response"
Basic Implementation
import openai
import json
# Define functions
tools = [
{
"type": "function",
"function": {
"name": "get_weather",
"description": "Get current weather for a location",
"parameters": {
"type": "object",
"properties": {
"location": {
"type": "string",
"description": "City name"
},
"unit": {
"type": "string",
"enum": ["celsius", "fahrenheit"],
"default": "celsius"
}
},
"required": ["location"]
}
}
}
]
# Call with functions
response = openai.chat.completions.create(
model="gpt-4-turbo",
messages=[{"role": "user", "content": "What's the weather in Tokyo?"}],
tools=tools
)
# Handle function call
if response.choices[0].message.tool_calls:
tool_call = response.choices[0].message.tool_calls[0]
args = json.loads(tool_call.function.arguments)
# Execute function
result = get_weather(args["location"], args.get("unit", "celsius"))
# Send result back
response = openai.chat.completions.create(
model="gpt-4-turbo",
messages=[
{"role": "user", "content": "What's the weather in Tokyo?"},
response.choices[0].message,
{
"role": "tool",
"tool_call_id": tool_call.id,
"content": json.dumps(result)
}
],
tools=tools
)
Production Patterns
Function Registry
from typing import Callable, Any
from dataclasses import dataclass
@dataclass
class FunctionDefinition:
name: str
description: str
parameters: dict
handler: Callable
requires_auth: bool = False
rate_limit: int = None
class FunctionRegistry:
def __init__(self):
self._functions: dict[str, FunctionDefinition] = {}
def register(
self,
name: str,
description: str,
parameters: dict,
requires_auth: bool = False,
rate_limit: int = None
):
def decorator(func: Callable):
self._functions[name] = FunctionDefinition(
name=name,
description=description,
parameters=parameters,
handler=func,
requires_auth=requires_auth,
rate_limit=rate_limit
)
return func
return decorator
def get_tools_schema(self) -> list[dict]:
return [
{
"type": "function",
"function": {
"name": f.name,
"description": f.description,
"parameters": f.parameters
}
}
for f in self._functions.values()
]
async def execute(
self,
name: str,
arguments: dict,
context: ExecutionContext
) -> Any:
func_def = self._functions.get(name)
if not func_def:
raise FunctionNotFoundError(name)
if func_def.requires_auth and not context.is_authenticated:
raise AuthenticationRequired(name)
if func_def.rate_limit:
await self._check_rate_limit(name, context.user_id, func_def.rate_limit)
return await func_def.handler(**arguments)
# Usage
registry = FunctionRegistry()
@registry.register(
name="search_documents",
description="Search internal documents",
parameters={
"type": "object",
"properties": {
"query": {"type": "string"},
"limit": {"type": "integer", "default": 10}
},
"required": ["query"]
},
requires_auth=True
)
async def search_documents(query: str, limit: int = 10):
return await document_store.search(query, limit)
Validation Layer
from pydantic import BaseModel, ValidationError
from typing import Type
class FunctionValidator:
def __init__(self):
self._schemas: dict[str, Type[BaseModel]] = {}
def register_schema(self, name: str, schema: Type[BaseModel]):
self._schemas[name] = schema
def validate(self, name: str, arguments: dict) -> BaseModel:
schema = self._schemas.get(name)
if not schema:
return arguments
try:
return schema(**arguments)
except ValidationError as e:
raise InvalidFunctionArguments(name, e.errors())
# Define schemas
class WeatherArgs(BaseModel):
location: str
unit: str = "celsius"
@validator("location")
def location_not_empty(cls, v):
if not v.strip():
raise ValueError("Location cannot be empty")
return v.strip()
validator = FunctionValidator()
validator.register_schema("get_weather", WeatherArgs)
Error Handling
class FunctionExecutor:
async def execute_with_error_handling(
self,
tool_call: ToolCall,
context: ExecutionContext
) -> ToolResult:
try:
args = json.loads(tool_call.function.arguments)
validated_args = self.validator.validate(
tool_call.function.name,
args
)
result = await self.registry.execute(
tool_call.function.name,
validated_args.dict(),
context
)
return ToolResult(
tool_call_id=tool_call.id,
success=True,
result=result
)
except json.JSONDecodeError as e:
return ToolResult(
tool_call_id=tool_call.id,
success=False,
error=f"Invalid JSON in arguments: {e}"
)
except InvalidFunctionArguments as e:
return ToolResult(
tool_call_id=tool_call.id,
success=False,
error=f"Invalid arguments: {e.errors}"
)
except FunctionNotFoundError as e:
return ToolResult(
tool_call_id=tool_call.id,
success=False,
error=f"Unknown function: {e.name}"
)
except Exception as e:
logger.exception(f"Function execution failed: {tool_call.function.name}")
return ToolResult(
tool_call_id=tool_call.id,
success=False,
error="Function execution failed"
)
Parallel Execution
async def execute_parallel_calls(
self,
tool_calls: list[ToolCall],
context: ExecutionContext
) -> list[ToolResult]:
"""Execute multiple function calls in parallel."""
tasks = [
self.execute_with_error_handling(call, context)
for call in tool_calls
]
results = await asyncio.gather(*tasks, return_exceptions=True)
return [
result if isinstance(result, ToolResult)
else ToolResult(
tool_call_id=tool_calls[i].id,
success=False,
error=str(result)
)
for i, result in enumerate(results)
]
Security Considerations
Input Sanitization
class SecureFunctionExecutor:
DANGEROUS_PATTERNS = [
r";\s*rm\s+-rf",
r";\s*DROP\s+TABLE",
r"<script>",
]
def sanitize_arguments(self, args: dict) -> dict:
sanitized = {}
for key, value in args.items():
if isinstance(value, str):
for pattern in self.DANGEROUS_PATTERNS:
if re.search(pattern, value, re.IGNORECASE):
raise SecurityViolation(f"Dangerous pattern in {key}")
sanitized[key] = self._escape_special_chars(value)
else:
sanitized[key] = value
return sanitized
Permission Scoping
function_permissions:
read_only:
- search_documents
- get_user_profile
- list_items
write_requires_approval:
- update_record
- send_email
- create_ticket
admin_only:
- delete_user
- modify_permissions
- access_audit_logs
Key Takeaways
- Function calling turns LLMs into action-taking agents
- Use a registry pattern for managing functions
- Validate all arguments with schemas (Pydantic)
- Handle errors gracefully—return useful messages
- Execute parallel calls when possible
- Implement security at every layer
- Rate limit sensitive functions
- Log all function executions for auditing
- Test with adversarial inputs
- Start with read-only functions, add writes carefully
Function calling is powerful. Use it responsibly.