deep-agent-middleware

Deep Agent Middleware

Add cross-cutting concerns to Deep Agents with built-in and custom middleware for logging, validation, rate limiting, and tool interception.

When to Use

• When you need to log all tool calls
• When you need to validate inputs before tools execute
• When you need rate limiting on API calls
• When you need custom behavior around tool execution

Available Operations

1. Use built-in middleware (TodoList, Filesystem, SubAgent, Summarization)
2. Create custom middleware with @wrap_tool_call
3. Chain multiple middleware
4. Add conditional middleware

Multi-Step Workflow

Step 1: Understand Built-in Middleware

These are automatically active:

Middleware	What it does	Can disable?
TodoListMiddleware	write_todos tool for planning	No
FilesystemMiddleware	ls, read_file, write_file, etc.	No
SubAgentMiddleware	task tool for delegation	No
SummarizationMiddleware	Compresses long conversations	No
AnthropicPromptCachingMiddleware	Token optimization for Anthropic	Auto
PatchToolCallsMiddleware	Repairs broken tool calls	Auto

Step 2: Create Custom Middleware

# agent/middleware.py
from langchain.agents.middleware import wrap_tool_call
import time
import logging

logger = logging.getLogger(__name__)

@wrap_tool_call
def log_tool_calls(request, handler):
    """Log every tool call with timing."""
    start = time.time()
    logger.info(f"[TOOL] {request.name} called with: {request.args}")

    result = handler(request)

    duration = time.time() - start
    logger.info(f"[TOOL] {request.name} completed in {duration:.2f}s")
    return result

@wrap_tool_call
def validate_file_paths(request, handler):
    """Prevent access to sensitive files."""
    if request.name in ("read_file", "write_file", "edit_file"):
        path = request.args.get("path", "")
        blocked = [".env", "secrets", "credentials", ".git/config"]
        if any(b in path for b in blocked):
            return f"Access denied: {path} is a protected file"

    return handler(request)

@wrap_tool_call
def rate_limit_api_calls(request, handler):
    """Rate limit external API calls."""
    api_tools = ["internet_search", "call_api"]
    if request.name in api_tools:
        time.sleep(1)  # Simple rate limit

    return handler(request)