USER REQUEST │ ▼ ┌─────────────┐ │ LLM │ ──▶ Analyzes request, decides function calls │ Reasoning │ └─────────────┘ │ ▼ (if function needed) ┌─────────────┐ │ Function │ ──▶ Validates, executes, returns result │ Executor │ └─────────────┘ │ ▼ ┌─────────────┐ │ LLM │ ──▶ Synthesizes final response │ Response │ └─────────────┘ 

# 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 

# 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") 

# 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.nameSource：https://dev.to/ipsphil_digital/building-ai-agents-with-function-calling-tools-patterns-production-systems-3j5i
My favorites