Skip to content

Exceptions

Custom exceptions for middleware control flow and error reporting. All exceptions inherit from MiddlewareError, making it easy to catch any middleware-related error with a single except clause.

The blocking exceptions (InputBlocked, ToolBlocked, OutputBlocked) are the primary way middleware communicates that a request should be rejected. Raise them from any hook to stop processing immediately.

MiddlewareError

pydantic_ai_middleware.MiddlewareError

Bases: Exception

Base exception for middleware errors.

Source code in src/pydantic_ai_middleware/exceptions.py
Python
11
12
class MiddlewareError(Exception):
    """Base exception for middleware errors."""

Base exception for all middleware errors.

MiddlewareConfigError

pydantic_ai_middleware.exceptions.MiddlewareConfigError

Bases: MiddlewareError

Raised when middleware configuration is invalid.

Source code in src/pydantic_ai_middleware/exceptions.py
Python
15
16
class MiddlewareConfigError(MiddlewareError):
    """Raised when middleware configuration is invalid."""

Raised when middleware configuration is invalid (unknown type names, bad predicate specs, malformed config files).

InputBlocked

pydantic_ai_middleware.InputBlocked

Bases: MiddlewareError

Raised when input is blocked by middleware.

Source code in src/pydantic_ai_middleware/exceptions.py
Python
19
20
21
22
23
24
class InputBlocked(MiddlewareError):
    """Raised when input is blocked by middleware."""

    def __init__(self, reason: str = "Input blocked") -> None:
        self.reason = reason
        super().__init__(reason)

Raised to block input processing.

Python
from pydantic_ai_middleware import InputBlocked

raise InputBlocked("Content not allowed")
raise InputBlocked()  # Uses default message

ToolBlocked

pydantic_ai_middleware.ToolBlocked

Bases: MiddlewareError

Raised when a tool call is blocked by middleware.

Source code in src/pydantic_ai_middleware/exceptions.py
Python
27
28
29
30
31
32
33
class ToolBlocked(MiddlewareError):
    """Raised when a tool call is blocked by middleware."""

    def __init__(self, tool_name: str, reason: str = "Tool blocked") -> None:
        self.tool_name = tool_name
        self.reason = reason
        super().__init__(f"Tool '{tool_name}' blocked: {reason}")

Raised to block a tool call.

Python
from pydantic_ai_middleware import ToolBlocked

raise ToolBlocked("dangerous_tool", "Not authorized")
raise ToolBlocked("tool_name")  # Uses default reason

OutputBlocked

pydantic_ai_middleware.OutputBlocked

Bases: MiddlewareError

Raised when output is blocked by middleware.

Source code in src/pydantic_ai_middleware/exceptions.py
Python
36
37
38
39
40
41
class OutputBlocked(MiddlewareError):
    """Raised when output is blocked by middleware."""

    def __init__(self, reason: str = "Output blocked") -> None:
        self.reason = reason
        super().__init__(reason)

Raised to block output.

Python
from pydantic_ai_middleware import OutputBlocked

raise OutputBlocked("Contains sensitive information")
raise OutputBlocked()  # Uses default message

BudgetExceededError

pydantic_ai_middleware.exceptions.BudgetExceededError

Bases: MiddlewareError

Raised when accumulated cost exceeds the configured budget limit.

Source code in src/pydantic_ai_middleware/exceptions.py
Python
108
109
110
111
112
113
114
115
116
117
118
119
120
class BudgetExceededError(MiddlewareError):
    """Raised when accumulated cost exceeds the configured budget limit."""

    def __init__(self, cost: float, budget: float) -> None:
        """Initialize the exception.

        Args:
            cost: The current accumulated cost in USD.
            budget: The configured budget limit in USD.
        """
        self.cost = cost
        self.budget = budget
        super().__init__(f"Budget exceeded: ${cost:.4f} >= ${budget:.4f} limit")

__init__(cost, budget)

Initialize the exception.

Parameters:

Name Type Description Default
cost float

The current accumulated cost in USD.

 required
budget float

The configured budget limit in USD.

 required
Source code in src/pydantic_ai_middleware/exceptions.py
Python
111
112
113
114
115
116
117
118
119
120
def __init__(self, cost: float, budget: float) -> None:
    """Initialize the exception.

    Args:
        cost: The current accumulated cost in USD.
        budget: The configured budget limit in USD.
    """
    self.cost = cost
    self.budget = budget
    super().__init__(f"Budget exceeded: ${cost:.4f} >= ${budget:.4f} limit")

Raised when accumulated cost exceeds the configured budget limit. Used by 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}")

ParallelExecutionFailed

pydantic_ai_middleware.exceptions.ParallelExecutionFailed

Bases: MiddlewareError

Raised when parallel middleware execution fails.

Contains information about which middleware failed and which succeeded, allowing for detailed error handling and debugging.

Source code in src/pydantic_ai_middleware/exceptions.py
Python
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
class ParallelExecutionFailed(MiddlewareError):
    """Raised when parallel middleware execution fails.

    Contains information about which middleware failed and which succeeded,
    allowing for detailed error handling and debugging.
    """

    def __init__(
        self,
        errors: list[Exception],
        results: list[Any] | None = None,
        message: str = "Parallel middleware execution failed",
    ) -> None:
        """Initialize the exception.

        Args:
            errors: List of exceptions from failed middleware.
            results: Optional list of successful results (for partial failures).
            message: Human-readable error message.
        """
        self.errors = errors
        self.results = results or []
        self.failed_count = len(errors)
        self.success_count = len(self.results)
        super().__init__(f"{message}: {self.failed_count} failed, {self.success_count} succeeded")

__init__(errors, results=None, message='Parallel middleware execution failed')

Initialize the exception.

Parameters:

Name Type Description Default
errors list[Exception]

List of exceptions from failed middleware.

 required
results list[Any] | None

Optional list of successful results (for partial failures).

 None
message str

Human-readable error message.

 'Parallel middleware execution failed'
Source code in src/pydantic_ai_middleware/exceptions.py
Python
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
def __init__(
    self,
    errors: list[Exception],
    results: list[Any] | None = None,
    message: str = "Parallel middleware execution failed",
) -> None:
    """Initialize the exception.

    Args:
        errors: List of exceptions from failed middleware.
        results: Optional list of successful results (for partial failures).
        message: Human-readable error message.
    """
    self.errors = errors
    self.results = results or []
    self.failed_count = len(errors)
    self.success_count = len(self.results)
    super().__init__(f"{message}: {self.failed_count} failed, {self.success_count} succeeded")

Raised when parallel middleware execution fails. Contains the list of errors and any successful results.

GuardrailTimeout

pydantic_ai_middleware.exceptions.GuardrailTimeout

Bases: MiddlewareError

Raised when an async guardrail times out.

This can occur when a guardrail takes longer than the configured timeout while running concurrently with the agent.

Source code in src/pydantic_ai_middleware/exceptions.py
Python
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
class GuardrailTimeout(MiddlewareError):
    """Raised when an async guardrail times out.

    This can occur when a guardrail takes longer than the configured
    timeout while running concurrently with the agent.
    """

    def __init__(self, guardrail_name: str, timeout: float) -> None:
        """Initialize the exception.

        Args:
            guardrail_name: Name/identifier of the guardrail that timed out.
            timeout: The timeout value in seconds that was exceeded.
        """
        self.guardrail_name = guardrail_name
        self.timeout = timeout
        super().__init__(f"Guardrail '{guardrail_name}' timed out after {timeout:.2f}s")

__init__(guardrail_name, timeout)

Initialize the exception.

Parameters:

Name Type Description Default
guardrail_name str

Name/identifier of the guardrail that timed out.

 required
timeout float

The timeout value in seconds that was exceeded.

 required
Source code in src/pydantic_ai_middleware/exceptions.py
Python
78
79
80
81
82
83
84
85
86
87
def __init__(self, guardrail_name: str, timeout: float) -> None:
    """Initialize the exception.

    Args:
        guardrail_name: Name/identifier of the guardrail that timed out.
        timeout: The timeout value in seconds that was exceeded.
    """
    self.guardrail_name = guardrail_name
    self.timeout = timeout
    super().__init__(f"Guardrail '{guardrail_name}' timed out after {timeout:.2f}s")

Raised when an async guardrail exceeds its configured timeout.

MiddlewareTimeout

pydantic_ai_middleware.exceptions.MiddlewareTimeout

Bases: MiddlewareError

Raised when a middleware hook exceeds its configured timeout.

Source code in src/pydantic_ai_middleware/exceptions.py
Python
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
class MiddlewareTimeout(MiddlewareError):
    """Raised when a middleware hook exceeds its configured timeout."""

    def __init__(self, middleware_name: str, timeout: float, hook_name: str = "") -> None:
        """Initialize the exception.

        Args:
            middleware_name: Name/class of the middleware that timed out.
            timeout: The timeout value in seconds that was exceeded.
            hook_name: The specific hook that timed out.
        """
        self.middleware_name = middleware_name
        self.timeout = timeout
        self.hook_name = hook_name
        detail = f" in {hook_name}" if hook_name else ""
        super().__init__(f"Middleware '{middleware_name}' timed out{detail} after {timeout:.2f}s")

__init__(middleware_name, timeout, hook_name='')

Initialize the exception.

Parameters:

Name Type Description Default
middleware_name str

Name/class of the middleware that timed out.

 required
timeout float

The timeout value in seconds that was exceeded.

 required
hook_name str

The specific hook that timed out.

 ''
Source code in src/pydantic_ai_middleware/exceptions.py
Python
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def __init__(self, middleware_name: str, timeout: float, hook_name: str = "") -> None:
    """Initialize the exception.

    Args:
        middleware_name: Name/class of the middleware that timed out.
        timeout: The timeout value in seconds that was exceeded.
        hook_name: The specific hook that timed out.
    """
    self.middleware_name = middleware_name
    self.timeout = timeout
    self.hook_name = hook_name
    detail = f" in {hook_name}" if hook_name else ""
    super().__init__(f"Middleware '{middleware_name}' timed out{detail} after {timeout:.2f}s")

Raised when a middleware hook exceeds its per-middleware timeout.

AggregationFailed

pydantic_ai_middleware.exceptions.AggregationFailed

Bases: MiddlewareError

Raised when parallel results cannot be aggregated.

This occurs when the aggregation strategy cannot produce a valid result from the parallel middleware outputs.

Source code in src/pydantic_ai_middleware/exceptions.py
Python
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
class AggregationFailed(MiddlewareError):
    """Raised when parallel results cannot be aggregated.

    This occurs when the aggregation strategy cannot produce a valid
    result from the parallel middleware outputs.
    """

    def __init__(
        self,
        strategy: AggregationStrategy,
        reason: str,
        errors: list[Exception] | None = None,
    ) -> None:
        """Initialize the exception.

        Args:
            strategy: The aggregation strategy that failed.
            reason: Explanation of why aggregation failed.
            errors: Optional list of underlying errors.
        """
        self.strategy = strategy
        self.reason = reason
        self.errors = errors or []
        super().__init__(f"Aggregation failed ({strategy.value}): {reason}")

__init__(strategy, reason, errors=None)

Initialize the exception.

Parameters:

Name Type Description Default
strategy AggregationStrategy

The aggregation strategy that failed.

 required
reason str

Explanation of why aggregation failed.

 required
errors list[Exception] | None

Optional list of underlying errors.

 None
Source code in src/pydantic_ai_middleware/exceptions.py
Python
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
def __init__(
    self,
    strategy: AggregationStrategy,
    reason: str,
    errors: list[Exception] | None = None,
) -> None:
    """Initialize the exception.

    Args:
        strategy: The aggregation strategy that failed.
        reason: Explanation of why aggregation failed.
        errors: Optional list of underlying errors.
    """
    self.strategy = strategy
    self.reason = reason
    self.errors = errors or []
    super().__init__(f"Aggregation failed ({strategy.value}): {reason}")

Raised when parallel results cannot be aggregated according to the chosen strategy.