Python Decorators for AI Agents: Building Reusable Tool and Middleware Patterns

Decorators as Agent Middleware

In web frameworks, middleware wraps every request with cross-cutting logic like authentication, logging, and rate limiting. AI agent frameworks need the same patterns for tool calls. Python decorators provide exactly this — they wrap functions with reusable behavior without modifying the function itself.

Every major AI framework uses decorators extensively. The OpenAI Agents SDK uses them for tool registration. LangChain uses them for chain composition. FastAPI uses them for route definition. Mastering decorators lets you build clean, composable agent architectures.

Decorator Fundamentals

A decorator is a function that takes a function and returns a new function. The @ syntax is syntactic sugar.

import functools
import time
from typing import Callable, TypeVar, ParamSpec

P = ParamSpec("P")
T = TypeVar("T")

def log_tool_call(func: Callable[P, T]) -> Callable[P, T]:
    @functools.wraps(func)
    def wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
        print(f"[TOOL] Calling {func.__name__} with {kwargs}")
        start = time.perf_counter()
        result = func(*args, **kwargs)
        elapsed = time.perf_counter() - start
        print(f"[TOOL] {func.__name__} completed in {elapsed:.3f}s")
        return result
    return wrapper

@log_tool_call
def web_search(query: str) -> str:
    # actual search implementation
    return f"Results for: {query}"

Always use functools.wraps. Without it, the decorated function loses its name, docstring, and type hints — which breaks tool registration in agent frameworks that inspect function metadata.

Async Decorators for Agent Tools

Most AI agent tools are async. Your decorators must handle both sync and async functions.

import asyncio
import functools
from typing import Callable, Any

def retry(max_attempts: int = 3, delay: float = 1.0):
    def decorator(func: Callable) -> Callable:
        @functools.wraps(func)
        async def async_wrapper(*args, **kwargs) -> Any:
            last_error = None
            for attempt in range(1, max_attempts + 1):
                try:
                    return await func(*args, **kwargs)
                except Exception as e:
                    last_error = e
                    if attempt < max_attempts:
                        wait = delay * (2 ** (attempt - 1))
                        print(f"Attempt {attempt} failed, retrying in {wait}s")
                        await asyncio.sleep(wait)
            raise last_error

        @functools.wraps(func)
        def sync_wrapper(*args, **kwargs) -> Any:
            last_error = None
            for attempt in range(1, max_attempts + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    last_error = e
                    if attempt < max_attempts:
                        wait = delay * (2 ** (attempt - 1))
                        time.sleep(wait)
            raise last_error

        if asyncio.iscoroutinefunction(func):
            return async_wrapper
        return sync_wrapper
    return decorator

@retry(max_attempts=3, delay=0.5)
async def call_llm(prompt: str) -> str:
    # API call that might fail
    pass

Rate Limiting Decorator

When your agent calls external APIs, rate limiting prevents quota exhaustion.

import asyncio
import functools
import time
from collections import deque

def rate_limit(calls_per_minute: int = 60):
    timestamps: deque = deque()

    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            now = time.monotonic()
            # Remove timestamps older than 60 seconds
            while timestamps and now - timestamps[0] > 60:
                timestamps.popleft()

            if len(timestamps) >= calls_per_minute:
                sleep_time = 60 - (now - timestamps[0])
                await asyncio.sleep(sleep_time)

            timestamps.append(time.monotonic())
            return await func(*args, **kwargs)
        return wrapper
    return decorator

@rate_limit(calls_per_minute=20)
async def embed_text(text: str) -> list[float]:
    # call embedding API
    pass

Tool Registration Decorator

Build a decorator that automatically registers functions as agent tools with metadata extracted from type hints and docstrings.

import functools
import inspect
from typing import get_type_hints

TOOL_REGISTRY: dict[str, dict] = {}

def agent_tool(name: str = None, description: str = None):
    def decorator(func):
        tool_name = name or func.__name__
        tool_desc = description or func.__doc__ or "No description"
        hints = get_type_hints(func)

        params = {}
        sig = inspect.signature(func)
        for param_name, param in sig.parameters.items():
            param_type = hints.get(param_name, str).__name__
            params[param_name] = {"type": param_type}

        TOOL_REGISTRY[tool_name] = {
            "function": func,
            "description": tool_desc,
            "parameters": params,
        }

        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            return await func(*args, **kwargs)
        return wrapper
    return decorator

@agent_tool(name="search", description="Search the web")
async def web_search(query: str, max_results: int = 5) -> str:
    return f"Found {max_results} results for {query}"

FAQ

Why does functools.wraps matter for AI agent tools?

Agent frameworks like the OpenAI Agents SDK inspect function metadata — the name, docstring, and type annotations — to generate tool definitions for the LLM. Without functools.wraps, the decorator replaces this metadata with the wrapper's metadata, causing the LLM to see incorrect tool names and missing descriptions.

Can I stack multiple decorators on one tool function?

Yes. Decorators apply bottom-up, so @retry @rate_limit @log_tool_call def func means the call passes through log_tool_call first, then rate_limit, then retry wraps the entire chain. Order matters — put retry outermost so it retries the entire decorated pipeline.

How do I test decorated functions in isolation?

Access the original function via func.__wrapped__ (available when you use functools.wraps). This lets you unit test the core logic without triggering retry delays, rate limits, or logging side effects.

---

#Python #Decorators #DesignPatterns #AIAgents #AgenticAI #LearnAI #AIEngineering