Building AI Agents with Function Calling: Tools, Patterns & Production Systems

Build production-ready AI agents that can execute code, call APIs, query databases, and interact with external systems through function calling.

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.

USER REQUEST │ ▼ ┌─────────────┐ │ LLM │ ──▶ Analyzes request, decides function calls │ Reasoning │ └─────────────┘ │ ▼ (if function needed) ┌─────────────┐ │ Function │ ──▶ Validates, executes, returns result │ Executor │ └─────────────┘ │ ▼ ┌─────────────┐ │ LLM │ ──▶ Synthesizes final response │ Response │ └─────────────┘ Enter fullscreen mode Exit fullscreen mode 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 Enter fullscreen mode Exit fullscreen mode 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 " ) Enter fullscreen mode Exit fullscreen mode 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 = [] Enter fullscreen mode Exit fullscreen mode 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 ] }...] " Enter fullscreen mode Exit fullscreen mode 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 Enter fullscreen mode Exit fullscreen mode 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 ) Enter fullscreen mode Exit fullscreen mode 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 " )) Enter fullscreen mode Exit fullscreen mode 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 ) Enter fullscreen mode Exit fullscreen mode 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}} " }} ]) Enter fullscreen mode Exit fullscreen mode 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 Enter fullscreen mode Exit fullscreen mode 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 () Enter fullscreen mode Exit fullscreen mode 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 Enter fullscreen mode Exit fullscreen mode 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 ) Enter fullscreen mode Exit fullscreen mode 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 Enter fullscreen mode Exit fullscreen mode 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 Enter fullscreen mode Exit fullscreen mode Resources OpenAI Function Calling Anthropic Tool Use JSON Schema Pydantic jsonschema Structured Outputs 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.