Building AI Agents with Function Calling: Complete Implementation Guide
Build production-ready AI agents that can execute code, call APIs, query databases, and interact with external systems through function calling.
Introduction
Function calling transforms LLMs from text generators into action-taking agents. This guide covers building robust function-calling agents with proper validation, error handling, and production patterns.
Function Calling Architecture
USER REQUEST
│
▼
┌─────────────┐
│ LLM │ ──▶ Analyzes request, decides function calls
│ Reasoning │
└─────────────┘
│
▼ (if function needed)
┌─────────────┐
│ Function │ ──▶ Validates, executes, returns result
│ Executor │
└─────────────┘
│
▼
┌─────────────┐
│ LLM │ ──▶ Synthesizes final response
│ Response │
└─────────────┘
Part 1: Function Definition & Schema
JSON Schema for Functions
# functions/schemas.py
from typing: Dict, List, Any, Optional, get_type_hints
from pydantic import BaseModel, Field, create_model
from dataclasses import dataclass
import inspect
import json
@dataclass
class FunctionSpec:
name: str
description: str
parameters: Dict # JSON Schema
required: List[str] = None
returns: Dict = None
category: str = "general"
version: str = "1.0"
def function_to_schema(func: callable) -> FunctionSpec:
"""Auto-generate function schema from Python function."""
sig = inspect.signature(func)
type_hints = get_type_hints(func)
docstring = inspect.getdoc(func) or ""
# Parse docstring for param descriptions
param_descriptions = _parse_docstring_params(docstring)
properties = {}
required = []
for name, param in sig.parameters.items():
if name in ("self", "cls"):
continue
param_type = type_hints.get(name, Any)
json_schema = _python_type_to_json_schema(param_type)
json_schema["description"] = param_descriptions.get(name, "")
properties[name] = json_schema
if param.default == inspect.Parameter.empty:
required.append(name)
return FunctionSpec(
name=func.__name__,
description=docstring.split("\n")[0] if docstring else "",
parameters={
"type": "object",
"properties": properties,
"required": required
},
required=required,
returns=_python_type_to_json_schema(sig.return_annotation) if sig.return_annotation != inspect.Signature.empty else {}
)
def _python_type_to_json_schema(py_type: type) -> Dict:
"""Convert Python type to JSON Schema."""
origin = getattr(py_type, "__origin__", None)
args = getattr(py_type, "__args__", ())
if py_type is str:
return {"type": "string"}
elif py_type is int:
return {"type": "integer"}
elif py_type is float:
return {"type": "number"}
elif py_type is bool:
return {"type": "boolean"}
elif origin is list or origin is List:
return {"type": "array", "items": _python_type_to_json_schema(args[0]) if args else {}}
elif origin is dict or origin is Dict:
return {"type": "object", "additionalProperties": _python_type_to_json_schema(args[1]) if len(args) > 1 else {}}
elif origin is Optional or origin is Union:
# Handle Optional[T] = Union[T, None]
non_none = [a for a in args if a is not type(None)]
if non_none:
return _python_type_to_json_schema(non_none[0])
return {"type": "null"}
elif hasattr(py_type, "__annotations__"): # Pydantic model
return py_type.model_json_schema()
else:
return {"type": "string"} # fallback
# Example functions with rich schemas
class FunctionLibrary:
@staticmethod
def execute_python(code: str, timeout: int = 30) -> Dict:
"""Execute Python code in sandboxed environment.
Args:
code: Python code to execute
timeout: Maximum execution time in seconds
"""
pass
@staticmethod
def http_request(
method: str,
url: str,
headers: Dict = None,
body: str = None,
timeout: int = 30
) -> Dict:
"""Make HTTP request to external API.
Args:
method: HTTP method (GET, POST, PUT, DELETE, etc.)
url: Target URL
headers: Optional request headers
body: Optional request body
timeout: Request timeout in seconds
"""
pass
@staticmethod
def query_database(sql: str, params: List = None) -> Dict:
"""Execute read-only SQL query.
Args:
sql: SELECT query to execute
params: Query parameters for prepared statement
"""
pass
@staticmethod
def search_web(query: str, max_results: int = 10) -> Dict:
"""Search the web for information.
Args:
query: Search query
max_results: Maximum number of results to return
"""
pass
@staticmethod
def file_operation(
operation: str,
path: str,
content: str = None
) -> Dict:
"""Perform file operations.
Args:
operation: read, write, append, delete, list
path: File or directory path
content: Content for write/append operations
"""
pass
Function Registry
# functions/registry.py
from typing: Dict, List, Callable, Any
from dataclasses import dataclass, field
import json
@dataclass
class RegisteredFunction:
spec: FunctionSpec
handler: Callable
permissions: List[str] = field(default_factory=list) # required permissions
rate_limit: int = 100 # calls per minute
timeout: int = 30
class FunctionRegistry:
def __init__(self):
self.functions: Dict[str, RegisteredFunction] = {}
self.categories: Dict[str, List[str]] = {}
def register(
self,
func: Callable,
spec: FunctionSpec = None,
permissions: List[str] = None,
rate_limit: int = 100,
timeout: int = 30
):
"""Register a function for LLM use."""
if spec is None:
spec = function_to_schema(func)
registered = RegisteredFunction(
spec=spec,
handler=func,
permissions=permissions or [],
rate_limit=rate_limit,
timeout=timeout
)
self.functions[spec.name] = registered
if spec.category not in self.categories:
self.categories[spec.category] = []
self.categories[spec.category].append(spec.name)
def get_schema(self, name: str) -> Dict:
"""Get OpenAI-compatible function schema."""
if name not in self.functions:
raise ValueError(f"Function {name} not registered")
spec = self.functions[name].spec
return {
"name": spec.name,
"description": spec.description,
"parameters": spec.parameters
}
def get_all_schemas(self, category: str = None) -> List[Dict]:
"""Get all function schemas, optionally filtered by category."""
names = self.categories.get(category, []) if category else list(self.functions.keys())
return [self.get_schema(name) for name in names]
def call(self, name: str, arguments: Dict) -> Any:
"""Execute a registered function."""
if name not in self.functions:
raise ValueError(f"Function {name} not registered")
registered = self.functions[name]
# Validate arguments against schema
# (Would use jsonschema.validate in production)
# Execute
return registered.handler(**arguments)
async def acall(self, name: str, arguments: Dict) -> Any:
"""Async execute a registered function."""
if name not in self.functions:
raise ValueError(f"Function {name} not registered")
registered = self.functions[name]
if asyncio.iscoroutinefunction(registered.handler):
return await registered.handler(**arguments)
else:
# Run sync function in executor
loop = asyncio.get_event_loop()
return await loop.run_in_executor(None, registered.handler, **arguments)
# Global registry
registry = FunctionRegistry()
# Register built-in functions
registry.register(FunctionLibrary.execute_python, category="code")
registry.register(FunctionLibrary.http_request, category="network")
registry.register(FunctionLibrary.query_database, category="data")
registry.register(FunctionLibrary.search_web, category="research")
registry.register(FunctionLibrary.file_operation, category="filesystem")
Part 2: Agent Implementation
Core Agent Loop
# agents/function_agent.py
from typing: Dict, List, Any, Optional
from dataclasses import dataclass, field
from openai import AsyncOpenAI
import asyncio
import json
@dataclass
class FunctionCall:
name: str
arguments: Dict
call_id: str
@dataclass
class AgentResponse:
content: str
function_calls: List[FunctionCall] = field(default_factory=list)
finish_reason: str = "stop"
usage: Dict = field(default_factory=dict)
class FunctionCallingAgent:
def __init__(
self,
client: AsyncOpenAI,
model: str = "gpt-4o",
registry: FunctionRegistry = None,
system_prompt: str = None,
max_iterations: int = 10,
temperature: float = 0.1
):
self.client = client
self.model = model
self.registry = registry or FunctionRegistry()
self.system_prompt = system_prompt or self._default_system_prompt()
self.max_iterations = max_iterations
self.temperature = temperature
self.conversation_history: List[Dict] = []
def _default_system_prompt(self) -> str:
return """You are an AI agent with access to various tools.
Use tools when needed to answer questions or complete tasks.
Think step by step. If you need to use a tool, call it.
Always provide a final answer to the user."""
def _build_messages(self, user_message: str) -> List[Dict]:
messages = [{"role": "system", "content": self.system_prompt}]
messages.extend(self.conversation_history)
messages.append({"role": "user", "content": user_message})
return messages
def _get_function_schemas(self) -> List[Dict]:
return self.registry.get_all_schemas()
async def run(self, user_message: str) -> AgentResponse:
"""Run agent loop until completion."""
messages = self._build_messages(user_message)
for iteration in range(self.max_iterations):
# Call LLM with function definitions
response = await self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=[{"type": "function", "function": schema} for schema in self._get_function_schemas()],
tool_choice="auto",
temperature=self.temperature,
max_tokens=4000
)
message = response.choices[0].message
# Add assistant message to history
messages.append({
"role": "assistant",
"content": message.content,
"tool_calls": message.tool_calls
})
# Check for function calls
if not message.tool_calls:
# No function calls - we're done
self.conversation_history.append({"role": "user", "content": user_message})
self.conversation_history.append({"role": "assistant", "content": message.content})
return AgentResponse(
content=message.content or "",
finish_reason=response.choices[0].finish_reason,
usage=response.usage.model_dump() if response.usage else {}
)
# Execute function calls
function_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
)
function_calls.append(call)
# Execute function
try:
result = await self.registry.acall(call.name, call.arguments)
result_str = json.dumps(result, default=str)
except Exception as e:
result_str = json.dumps({"error": str(e)})
# Add function result to messages
messages.append({
"role": "tool",
"tool_call_id": call.call_id,
"content": result_str
})
# Continue loop for LLM to process results
# Max iterations reached
return AgentResponse(
content="Max iterations reached without completion.",
finish_reason="length",
function_calls=function_calls
)
def clear_history(self):
self.conversation_history = []
Streaming Agent
# agents/streaming_agent.py
from typing: AsyncGenerator, Dict, List, Any
from openai import AsyncOpenAI
import json
class StreamingFunctionAgent:
def __init__(self, client: AsyncOpenAI, model: str = "gpt-4o", registry=None):
self.client = client
self.model = model
self.registry = registry or FunctionRegistry()
async def run_stream(self, user_message: str) -> AsyncGenerator[str, None]:
"""Stream agent execution with function calls."""
messages = [
{"role": "system", "content": self._default_system_prompt()},
{"role": "user", "content": user_message}
]
while True:
stream = await self.client.chat.completions.create(
model=self.model,
messages=messages,
tools=[{"type": "function", "function": schema} for schema in self.registry.get_all_schemas()],
tool_choice="auto",
temperature=0.1,
stream=True
)
function_calls = []
current_content = ""
current_tool_call = None
async for chunk in stream:
delta = chunk.choices[0].delta
if delta.content:
current_content += delta.content
yield delta.content
if delta.tool_calls:
for tc in delta.tool_calls:
if tc.index >= len(function_calls):
function_calls.append({
"id": tc.id,
"name": tc.function.name,
"arguments": ""
})
current_tool_call = function_calls[-1]
else:
current_tool_call = function_calls[tc.index]
if tc.function.name:
current_tool_call["name"] = tc.function.name
if tc.function.arguments:
current_tool_call["arguments"] += tc.function.arguments
# Add assistant message
messages.append({
"role": "assistant",
"content": current_content,
"tool_calls": [
{"id": fc["id"], "type": "function", "function": {"name": fc["name"], "arguments": fc["arguments"]}}
for fc in function_calls
] if function_calls else None
})
if not function_calls:
yield "\n[Done]"
break
# Execute functions and yield results
for fc in function_calls:
yield f"\n[Calling {fc['name']}...]"
try:
result = await self.registry.acall(fc["name"], json.loads(fc["arguments"]))
result_str = json.dumps(result, default=str)
except Exception as e:
result_str = json.dumps({"error": str(e)})
messages.append({
"role": "tool",
"tool_call_id": fc["id"],
"content": result_str
})
yield f" [Result: {result_str[:100]}...]"
Part 3: Validation & Safety
Input Validation
# functions/validation.py
from typing: Dict, Any, List
from pydantic import BaseModel, ValidationError
import jsonschema
class FunctionValidator:
def __init__(self, registry: FunctionRegistry):
self.registry = registry
def validate_call(self, name: str, arguments: Dict) -> tuple[bool, List[str]]:
"""Validate function call arguments against schema."""
if name not in self.registry.functions:
return False, [f"Unknown function: {name}"]
spec = self.registry.functions[name].spec
schema = spec.parameters
try:
jsonschema.validate(arguments, schema)
return True, []
except jsonschema.ValidationError as e:
return False, [f"Validation error: {e.message} at {'.'.join(str(x) for x in e.path)}"]
def sanitize_arguments(self, name: str, arguments: Dict) -> Dict:
"""Sanitize arguments (remove extra fields, apply defaults)."""
if name not in self.registry.functions:
return arguments
spec = self.registry.functions[name].spec
properties = spec.parameters.get("properties", {})
# Keep only defined properties
sanitized = {}
for key, value in arguments.items():
if key in properties:
sanitized[key] = value
# Apply defaults for missing required fields
for req in spec.parameters.get("required", []):
if req not in sanitized and req in properties:
prop = properties[req]
if "default" in prop:
sanitized[req] = prop["default"]
return sanitized
Output Validation
# functions/output_validation.py
from typing: Dict, Any, Callable
from pydantic import BaseModel, validator
import json
class OutputValidator:
def __init__(self):
self.validators: Dict[str, Callable] = {}
def register_validator(self, function_name: str, validator_fn: Callable[[Any], bool]):
"""Register custom validator for function output."""
self.validators[function_name] = validator_fn
def validate_output(self, function_name: str, output: Any) -> tuple[bool, str]:
"""Validate function output."""
if function_name not in self.validators:
return True, ""
try:
result = self.validators[function_name](output)
if result:
return True, ""
return False, f"Output validation failed for {function_name}"
except Exception as e:
return False, f"Validation error: {e}"
# Example validators
def validate_code_execution(output: Dict) -> bool:
"""Validate code execution output."""
if not isinstance(output, dict):
return False
if "success" not in output:
return False
if output["success"] and "result" not in output:
return False
return True
def validate_http_response(output: Dict) -> bool:
"""Validate HTTP response."""
if not isinstance(output, dict):
return False
if "status_code" not in output:
return False
return True
validator = OutputValidator()
validator.register_validator("execute_python", validate_code_execution)
validator.register_validator("http_request", validate_http_response)
Security Sandbox
# functions/sandbox.py
"""Secure function execution sandbox."""
import subprocess
import tempfile
import os
import resource
import signal
from typing: Dict, Any
from dataclasses import dataclass
@dataclass
class SandboxConfig:
max_cpu_seconds: int = 30
max_memory_mb: int = 512
max_processes: int = 10
allowed_modules: List[str] = None
blocked_modules: List[str] = field(default_factory=lambda: [
"os", "sys", "subprocess", "shutil", "pathlib",
"importlib", "pkgutil", "modulefinder", "runpy",
"socket", "urllib", "http", "ftplib", "telnetlib",
"sqlite3", "pickle", "marshal", "shelve"
])
class PythonSandbox:
def __init__(self, config: SandboxConfig = None):
self.config = config or SandboxConfig()
def execute(self, code: str, timeout: int = 30) -> Dict:
"""Execute Python code in sandboxed subprocess."""
# Write code to temp file
with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
# Add safety wrapper
wrapped_code = self._wrap_code(code)
f.write(wrapped_code)
temp_path = f.name
try:
# Set resource limits
def set_limits():
resource.setrlimit(resource.RLIMIT_CPU, (self.config.max_cpu_seconds, self.config.max_cpu_seconds))
resource.setrlimit(resource.RLIMIT_AS, (self.config.max_memory_mb * 1024 * 1024, self.config.max_memory_mb * 1024 * 1024))
resource.setrlimit(resource.RLIMIT_NPROC, (self.config.max_processes, self.config.max_processes))
# Run with timeout
result = subprocess.run(
["python", temp_path],
capture_output=True,
text=True,
timeout=timeout,
preexec_fn=set_limits if os.name != 'nt' else None
)
return {
"success": result.returncode == 0,
"stdout": result.stdout,
"stderr": result.stderr,
"return_code": result.returncode
}
except subprocess.TimeoutExpired:
return {"success": False, "error": "Execution timeout"}
except Exception as e:
return {"success": False, "error": str(e)}
finally:
try:
os.unlink(temp_path)
except:
pass
def _wrap_code(self, code: str) -> str:
"""Wrap user code with safety checks."""
return f"""
import sys
import json
# Redirect stdout/stderr
import io
old_stdout = sys.stdout
old_stderr = sys.stderr
sys.stdout = io.StringIO()
sys.stderr = io.StringIO()
result = {{"success": False, "result": None, "stdout": "", "stderr": ""}}
try:
{self._indent(code, 8)}
result["success"] = True
result["result"] = locals().get("result", None)
except Exception as e:
result["success"] = False
result["error"] = str(e)
import traceback
result["traceback"] = traceback.format_exc()
finally:
result["stdout"] = sys.stdout.getvalue()
result["stderr"] = sys.stderr.getvalue()
sys.stdout = old_stdout
sys.stderr = old_stderr
print(json.dumps(result))
"""
def _indent(self, text: str, spaces: int) -> str:
return "\n".join(" " * spaces + line for line in text.split("\n"))
Part 4: Advanced Patterns
Parallel Function Execution
# functions/parallel.py
from typing: List, Dict, Any
import asyncio
class ParallelFunctionExecutor:
def __init__(self, registry: FunctionRegistry, max_concurrent: int = 5):
self.registry = registry
self.semaphore = asyncio.Semaphore(max_concurrent)
async def execute_batch(self, calls: List[Dict]) -> List[Any]:
"""Execute multiple function calls in parallel."""
async def execute_one(call):
async with self.semaphore:
try:
return await self.registry.acall(call["name"], call["arguments"])
except Exception as e:
return {"error": str(e), "call": call}
tasks = [execute_one(call) for call in calls]
return await asyncio.gather(*tasks)
async def execute_with_dependencies(self, calls: List[Dict]) -> List[Any]:
"""Execute calls respecting dependencies."""
# calls format: {"name": "", "arguments": {}, "depends_on": [call_ids]}
completed = {}
remaining = {call.get("id", i): call for i, call in enumerate(calls)}
while remaining:
# Find callable tasks (no unmet dependencies)
callable_tasks = []
for call_id, call in remaining.items():
deps = call.get("depends_on", [])
if all(d in completed for d in deps):
callable_tasks.append((call_id, call))
if not callable_tasks:
# Circular dependency or missing dependency
raise ValueError("Unresolvable dependencies")
# Execute callable tasks in parallel
tasks = []
for call_id, call in callable_tasks:
task = self._execute_with_deps(call_id, call, completed)
tasks.append(task)
results = await asyncio.gather(*tasks)
for (call_id, _), result in zip(callable_tasks, results):
completed[call_id] = result
del remaining[call_id]
return [completed.get(call.get("id", i)) for i, call in enumerate(calls)]
async def _execute_with_deps(self, call_id: str, call: Dict, completed: Dict) -> Any:
# Substitute dependency results into arguments
args = call["arguments"].copy()
for key, value in args.items():
if isinstance(value, str) and value.startswith("$ref:"):
dep_id = value[5:]
if dep_id in completed:
args[key] = completed[dep_id]
async with self.semaphore:
return await self.registry.acall(call["name"], args)
Function Composition
# functions/composition.py
from typing: Dict, Any, List, Callable
from dataclasses import dataclass
@dataclass
class FunctionPipeline:
"""Chain functions together: output of one feeds into next."""
steps: List[Dict] # {"name": "", "arguments": {}, "output_to": "next_arg"}
class FunctionComposer:
def __init__(self, registry: FunctionRegistry):
self.registry = registry
async def compose(self, pipeline: FunctionPipeline, initial_input: Dict) -> Dict:
"""Execute function pipeline."""
context = initial_input.copy()
for step in pipeline.steps:
# Merge context into arguments
args = {**step.get("arguments", {}), **context}
# Execute
result = await self.registry.acall(step["name"], args)
# Store result for next steps
if "output_to" in step:
context[step["output_to"]] = result
else:
context[step["name"]] = result
context["last_result"] = result
return context
# Example: Research -> Summarize -> Save
research_pipeline = FunctionPipeline(steps=[
{"name": "search_web", "arguments": {"query": "AI agents 2024"}, "output_to": "search_results"},
{"name": "execute_python", "arguments": {
"code": """
# Process search results
result = [r for r in search_results if '2024' in r.get('snippet', '')]
"""
}, "output_to": "filtered_results"},
{"name": "file_operation", "arguments": {
"operation": "write",
"path": "research_summary.json",
"content": "{{filtered_results}}"
}}
])
Part 5: Observability
Function Call Tracing
# functions/observability.py
from dataclasses import dataclass, field
from typing: Dict, Any, List
from datetime import datetime
import time
import uuid
@dataclass
class FunctionTrace:
trace_id: str
function_name: str
arguments: Dict
start_time: float
end_time: float = 0
duration_ms: float = 0
success: bool = False
result: Any = None
error: str = ""
metadata: Dict = field(default_factory=dict)
class FunctionTracer:
def __init__(self):
self.traces: List[FunctionTrace] = []
self.active_traces: Dict[str, FunctionTrace] = {}
def start_trace(self, function_name: str, arguments: Dict, trace_id: str = None) -> str:
trace_id = trace_id or str(uuid.uuid4())[:8]
trace = FunctionTrace(
trace_id=trace_id,
function_name=function_name,
arguments=arguments,
start_time=time.time()
)
self.active_traces[trace_id] = trace
return trace_id
def end_trace(self, trace_id: str, success: bool, result: Any = None, error: str = ""):
if trace_id not in self.active_traces:
return
trace = self.active_traces[trace_id]
trace.end_time = time.time()
trace.duration_ms = (trace.end_time - trace.start_time) * 1000
trace.success = success
trace.result = result
trace.error = error
self.traces.append(trace)
del self.active_traces[trace_id]
def get_stats(self, function_name: str = None, window_seconds: int = 300) -> Dict:
cutoff = time.time() - window_seconds
recent = [t for t in self.traces if t.end_time > cutoff]
if function_name:
recent = [t for t in recent if t.function_name == function_name]
if not recent:
return {}
durations = [t.duration_ms for t in recent]
successes = [t for t in recent if t.success]
return {
"function": function_name or "all",
"total_calls": len(recent),
"success_rate": len(successes) / len(recent),
"avg_duration_ms": sum(durations) / len(durations),
"p50_duration_ms": sorted(durations)[len(durations) // 2],
"p95_duration_ms": sorted(durations)[int(len(durations) * 0.95)],
"p99_duration_ms": sorted(durations)[int(len(durations) * 0.99)],
"errors": [t.error for t in recent if not t.success]
}
# Global tracer
tracer = FunctionTracer()
# Decorator for automatic tracing
def trace_function(func_name: str = None):
def decorator(handler):
async def wrapper(*args, **kwargs):
name = func_name or handler.__name__
trace_id = tracer.start_trace(name, {"args": str(args), "kwargs": str(kwargs)})
try:
result = await handler(*args, **kwargs)
tracer.end_trace(trace_id, True, result)
return result
except Exception as e:
tracer.end_trace(trace_id, False, error=str(e))
raise
return wrapper
return decorator
Part 6: Testing
Unit Testing Functions
# tests/test_functions.py
import pytest
from functions.registry import FunctionRegistry, registry
from functions.schemas import function_to_schema
from functions.validation import FunctionValidator
class TestFunctionRegistry:
def test_register_function(self):
reg = FunctionRegistry()
def add(a: int, b: int) -> int:
"""Add two numbers."""
return a + b
reg.register(add, category="math")
assert "add" in reg.functions
schema = reg.get_schema("add")
assert schema["name"] == "add"
assert "a" in schema["parameters"]["properties"]
assert "b" in schema["parameters"]["properties"]
def test_call_function(self):
reg = FunctionRegistry()
def multiply(x: int, y: int) -> int:
return x * y
reg.register(multiply)
result = reg.call("multiply", {"x": 5, "y": 3})
assert result == 15
class TestFunctionValidator:
def test_valid_arguments(self):
validator = FunctionValidator(registry)
valid, errors = validator.validate_call("execute_python", {"code": "print('hello')", "timeout": 30})
assert valid
assert len(errors) == 0
def test_invalid_arguments(self):
validator = FunctionValidator(registry)
valid, errors = validator.validate_call("execute_python", {"code": 123}) # wrong type
assert not valid
assert len(errors) > 0
def test_sanitize_arguments(self):
validator = FunctionValidator(registry)
sanitized = validator.sanitize_arguments("execute_python", {"code": "x=1", "timeout": 30, "extra": "ignored"})
assert "code" in sanitized
assert "timeout" in sanitized
assert "extra" not in sanitized
class TestSandbox:
def test_safe_execution(self):
from functions.sandbox import PythonSandbox
sandbox = PythonSandbox()
result = sandbox.execute("result = 2 + 2")
assert result["success"]
assert result["result"] == 4
def test_blocked_modules(self):
from functions.sandbox import PythonSandbox
sandbox = PythonSandbox()
result = sandbox.execute("import os; os.system('ls')")
assert not result["success"]
assert "error" in result or "blocked" in result.get("stderr", "").lower()
Part 7: Production Deployment
Kubernetes Deployment
# k8s/function-agent.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
name: function-calling-agent
spec:
replicas: 3
selector:
matchLabels:
app: function-agent
template:
metadata:
labels:
app: function-agent
spec:
containers:
- name: agent
image: myorg/function-agent:v1.0
env:
- name: OPENAI_API_KEY
valueFrom:
secretKeyRef:
name: ai-secrets
key: openai-key
- name: FUNCTION_REGISTRY
value: "builtin"
- name: SANDBOX_ENABLED
value: "true"
- name: MAX_ITERATIONS
value: "10"
resources:
requests:
memory: "2Gi"
cpu: "1000m"
limits:
memory: "4Gi"
cpu: "2000m"
livenessProbe:
httpGet:
path: /health
port: 8000
initialDelaySeconds: 30
readinessProbe:
httpGet:
path: /ready
port: 8000
initialDelaySeconds: 10
---
apiVersion: v1
kind: Service
metadata:
name: function-agent
spec:
selector:
app: function-agent
ports:
- port: 80
targetPort: 8000
FastAPI Server
# server.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing: List, Optional
import asyncio
app = FastAPI(title="Function Calling Agent API")
class ChatRequest(BaseModel):
message: str
conversation_id: Optional[str] = None
model: Optional[str] = "gpt-4o"
temperature: Optional[float] = 0.1
max_iterations: Optional[int] = 10
class ChatResponse(BaseModel):
response: str
conversation_id: str
function_calls: List[Dict] = []
usage: Dict = {}
# Initialize agent
agent = FunctionCallingAgent(
client=AsyncOpenAI(api_key=os.getenv("OPENAI_API_KEY")),
model="gpt-4o"
)
@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
try:
result = await agent.run(request.message)
return ChatResponse(
response=result.content,
conversation_id=request.conversation_id or str(uuid.uuid4()),
function_calls=[{"name": fc.name, "arguments": fc.arguments} for fc in result.function_calls],
usage=result.usage
)
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.post("/chat/stream")
async def chat_stream(request: ChatRequest):
from fastapi.responses import StreamingResponse
async def generate():
streaming_agent = StreamingFunctionAgent(
client=agent.client,
model=request.model,
registry=agent.registry
)
async for chunk in streaming_agent.run_stream(request.message):
yield f"data: {json.dumps({'chunk': chunk})}\n\n"
return StreamingResponse(generate(), media_type="text/event-stream")
@app.get("/functions")
async def list_functions(category: Optional[str] = None):
return {"functions": agent.registry.get_all_schemas(category)}
@app.get("/health")
async def health():
return {"status": "healthy"}
@app.get("/metrics")
async def metrics():
return {"traces": tracer.get_stats()}
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
Part 8: Best Practices
Function Design Guidelines
FUNCTION DESIGN PRINCIPLES:
1. SINGLE RESPONSIBILITY
- Each function does ONE thing well
- Clear, descriptive name
- Minimal parameters
2. IDEMPOTENCY
- Same inputs = same outputs
- Safe to retry
- No side effects for read operations
3. EXPLICIT CONTRACTS
- Full JSON Schema with descriptions
- Document all error cases
- Version your functions
4. FAILURE HANDLING
- Return structured errors
- Include retry guidance
- Log for debugging
5. SECURITY
- Validate all inputs
- Sandbox code execution
- Rate limit sensitive operations
- Audit logging
6. OBSERVABILITY
- Trace every call
- Metrics on latency, errors
- Alert on anomalies
Anti-Patterns to Avoid
❌ OVERLOADING FUNCTIONS
- One function with 20 parameters
- Multiple operations via "mode" parameter
❌ IMPLICIT STATE
- Functions that depend on hidden context
- Global variables between calls
❌ UNBOUNDED EXECUTION
- No timeout on external calls
- No rate limiting
- No circuit breakers
❌ POOR ERROR HANDLING
- Generic "error" responses
- No retry guidance
- Swallowing exceptions
❌ MISSING VALIDATION
- Trusting LLM arguments blindly
- No schema validation
- No output verification
Resources
Function calling is the bridge between LLMs and the real world. Design your functions like APIs: clear contracts, robust validation, comprehensive observability. The agent is only as reliable as its weakest function.
Top comments (0)