### Bash: Running Standalone Example Scripts

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/examples/index.md

Provides command-line instructions for running standalone example scripts located in the `examples/` directory using the `uv` command. These scripts demonstrate specific functionalities like PII detection, prompt injection screening, toxicity checking, and JSON schema validation. This requires the `uv` tool to be installed and the respective example Python files to be present.

```bash
uv run python examples/pii_detection.py
uv run python examples/prompt_injection.py
uv run python examples/toxicity_check.py
uv run python examples/json_schema_validation.py
```

--------------------------------

### Install pydantic-ai-middleware from source

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/installation.md

Installs the pydantic-ai-middleware package by cloning the repository and installing it in editable mode. This requires Git and pip. Ensure you have Python 3.10+ and pydantic-ai-slim >= 1.39.

```bash
git clone https://github.com/vstorm-co/pydantic-ai-middleware.git
cd pydantic-ai-middleware
pip install -e .
```

--------------------------------

### Install pydantic-ai-middleware with uv

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/installation.md

Installs the pydantic-ai-middleware package using uv, a fast Python package installer and dependency manager. Ensure you have Python 3.10+ and pydantic-ai-slim >= 1.39 installed.

```bash
uv add pydantic-ai-middleware
```

--------------------------------

### Verify pydantic-ai-middleware Installation

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/installation.md

A simple Python script to verify that the pydantic-ai-middleware package has been installed correctly. It imports necessary components and prints a success message.

```python
from pydantic_ai_middleware import MiddlewareAgent, AgentMiddleware

print("Installation successful!")
```

--------------------------------

### Install pydantic-ai-middleware

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/AGENTS.md

Installs the pydantic-ai-middleware library using the provided Makefile. This is a common first step for setting up the project.

```bash
make install
```

--------------------------------

### Run PII Detection Example using uv in Bash

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/examples/pii-detection.md

Provides the command to execute a runnable demo of the PII detection middleware. This example is typically found in the `examples/pii_detection.py` file and demonstrates the middleware's functionality in a complete application context.

```bash
uv run python examples/pii_detection.py

```

--------------------------------

### Complete Lifecycle Trace with All Hooks (Python)

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/examples/hooks_showcase/presentation.html

This example demonstrates a comprehensive trace of the AI processing lifecycle by utilizing all available hooks. This is useful for understanding the flow of execution from start to finish, including prompt handling, tool interactions, and error management. It requires implementing handlers for each hook.

```python
from pydantic_ai_middleware import before_run, after_run, before_model_request, after_model_request, before_tool_call, after_tool_call, on_tool_error, on_error, tool_blocking, permission_handler

@before_run
def trace_start(prompt: str):
    print(f"Pipeline started with prompt: {prompt}")

@before_model_request
def trace_model_request():
    print("Before LLM model request")

@after_model_request
def trace_model_response(response: any):
    print("After LLM model response")

@before_tool_call
def trace_tool_call(tool_name: str, tool_args: dict):
    print(f"Before tool call: {tool_name}")

@after_tool_call
def trace_tool_result(tool_name: str, tool_args: dict, result: any, duration: float):
    print(f"After tool call: {tool_name}")

@on_tool_error
def trace_tool_error(tool_name: str, tool_args: dict, error: Exception):
    print(f"Tool error: {tool_name} failed")

@on_error
def trace_general_error(error: Exception):
    print(f"General error: {error}")

@tool_blocking
def trace_tool_blocking() -> ToolPermissionResult:
    print("Checking tool blocking")
    return ToolPermissionResult.ALLOW

@permission_handler
def trace_permission_handler() -> ToolDecision:
    print("Handling permission")
    return ToolDecision.ALLOW

@after_run
def trace_end():
    print("Pipeline finished")
```

--------------------------------

### Install pydantic-ai-middleware with pip

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/installation.md

Installs the pydantic-ai-middleware package using pip, the standard Python package installer. Ensure you have Python 3.10+ and pydantic-ai-slim >= 1.39 installed.

```bash
pip install pydantic-ai-middleware
```

--------------------------------

### Async Callback for Cost Persistence

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/examples/cost-tracking.md

Demonstrates using an asynchronous callback function to persist cost data to a database after each agent run. This example simulates a database call.

```python
from pydantic_ai import Agent
from pydantic_ai.models.test import TestModel
from pydantic_ai_middleware import MiddlewareAgent, MiddlewareContext
from pydantic_ai_middleware.cost_tracking import (
    CostInfo,
    create_cost_tracking_middleware,
)


async def persist_cost(info: CostInfo) -> None:
    """Save cost data to a database."""
    # Replace with your actual database call
    print(
        f"Saving: run={info.run_count}, "
        f"tokens_in={info.run_request_tokens}, "
        f"tokens_out={info.run_response_tokens}, "
        f"cost={info.run_cost_usd}"
    )


cost_mw = create_cost_tracking_middleware(
    model_name="openai:gpt-4.1",
    on_cost_update=persist_cost,
)

agent = MiddlewareAgent(
    agent=Agent(model=TestModel()),
    middleware=[cost_mw],
    context=MiddlewareContext(),
)

result = await agent.run("Generate a summary")
```

--------------------------------

### Example YAML Pipeline Configuration

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/advanced/config-loading.md

Illustrates a middleware pipeline defined in YAML format. This example shows a similar structure to the JSON example, including logging and parallel execution of middleware components.

```yaml
- type: logging
- parallel:
    middleware:
      - type: pii_guard
      - type: profanity_filter
    strategy: all_must_pass
```

--------------------------------

### Cost Tracking Middleware - Quick Start

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/advanced/cost-tracking.md

Demonstrates how to initialize and use the CostTrackingMiddleware with an Agent, including setting a budget limit and a callback for cost updates.

```APIDOC
## POST /agent/run (Example Usage)

### Description
This example shows how to integrate `CostTrackingMiddleware` into an `Agent` to track token usage and costs, with a budget limit and a callback function.

### Method
POST

### Endpoint
`/agent/run` (Conceptual - actual execution is within the agent framework)

### Parameters
#### Path Parameters
None

#### Query Parameters
None

#### Request Body
None (This is a conceptual example of middleware integration, not a direct API call)

### Request Example
```python
from pydantic_ai import Agent
from pydantic_ai.models.test import TestModel
from pydantic_ai_middleware import MiddlewareAgent, MiddlewareContext
from pydantic_ai_middleware.cost_tracking import create_cost_tracking_middleware

async def main():
    cost_mw = create_cost_tracking_middleware(
        model_name="openai:gpt-4.1",
        budget_limit_usd=5.0,
        on_cost_update=lambda info: print(
            f"Run #{info.run_count}: ${info.run_cost_usd:.4f} "
            f"(total: ${info.total_cost_usd:.4f})"
        ),
    )

    agent = MiddlewareAgent(
        agent=Agent(model=TestModel()),
        middleware=[cost_mw],
        context=MiddlewareContext(),
    )

    result = await agent.run("Summarize this document")
    print(result)

# To run this example:
# import asyncio
# asyncio.run(main())
```

### Response
#### Success Response (200)
- **result** (any) - The result of the agent's run.

#### Response Example
```json
{
  "example": "The summarized document content..."
}
```

### Error Handling
- **BudgetExceededError**: Raised if the `budget_limit_usd` is reached.
```

--------------------------------

### Install Pydantic AI Middleware with YAML Support (Bash)

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/index.md

This command installs the `pydantic-ai-middleware` package along with the necessary dependencies for YAML configuration file support. This allows you to define and load middleware pipelines from YAML files.

```bash
pip install pydantic-ai-middleware[yaml]

```

--------------------------------

### Test Structure Example (Python)

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/CONTRIBUTING.md

Demonstrates the expected structure for writing tests in the pydantic-ai-middleware project. It includes examples for basic and asynchronous functionality using pytest.

```python
class TestFeatureName:
    """Tests for FeatureName."""

    def test_basic_functionality(self) -> None:
        """Test basic functionality."""
        ...

    async def test_async_functionality(self) -> None:
        """Test async functionality."""
        ...
```

--------------------------------

### ConditionalMiddleware Setup (Python)

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/api/conditional.md

Demonstrates how to initialize ConditionalMiddleware with a condition function and two different middleware instances. The `is_tool_hook` function determines which middleware (`ToolAuditMiddleware` or `GeneralAuditMiddleware`) is executed based on the hook type in the context.

```python
from pydantic_ai_middleware import ConditionalMiddleware
from pydantic_ai_middleware.context import HookType, ScopedContext

def is_tool_hook(ctx: ScopedContext | None) -> bool:
    """Route tool-related hooks to specialized middleware."""
    if ctx is None:
        return False
    return ctx.current_hook in (HookType.BEFORE_TOOL_CALL, HookType.AFTER_TOOL_CALL)

middleware = ConditionalMiddleware(
    condition=is_tool_hook,
    when_true=ToolAuditMiddleware(),
    when_false=GeneralAuditMiddleware(),
)
```

--------------------------------

### before_run Hook Implementation

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/concepts/hooks.md

Example implementation of the 'before_run' hook, which is executed before the agent starts. It can optionally modify the prompt that will be processed.

```python
async def before_run(
    self,
    prompt: str | Sequence[Any],
    deps: DepsT | None,
    ctx: ScopedContext | None = None,
) -> str | Sequence[Any]:
    # Return modified prompt
    return prompt
```

--------------------------------

### Python: Basic Agent with Logging Middleware

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/examples/index.md

Demonstrates how to create a basic AI agent using pydantic-ai and wrap it with a custom logging middleware. The `SimpleLogger` middleware intercepts agent calls before and after execution to print the prompt and output, showcasing a simple before/after hook mechanism. This example requires the `pydantic-ai` and `pydantic-ai-middleware` libraries.

```python
from pydantic_ai import Agent
from pydantic_ai_middleware import MiddlewareAgent, AgentMiddleware

class SimpleLogger(AgentMiddleware[None]):
    async def before_run(self, prompt, deps, ctx):
        print(f">> {prompt}")
        return prompt

    async def after_run(self, prompt, output, deps, ctx):
        print(f"<< {output}")
        return output

agent = MiddlewareAgent(
    agent=Agent('openai:gpt-4o'),
    middleware=[SimpleLogger()],
)

result = await agent.run("Hello!")
# >> Hello!
# << Hi there! How can I help you?
```

--------------------------------

### Implement 'before_run' Hook for Logging

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/examples/hooks_showcase/presentation.html

Shows how to implement the 'before_run' hook to log the incoming user prompt before the agent processes it. This example logs the prompt and returns it unmodified.

```python
from pydantic_ai import Agent
from pydantic_ai_middleware import AgentMiddleware, MiddlewareAgent

class BeforeRunLogger(AgentMiddleware[None]):

    async def before_run(self, prompt, deps, ctx):
        print(f"[before_run] {prompt!r}")
        return prompt

agent = Agent("openai:gpt-4.1")
mw_agent = MiddlewareAgent(
    agent=agent,
    middleware=[BeforeRunLogger()],
)

result = await mw_agent.run(
    "What is the capital of France?"
)
```

--------------------------------

### Python: Combining Multiple Middleware Components

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/examples/index.md

Illustrates how to combine several middleware components, such as Rate Limiting, Logging, Security, and Metrics, into a single `MiddlewareAgent`. This approach allows for modular handling of different concerns within an AI application, promoting cleaner and more maintainable code. The example assumes the existence of `RateLimitMiddleware`, `LoggingMiddleware`, `SecurityMiddleware`, and `MetricsMiddleware` classes.

```python
agent = MiddlewareAgent(
    agent=base_agent,
    middleware=[
        RateLimitMiddleware(max_calls=10, window=60),
        LoggingMiddleware(),
        SecurityMiddleware(),
        MetricsMiddleware(),
    ],
)
```

--------------------------------

### Install Pydantic AI Middleware (Bash)

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/index.md

This command installs the `pydantic-ai-middleware` package using pip. It is the standard way to add the library to your Python project's dependencies.

```bash
pip install pydantic-ai-middleware

```

--------------------------------

### Install and Test pydantic-ai-middleware

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/README.md

Provides bash commands to clone the repository, install dependencies, and run tests. It emphasizes the requirement for 100% test coverage for contributions.

```bash
git clone https://github.com/vstorm-co/pydantic-ai-middleware.git
cd pydantic-ai-middleware
make install
make test  # 100% coverage required
make all   # lint + typecheck + test
```

--------------------------------

### Clone Repository and Install Dependencies (Bash)

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/CONTRIBUTING.md

Clones the pydantic-ai-middleware repository and installs all project dependencies using uv. It also sets up pre-commit hooks for code quality.

```bash
git clone https://github.com/vstorm-co/pydantic-ai-middleware.git
cd pydantic-ai-middleware
make install
```

--------------------------------

### Log Tool Invocations with before_tool_call Hook (Python)

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/examples/hooks_showcase/presentation.html

This example demonstrates logging tool invocations using the `before_tool_call` hook. It helps in understanding which tools are being used and with what arguments. The hook receives the tool name and its arguments.

```python
from pydantic_ai_middleware import before_tool_call

@before_tool_call
def log_tool_invocation(tool_name: str, tool_args: dict):
    print(f"Tool '{tool_name}' invoked with arguments: {tool_args}")
```

--------------------------------

### Decorator Syntax for Middleware (Python)

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/examples/hooks_showcase/presentation.html

This example shows how to apply middleware using decorator syntax, specifically the `@before_run` decorator. This provides a clean and Pythonic way to integrate middleware into your functions. It's equivalent to defining a separate function and registering it.

```python
from pydantic_ai_middleware import before_run

@before_run
def my_decorator_middleware(prompt: str):
    print("This middleware is applied using decorator syntax.")
```

--------------------------------

### Basic Cost Tracking with Pydantic AI Middleware

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/examples/cost-tracking.md

Demonstrates basic cost tracking by creating a middleware instance and attaching it to an agent. It prints run-specific and total costs upon each update.

```python
from pydantic_ai import Agent
from pydantic_ai.models.test import TestModel
from pydantic_ai_middleware import MiddlewareAgent, MiddlewareContext
from pydantic_ai_middleware.cost_tracking import create_cost_tracking_middleware


def on_cost(info):
    print(
        f"Run #{info.run_count}: "
        f"{info.run_request_tokens} in / {info.run_response_tokens} out"
    )
    if info.run_cost_usd is not None:
        print(f"  Run cost:   ${info.run_cost_usd:.4f}")
        print(f"  Total cost: ${info.total_cost_usd:.4f}")


cost_mw = create_cost_tracking_middleware(
    model_name="openai:gpt-4.1",
    on_cost_update=on_cost,
)

agent = MiddlewareAgent(
    agent=Agent(model=TestModel()),
    middleware=[cost_mw],
    context=MiddlewareContext(),
)

result = await agent.run("What is the capital of France?")
```

--------------------------------

### Parallel Middleware Performance Example

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/advanced/parallel-execution.md

Illustrates the performance benefits of `ParallelMiddleware` by comparing sequential execution time with parallel execution time. This example uses middleware components with simulated delays to highlight how parallel execution can achieve a total time equal to the longest single middleware execution, rather than the sum of all.

```python
from pydantic_ai_middleware import ParallelMiddleware, AggregationStrategy

# Sequential: 0.5s + 0.5s + 0.5s + 0.5s = 2.0s total
# Parallel:  max(0.5s, 0.5s, 0.5s, 0.5s) = 0.5s total

parallel_validators = ParallelMiddleware(
    middleware=[
        ProfanityFilter(delay=0.5),
        PIIDetector(delay=0.5),
        PromptInjectionGuard(delay=0.5),
        ToxicityChecker(delay=0.5),
    ],
    strategy=AggregationStrategy.ALL_MUST_PASS,
)
```

--------------------------------

### Run Toxicity Check Example in Bash

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/examples/toxicity-check.md

This bash command executes a runnable demo script for the toxicity check middleware. It assumes the script is located at `examples/toxicity_check.py` and uses the `uv` command-line tool for running Python applications.

```bash
uv run python examples/toxicity_check.py

```

--------------------------------

### Implement 'after_run' Hook for Timing

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/examples/hooks_showcase/presentation.html

Demonstrates implementing both 'before_run' and 'after_run' hooks to measure the execution time of an agent. The 'before_run' hook records the start time, and 'after_run' calculates and prints the elapsed time along with the output.

```python
import time
from pydantic_ai import Agent
from pydantic_ai_middleware import AgentMiddleware, MiddlewareAgent

class TimingMiddleware(AgentMiddleware[None]):

    def __init__(self):
        self._start = 0

    async def before_run(self, prompt, deps, ctx):
        self._start = time.perf_counter()
        print("[before_run] Timer started")
        return prompt

    async def after_run(self, prompt, output, deps, ctx):
        elapsed = time.perf_counter() - self._start
        print(f"[after_run] {output!r}")
        print(f"[after_run] Elapsed: {elapsed:.3f}s")
        return output

agent = Agent("openai:gpt-4.1")
mw_agent = MiddlewareAgent(
    agent=agent,
    middleware=[TimingMiddleware()],
)

result = await mw_agent.run("Calculate 2 + 2.")
```

--------------------------------

### Implement 'before_tool_call' Hook

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/examples/hooks_showcase/presentation.html

Shows how to implement the 'before_tool_call' hook to intercept and log tool invocations. This example logs the tool name and its arguments before the tool is executed.

```python
from pydantic_ai_middleware import AgentMiddleware, MiddlewareAgent
from pydantic_ai.tools import FunctionToolset
from pydantic_ai import Agent

class BeforeToolCallLogger(AgentMiddleware[None]):

    async def before_tool_call(
        self, tool_name, tool_args, deps, ctx
    ):
        print(f"[before] {tool_name}({tool_args})")
        return tool_args

toolset = FunctionToolset()

@toolset.tool
def add(a: int, b: int) -> str:
    """Add two numbers."""
    return str(a + b)

agent = Agent("openai:gpt-4.1", tools=toolset)
mw_agent = MiddlewareAgent(
    agent=agent,
    middleware=[BeforeToolCallLogger()],
)

result = await mw_agent.run("Add 5 and 3.")
```

--------------------------------

### Basic Logging with AgentMiddleware in Python

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/examples/logging.md

Implements basic logging for agent and tool activities using the AgentMiddleware. It logs agent start/end, tool calls, and errors. Dependencies: pydantic-ai-middleware, logging.

```python
import logging
from pydantic_ai_middleware import AgentMiddleware

logger = logging.getLogger(__name__)

class LoggingMiddleware(AgentMiddleware[None]):
    async def before_run(self, prompt, deps, ctx):
        logger.info(f"Agent started with: {prompt[:100]}...")
        return prompt

    async def after_run(self, prompt, output, deps, ctx):
        logger.info(f"Agent finished with: {output}")
        return output

    async def before_tool_call(
        self, tool_name, tool_args, deps, ctx
    ):
        logger.info(f"Tool call: {tool_name}({tool_args})")
        return tool_args

    async def on_error(self, error, deps, ctx):
        logger.error(f"Agent error: {error}")
        return None
```

--------------------------------

### Budget Enforcement with Pydantic AI Middleware

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/examples/cost-tracking.md

Shows how to enforce a budget limit using cost tracking middleware. If the cost exceeds the specified limit, a BudgetExceededError is raised, and the loop breaks.

```python
from pydantic_ai import Agent
from pydantic_ai.models.test import TestModel
from pydantic_ai_middleware import MiddlewareAgent, MiddlewareContext
from pydantic_ai_middleware.cost_tracking import create_cost_tracking_middleware
from pydantic_ai_middleware.exceptions import BudgetExceededError

cost_mw = create_cost_tracking_middleware(
    model_name="openai:gpt-4.1",
    budget_limit_usd=0.50,
)

agent = MiddlewareAgent(
    agent=Agent(model=TestModel()),
    middleware=[cost_mw],
    context=MiddlewareContext(),
)

prompts = [
    "Summarize this report",
    "Translate to Spanish",
    "Generate test cases",
]

for prompt in prompts:
    try:
        result = await agent.run(prompt)
        print(f"OK: {prompt}")
    except BudgetExceededError as e:
        print(f"Stopped: ${e.cost:.4f} >= ${e.budget:.4f} limit")
        break
```

--------------------------------

### Example JSON Pipeline Configuration

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/advanced/config-loading.md

Demonstrates a complex middleware pipeline defined in JSON format. It includes logging, parallel execution of PII guard and profanity filter, and conditional execution based on an 'is_admin' predicate.

```json
[
  {"type": "logging"},
  {
    "parallel": {
      "middleware": [
        {"type": "pii_guard"},
        {"type": "profanity_filter"}
      ],
      "strategy": "all_must_pass"
    }
  },
  {
    "when": {
      "predicate": "is_admin",
      "then": [{"type": "audit"}],
      "else": [{"type": "basic_audit"}]
    }
  }
]
```

--------------------------------

### Run Prompt Injection Demo Example

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/examples/prompt-injection.md

This command executes a complete runnable demo of the prompt injection middleware. The demo includes 8 different injection patterns and showcases the middleware's functionality in a practical scenario. It requires the `uv` command-line tool to run.

```bash
uv run python examples/prompt_injection.py

```

--------------------------------

### Build and Serve Documentation (Bash)

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/CONTRIBUTING.md

Commands to build the project's documentation and serve it locally for review. This is essential for ensuring documentation is up-to-date with code changes.

```bash
# Build documentation
make docs

# Serve documentation locally
make docs-serve
```

--------------------------------

### Resetting Cost Counters

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/examples/cost-tracking.md

Illustrates how to reset the cost tracking counters (total cost and run count) using the `reset()` method of the `CostTrackingMiddleware` instance. This is useful for starting new billing periods.

```python
from pydantic_ai_middleware.cost_tracking import CostTrackingMiddleware

cost_mw = CostTrackingMiddleware(model_name="openai:gpt-4.1")

# ... run agent multiple times ...

print(f"Session cost: ${cost_mw.total_cost:.4f}")
print(f"Session runs: {cost_mw.run_count}")

# Start a new billing period
cost_mw.reset()
print(f"After reset: ${cost_mw.total_cost:.4f}")
```

--------------------------------

### Build and Serve Documentation

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/AGENTS.md

Builds the project documentation and serves it locally. This allows developers to preview the documentation as it would appear when published.

```bash
make docs-serve
```

--------------------------------

### Audit Logging with AgentMiddleware in Python

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/examples/logging.md

Implements audit logging to track user actions within the agent. It logs agent start and tool call events, including user and session information. Dependencies: pydantic-ai-middleware, dataclasses.

```python
from dataclasses import dataclass
from pydantic_ai_middleware import AgentMiddleware

@dataclass
class AuditDeps:
    user_id: str
    session_id: str

class AuditMiddleware(AgentMiddleware[AuditDeps]):
    async def before_run(self, prompt, deps, ctx):
        await self._record({
            "action": "agent_start",
            "user_id": deps.user_id,
            "session_id": deps.session_id,
            "prompt": prompt,
        })
        return prompt

    async def before_tool_call(
        self, tool_name, tool_args, deps, ctx
    ):
        await self._record({
            "action": "tool_call",
            "user_id": deps.user_id,
            "tool": tool_name,
            "args": tool_args,
        })
        return tool_args

    async def _record(self, data):
        # Save to database, send to logging service, etc.
        pass
```

--------------------------------

### Structured Logging with AgentMiddleware in Python

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/examples/logging.md

Implements structured logging using JSON format for agent activities. It logs run start, run end, and tool calls with detailed event information. Dependencies: pydantic-ai-middleware, datetime, json.

```python
import json
from datetime import datetime
from pydantic_ai_middleware import AgentMiddleware

class StructuredLogger(AgentMiddleware[None]):
    def __init__(self):
        self.run_id = None

    async def before_run(self, prompt, deps, ctx):
        self.run_id = datetime.now().isoformat()
        self._log({
            "event": "run_start",
            "run_id": self.run_id,
            "prompt_length": len(str(prompt)),
        })
        return prompt

    async def after_run(self, prompt, output, deps, ctx):
        self._log({
            "event": "run_end",
            "run_id": self.run_id,
            "output_length": len(str(output)),
        })
        return output

    async def before_tool_call(
        self, tool_name, tool_args, deps, ctx
    ):
        self._log({
            "event": "tool_call",
            "run_id": self.run_id,
            "tool": tool_name,
            "args": tool_args,
        })
        return tool_args

    def _log(self, data):
        print(json.dumps(data))
```

--------------------------------

### Log Incoming Prompts with before_run Hook (Python)

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/examples/hooks_showcase/presentation.html

This snippet demonstrates how to use the `before_run` hook to log incoming prompts. It's useful for debugging and monitoring the initial input to the AI model. No external dependencies are required beyond the pydantic-ai-middleware library.

```python
from pydantic_ai_middleware import before_run

@before_run
def log_prompt(prompt: str):
    print(f"Incoming prompt: {prompt}")
```

--------------------------------

### Install and Use Pydantic AI Middleware

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/README.md

Install the pydantic-ai-middleware package using pip. Then, wrap any pydantic-ai Agent with MiddlewareAgent to add custom middleware for input validation, output filtering, and prompt injection protection.

```bash
pip install pydantic-ai-middleware
```

```python
from pydantic_ai import Agent
from pydantic_ai_middleware import MiddlewareAgent, AgentMiddleware, InputBlocked

class ContentFilter(AgentMiddleware[None]):
    """Block dangerous prompts before they reach the LLM."""

    async def before_run(self, prompt, deps, ctx=None):
        if "ignore all instructions" in prompt.lower():
            raise InputBlocked("Prompt injection attempt blocked")
        return prompt

    async def after_run(self, prompt, output, deps, ctx=None):
        # Redact sensitive patterns from the LLM response
        return output.replace("SSN:", "[REDACTED]")

# Wrap any pydantic-ai Agent with middleware
base_agent = Agent("openai:gpt-4o", instructions="You are a helpful assistant.")
agent = MiddlewareAgent(agent=base_agent, middleware=[ContentFilter()])

result = await agent.run("Hello, how are you?")
print(result.output)
```

--------------------------------

### Use Decorators for Simple Middleware Functions (Python)

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/index.md

This example illustrates how to create middleware using decorators like `@before_run` and `@before_tool_call`. These decorators allow for concise definition of middleware functions that hook into specific agent lifecycle stages. The `block_dangerous_tools` function demonstrates how to intercept and modify tool calls based on certain conditions.

```python
from pydantic_ai_middleware import before_run, after_run, ToolBlocked, before_tool_call

@before_run
async def log_input(prompt, deps, ctx):
    print(f"Input: {prompt}")
    return prompt

@before_tool_call
async def block_dangerous_tools(tool_name, tool_args, deps, ctx):
    if tool_name == "delete_file":
        raise ToolBlocked(tool_name, "Not allowed")
    return tool_args

```

--------------------------------

### Create Custom Agent Middleware in Python

Source: https://context7.com/vstorm-co/pydantic-ai-middleware/llms.txt

Illustrates how to create custom middleware by inheriting from `AgentMiddleware`. This example shows how to implement rate limiting logic in `before_run` and how to restrict middleware application to specific tools using `tool_names`. It also lists the available lifecycle hooks for custom middleware development.

```python
from pydantic_ai_middleware import AgentMiddleware, ToolBlocked

class RateLimiter(AgentMiddleware[None]):
    """Limit requests per minute with state tracking."""

    def __init__(self, max_per_minute: int = 10):
        self.max_per_minute = max_per_minute
        self._timestamps: list[float] = []

    async def before_run(self, prompt, deps, ctx=None):
        import time
        now = time.time()
        self._timestamps = [t for t in self._timestamps if now - t < 60]
        if len(self._timestamps) >= self.max_per_minute:
            raise InputBlocked("Rate limit exceeded — try again later")
        self._timestamps.append(now)
        return prompt

class SensitiveToolGuard(AgentMiddleware[None]):
    """Block specific tools based on security policy."""

    tool_names = {"delete_file", "execute_command"}  # Only intercept these tools

    async def before_tool_call(self, tool_name, tool_args, deps, ctx=None):
        raise ToolBlocked(tool_name, "This tool requires admin approval")

# Hooks available:
# - before_run(prompt, deps, ctx) -> prompt
# - after_run(prompt, output, deps, ctx) -> output
# - before_model_request(messages, deps, ctx) -> messages
# - before_tool_call(tool_name, tool_args, deps, ctx) -> tool_args | ToolPermissionResult
# - on_tool_error(tool_name, tool_args, error, deps, ctx) -> Exception | None
# - after_tool_call(tool_name, tool_args, result, deps, ctx) -> result
# - on_error(error, deps, ctx) -> Exception | None

```

--------------------------------

### Chaining Pydantic AI Middleware with Mixed Filtering

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/examples/tool-name-filtering.md

This example shows how to combine different middleware components, including class-based and decorator-based ones, into a single MiddlewareChain. Each middleware in the chain can have its own specific tool filtering (e.g., EmailValidator for emails, sandbox_check for code execution, cache_results for web search), and a default middleware like LoggingMiddleware can apply to all tools.

```python
from pydantic_ai_middleware import MiddlewareChain

# Assuming EmailValidator, sandbox_check, and cache_results are defined as above
# and LoggingMiddleware is a pre-defined or custom middleware

chain = MiddlewareChain([
    EmailValidator(),    # only send_email, draft_email
    sandbox_check,       # only execute_code
    cache_results,       # only web_search
    LoggingMiddleware(), # all tools (tool_names = None)
])
```

--------------------------------

### Asynchronous Middleware Test

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/AGENTS.md

An example of an asynchronous test function for the MiddlewareAgent. It demonstrates setting up a basic agent and middleware, then running a test case.

```python
from pydantic_ai import Agent
from pydantic_ai.models.test import TestModel
from pydantic_ai_middleware import MiddlewareAgent, AgentMiddleware

async def test_middleware():
    model = TestModel()
    model.custom_output_text = "test"
    agent = Agent(model, output_type=str)
    middleware_agent = MiddlewareAgent(agent, middleware=[...])
    result = await middleware_agent.run("test")
```

--------------------------------

### on_tool_error Hook Implementation

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/concepts/hooks.md

Example implementation of the 'on_tool_error' hook, triggered when a tool execution results in an exception. It can be used to handle or replace the error.

```python
from typing import Any, Dict
# Assuming DepsT is defined elsewhere
DepsT = None

async def on_tool_error(
    self,
    tool_name: str,
    tool_args: Dict[str, Any],
    error: Exception,
    deps: DepsT | None,
    ctx: ScopedContext | None = None,
) -> Exception | None:
    # Return None to re-raise original
    # Return exception to raise different one
    return None
```

--------------------------------

### Configure Prompt Injection Middleware Threshold

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/examples/prompt-injection.md

This example shows how to instantiate the `PromptInjectionMiddleware` with different risk thresholds. A lower threshold (e.g., 0.5) blocks more injection attempts, including medium and high severity ones. A higher threshold (e.g., 0.85) is stricter and only blocks the most severe attempts.

```python
# Default (0.5) - blocks medium and high severity
middleware = PromptInjectionMiddleware(threshold=0.5)

# Strict (0.85) - only blocks severe attempts
middleware = PromptInjectionMiddleware(threshold=0.85)

```

--------------------------------

### before_model_request Hook Implementation

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/concepts/hooks.md

Example implementation of the 'before_model_request' hook, called just before sending data to the language model. It can be used to modify the messages being sent.

```python
from typing import Any, Sequence
# Assuming ModelMessage and DepsT are defined elsewhere
ModelMessage = Any
DepsT = None

async def before_model_request(
    self,
    messages: list[ModelMessage],
    deps: DepsT | None,
    ctx: ScopedContext | None = None,
) -> list[ModelMessage]:
    # Return modified messages
    return messages
```

--------------------------------

### after_run Hook Implementation

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/concepts/hooks.md

Example implementation of the 'after_run' hook, executed after the agent has finished processing. This hook allows for modification of the final output.

```python
async def after_run(
    self,
    prompt: str | Sequence[Any],
    output: Any,
    deps: DepsT | None,
    ctx: ScopedContext | None = None,
) -> Any:
    # Return modified output
    return output
```

--------------------------------

### Token and USD Cost Tracking with CostTrackingMiddleware (Python)

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/examples/hooks_showcase/presentation.html

This example shows how to track token and USD costs using the `CostTrackingMiddleware`. This middleware automatically monitors the token usage and associated costs of LLM API calls, providing valuable insights for budget management. It requires configuration with API keys and pricing details.

```python
from pydantic_ai_middleware import CostTrackingMiddleware

# Assuming CostTrackingMiddleware is configured elsewhere
# Example usage within a pydantic-ai-middleware setup:
# middleware = [
#     CostTrackingMiddleware(api_key='YOUR_API_KEY', pricing={'gpt-4': 0.03})
# ]
```

--------------------------------

### Handle BudgetExceededError Exception (Python)

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/api/exceptions.md

Provides an example of catching the `BudgetExceededError`, which is raised when the accumulated cost surpasses the configured budget. This is typically used with `CostTrackingMiddleware`.

```python
from pydantic_ai_middleware.exceptions import BudgetExceededError

# Raised automatically by CostTrackingMiddleware
# You can also catch it:
try:
    result = await agent.run("prompt")
except BudgetExceededError as e:
    print(f"Over budget: ${e.cost:.4f} >= ${e.budget:.4f}")
```

--------------------------------

### Python: Execute script 05_after_tool_call.py

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/examples/hooks_showcase/presentation.html

Illustrates the console output from 05_after_tool_call.py, highlighting tool execution, result logging, and execution time.

```python
import asyncio
from pydantic_ai_middleware import PydanticAIAgent

async def main():
    agent = PydanticAIAgent(
        tools=[
            lambda a, b: a * b
        ]
    )
    await agent.run("What is 12 times 8?")

if __name__ == "__main__":
    asyncio.run(main())
```

--------------------------------

### Implement 'after_tool_call' Hook

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/examples/hooks_showcase/presentation.html

Demonstrates implementing the 'after_tool_call' hook to inspect the result of a tool invocation. This example logs the tool name and its result after the tool has been executed.

```python
from pydantic_ai_middleware import AgentMiddleware, MiddlewareAgent
from pydantic_ai.tools import FunctionToolset
from pydantic_ai import Agent

class AfterToolCallLogger(AgentMiddleware[None]):

    async def after_tool_call(
        self, tool_name, tool_args, result, deps, ctx
    ):
        print(f"[after] {tool_name} → {result!r}")
        return result

toolset = FunctionToolset()

@toolset.tool
def multiply(a: int, b: int) -> str:
    """Multiply two numbers."""
    return str(a * b)

agent = Agent("openai:gpt-4.1", tools=toolset)
mw_agent = MiddlewareAgent(
    agent=agent,
    middleware=[AfterToolCallLogger()],
)

result = await mw_agent.run("Multiply 6 by 7.")
```

--------------------------------

### Configure Production AI Agent with Middleware

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/examples/hooks_showcase/presentation.html

This Python code snippet illustrates how to create a production-ready AI agent using `pydantic-deep`. It shows the configuration of default middleware like cost tracking and context management, alongside the integration of custom middleware and hooks for enhanced functionality such as auditing and permissions.

```python
from pydantic_deep import create_deep_agent
from pydantic_ai_middleware import AgentMiddleware, MiddlewareContext

# Assume these are defined elsewhere
class AuditMiddleware(AgentMiddleware[None]):
    async def before_run(self, prompt, deps, ctx: MiddlewareContext):
        print(f"[Audit] Before run: {prompt[:50]}...")
        return prompt

class PermissionMiddleware(AgentMiddleware[None]):
    async def before_run(self, prompt, deps, ctx: MiddlewareContext):
        print(f"[Permission] Checking permissions for prompt...")
        # Add permission logic here
        return prompt

def my_cost_callback(info):
    print(f"Cost update: Run #{info.run_count}, Cost: ${info.run_cost_usd:.6f}")

def my_ctx_callback(ctx_info):
    print(f"Context update: Tokens used: {ctx_info.tokens_used}")

# Placeholder for custom middleware and hooks
audit_mw = AuditMiddleware()
permission_mw = PermissionMiddleware()
safety_gate = None # Assume defined
audit_logger = None # Assume defined

# Create the deep agent with various configurations
agent = create_deep_agent(
    model="openai:gpt-4.1",

    # Cost tracking (enabled by default)
    cost_tracking=True,
    cost_budget_usd=5.00,
    on_cost_update=my_cost_callback,

    # Context manager (enabled by default)
    context_manager=True,
    context_manager_max_tokens=200_000,
    on_context_update=my_ctx_callback,

    # Custom middleware
    middleware=[audit_mw, permission_mw],

    # Claude Code-style hooks
    hooks=[safety_gate, audit_logger],
)

# The actual agent returned might be wrapped in MiddlewareAgent
# if any middleware or hooks are active.
# The following logic is illustrative of how pydantic-deep might handle this:

all_middleware = [
    audit_mw,
    permission_mw,
    # HooksMiddleware would be added here if hooks are provided
    # CheckpointMiddleware if enabled
    # ContextManagerMiddleware (default ON)
    # CostTrackingMiddleware (default ON)
]
middleware_context = MiddlewareContext() # Example context
permission_handler = None # Example handler

# Simplified representation of the final agent creation
# In reality, pydantic-deep handles the wrapping logic internally.
if all_middleware or cost_tracking or hooks:
    # final_agent = MiddlewareAgent(
    #     agent=agent,
    #     middleware=all_middleware,
    #     context=middleware_context,
    #     permission_handler=permission_handler,
    # )
    print("Agent configured with middleware and hooks.")
else:
    # final_agent = agent
    print("Agent configured without additional middleware/hooks.")

# Example of how the agent might be used (conceptual)
# async def main():
#     response = await agent.run("What is the capital of France?")
#     print(response)

# if __name__ == "__main__":
#     import asyncio
#     asyncio.run(main())

```

--------------------------------

### on_error Hook Implementation

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/concepts/hooks.md

Example implementation of the 'on_error' hook, called when any unhandled exception occurs during agent execution. It allows for centralized error handling.

```python
from typing import Any
# Assuming DepsT is defined elsewhere
DepsT = None

async def on_error(
    self,
    error: Exception,
    deps: DepsT | None,
    ctx: ScopedContext | None = None,
) -> Exception | None:
    # Return None to re-raise original
    # Return exception to raise different one
    return None
```

--------------------------------

### after_tool_call Hook Implementation

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/concepts/hooks.md

Example implementation of the 'after_tool_call' hook, which runs after a tool has successfully executed. It provides an opportunity to modify the tool's result.

```python
from typing import Any, Dict
# Assuming DepsT is defined elsewhere
DepsT = None

async def after_tool_call(
    self,
    tool_name: str,
    tool_args: Dict[str, Any],
    result: Any,
    deps: DepsT | None,
    ctx: ScopedContext | None = None,
) -> Any:
    # Return modified result
    return result
```

--------------------------------

### before_tool_call Hook Implementation

Source: https://github.com/vstorm-co/pydantic-ai-middleware/blob/main/docs/concepts/hooks.md

Example implementation of the 'before_tool_call' hook, invoked prior to executing a tool. It allows modification of tool arguments or blocking the call.

```python
from typing import Any, Dict
# Assuming ToolPermissionResult and DepsT are defined elsewhere
ToolPermissionResult = Any
DepsT = None

async def before_tool_call(
    self,
    tool_name: str,
    tool_args: Dict[str, Any],
    deps: DepsT | None,
    ctx: ScopedContext | None = None,
) -> Dict[str, Any] | ToolPermissionResult:
    # Return modified arguments (dict) or ToolPermissionResult
    return tool_args
```