Pydantic AI Middleware

Intercept, transform, and guard every AI call

Lightweight middleware library with 7 lifecycle hooks for Pydantic AI. Supports parallel execution, async guardrails, cost tracking, and permission decisions.

Installation

pip install pydantic-ai-middleware

Pydantic AI Middleware provides 7 lifecycle hooks that let you intercept every stage of an agent run: before_run, after_run, before_model_request, before_tool_call, after_tool_call, on_tool_error, and on_run_error. Multiple middleware hooks execute in parallel with configurable aggregation strategies (first_response, all_responses, majority_vote). Async guardrails support three timing modes — BLOCKING (wait before proceeding), CONCURRENT (run alongside the agent), and ASYNC_POST (fire-and-forget after completion).

Features

Quick Start

from pydantic_ai import Agent
from pydantic_ai_middleware import MiddlewareAgent, AgentMiddleware

class CostTracker(AgentMiddleware[None]):
    """Track token usage and costs across all agent runs."""

    async def after_run(self, prompt, output, deps, ctx=None):
        print(f"Tokens used: {ctx.usage.total_tokens}")
        return output

base_agent = Agent("openai:gpt-4o", instructions="You are a helpful assistant.")
agent = MiddlewareAgent(agent=base_agent, middleware=[CostTracker()])
result = await agent.run("Hello, how are you?")

Use Cases

Hook Lifecycle

flowchart LR
    Input[Input] --> before_run[before_run]
    before_run --> before_model[before_model_request]
    before_model --> Agent[Agent]
    Agent -->|tool call| before_tool[before_tool_call]
    before_tool --> Tool[Tool]
    Tool --> after_tool[after_tool_call]
    after_tool --> Agent
    Tool -.->|error| on_error[on_tool_error]
    on_error -.-> Agent
    Agent -->|finish| after_run[after_run]
    after_run --> Output[Output]