Middleware¶
Entrypoint to using Middleware plugins with Agents.
Reference docs
This page contains reference documentation for Middleware. See the docs for conceptual guides, tutorials, and examples on using Middleware.
| CLASS | DESCRIPTION |
|---|---|
ContextEditingMiddleware |
Middleware that automatically prunes tool results to manage context size. |
HumanInTheLoopMiddleware |
Human in the loop middleware. |
LLMToolSelectorMiddleware |
Uses an LLM to select relevant tools before calling the main model. |
LLMToolEmulator |
Middleware that emulates specified tools using an LLM instead of executing them. |
ModelCallLimitMiddleware |
Middleware that tracks model call counts and enforces limits. |
ModelFallbackMiddleware |
Automatic fallback to alternative models on errors. |
PIIMiddleware |
Detect and handle Personally Identifiable Information (PII) in agent conversations. |
PIIDetectionError |
Raised when configured to block on detected sensitive values. |
SummarizationMiddleware |
Middleware that summarizes conversation history when token limits are approached. |
ToolCallLimitMiddleware |
Middleware that tracks tool call counts and enforces limits. |
AgentMiddleware |
Base middleware class for an agent. |
AgentState |
State schema for the agent. |
ClearToolUsesEdit |
Configuration for clearing tool outputs when token limits are exceeded. |
InterruptOnConfig |
Configuration for an action requiring human in the loop. |
ModelRequest |
Model request information for the agent. |
ModelResponse |
Response from model execution including messages and optional structured output. |
ModelRequest |
Model request information for the agent. |
Bases: AgentMiddleware
Middleware that automatically prunes tool results to manage context size.
The middleware applies a sequence of edits when the total input token count
exceeds configured thresholds. Currently the ClearToolUsesEdit strategy is
supported, aligning with Anthropic's clear_tool_uses_20250919 behaviour.
class-attribute
instance-attribute
¶
state_schema: type[StateT] = cast('type[StateT]', AgentState)
The schema for state passed to the middleware nodes.
property
¶
name: str
The name of the middleware instance.
Defaults to the class name, but can be overridden for custom naming.
Logic to run before the agent execution starts.
async
¶
Async logic to run before the agent execution starts.
Logic to run before the model is called.
async
¶
Async logic to run before the model is called.
Logic to run after the model is called.
async
¶
Async logic to run after the model is called.
Logic to run after the agent execution completes.
async
¶
Async logic to run after the agent execution completes.
wrap_tool_call(
request: ToolCallRequest,
handler: Callable[[ToolCallRequest], ToolMessage | Command],
) -> ToolMessage | Command
Intercept tool execution for retries, monitoring, or modification.
Multiple middleware compose automatically (first defined = outermost).
Exceptions propagate unless handle_tool_errors is configured on ToolNode.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Tool call request with call
TYPE:
|
handler
|
Callable to execute the tool (can be called multiple times).
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ToolMessage | Command
|
|
The handler callable can be invoked multiple times for retry logic. Each call to handler is independent and stateless.
Examples:
Modify request before execution:
def wrap_tool_call(self, request, handler):
request.tool_call["args"]["value"] *= 2
return handler(request)
Retry on error (call handler multiple times):
def wrap_tool_call(self, request, handler):
for attempt in range(3):
try:
result = handler(request)
if is_valid(result):
return result
except Exception:
if attempt == 2:
raise
return result
Conditional retry based on response:
async
¶
awrap_tool_call(
request: ToolCallRequest,
handler: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command]],
) -> ToolMessage | Command
Intercept and control async tool execution via handler callback.
The handler callback executes the tool call and returns a ToolMessage or
Command. Middleware can call the handler multiple times for retry logic, skip
calling it to short-circuit, or modify the request/response. Multiple middleware
compose with first in list as outermost layer.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Tool call request with call
TYPE:
|
handler
|
Async callable to execute the tool and returns
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ToolMessage | Command
|
|
The handler callable can be invoked multiple times for retry logic. Each call to handler is independent and stateless.
Examples:
Async retry on error:
__init__(
*,
edits: Iterable[ContextEdit] | None = None,
token_count_method: Literal["approximate", "model"] = "approximate"
) -> None
Initializes a context editing middleware instance.
| PARAMETER | DESCRIPTION |
|---|---|
edits
|
Sequence of edit strategies to apply. Defaults to a single
TYPE:
|
token_count_method
|
Whether to use approximate token counting (faster, less accurate) or exact counting implemented by the chat model (potentially slower, more accurate).
TYPE:
|
wrap_model_call(
request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]
) -> ModelCallResult
Apply context edits before invoking the model via handler.
async
¶
awrap_model_call(
request: ModelRequest, handler: Callable[[ModelRequest], Awaitable[ModelResponse]]
) -> ModelCallResult
Apply context edits before invoking the model via handler (async version).
Bases: AgentMiddleware
Human in the loop middleware.
class-attribute
instance-attribute
¶
state_schema: type[StateT] = cast('type[StateT]', AgentState)
The schema for state passed to the middleware nodes.
property
¶
name: str
The name of the middleware instance.
Defaults to the class name, but can be overridden for custom naming.
Logic to run before the agent execution starts.
async
¶
Async logic to run before the agent execution starts.
Logic to run before the model is called.
async
¶
Async logic to run before the model is called.
async
¶
Async logic to run after the model is called.
wrap_model_call(
request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]
) -> ModelCallResult
Intercept and control model execution via handler callback.
The handler callback executes the model request and returns a ModelResponse.
Middleware can call the handler multiple times for retry logic, skip calling
it to short-circuit, or modify the request/response. Multiple middleware
compose with first in list as outermost layer.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Model request to execute (includes state and runtime).
TYPE:
|
handler
|
Callback that executes the model request and returns
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ModelCallResult
|
|
Examples:
Retry on error:
def wrap_model_call(self, request, handler):
for attempt in range(3):
try:
return handler(request)
except Exception:
if attempt == 2:
raise
Rewrite response:
def wrap_model_call(self, request, handler):
response = handler(request)
ai_msg = response.result[0]
return ModelResponse(
result=[AIMessage(content=f"[{ai_msg.content}]")],
structured_response=response.structured_response,
)
Error to fallback:
def wrap_model_call(self, request, handler):
try:
return handler(request)
except Exception:
return ModelResponse(result=[AIMessage(content="Service unavailable")])
Cache/short-circuit:
def wrap_model_call(self, request, handler):
if cached := get_cache(request):
return cached # Short-circuit with cached result
response = handler(request)
save_cache(request, response)
return response
Simple AIMessage return (converted automatically):
async
¶
awrap_model_call(
request: ModelRequest, handler: Callable[[ModelRequest], Awaitable[ModelResponse]]
) -> ModelCallResult
Intercept and control async model execution via handler callback.
The handler callback executes the model request and returns a ModelResponse.
Middleware can call the handler multiple times for retry logic, skip calling
it to short-circuit, or modify the request/response. Multiple middleware
compose with first in list as outermost layer.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Model request to execute (includes state and runtime).
TYPE:
|
handler
|
Async callback that executes the model request and returns
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ModelCallResult
|
ModelCallResult |
Examples:
Retry on error:
Logic to run after the agent execution completes.
async
¶
Async logic to run after the agent execution completes.
wrap_tool_call(
request: ToolCallRequest,
handler: Callable[[ToolCallRequest], ToolMessage | Command],
) -> ToolMessage | Command
Intercept tool execution for retries, monitoring, or modification.
Multiple middleware compose automatically (first defined = outermost).
Exceptions propagate unless handle_tool_errors is configured on ToolNode.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Tool call request with call
TYPE:
|
handler
|
Callable to execute the tool (can be called multiple times).
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ToolMessage | Command
|
|
The handler callable can be invoked multiple times for retry logic. Each call to handler is independent and stateless.
Examples:
Modify request before execution:
def wrap_tool_call(self, request, handler):
request.tool_call["args"]["value"] *= 2
return handler(request)
Retry on error (call handler multiple times):
def wrap_tool_call(self, request, handler):
for attempt in range(3):
try:
result = handler(request)
if is_valid(result):
return result
except Exception:
if attempt == 2:
raise
return result
Conditional retry based on response:
async
¶
awrap_tool_call(
request: ToolCallRequest,
handler: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command]],
) -> ToolMessage | Command
Intercept and control async tool execution via handler callback.
The handler callback executes the tool call and returns a ToolMessage or
Command. Middleware can call the handler multiple times for retry logic, skip
calling it to short-circuit, or modify the request/response. Multiple middleware
compose with first in list as outermost layer.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Tool call request with call
TYPE:
|
handler
|
Async callable to execute the tool and returns
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ToolMessage | Command
|
|
The handler callable can be invoked multiple times for retry logic. Each call to handler is independent and stateless.
Examples:
Async retry on error:
__init__(
interrupt_on: dict[str, bool | InterruptOnConfig],
*,
description_prefix: str = "Tool execution requires approval"
) -> None
Initialize the human in the loop middleware.
| PARAMETER | DESCRIPTION |
|---|---|
interrupt_on
|
Mapping of tool name to allowed actions. If a tool doesn't have an entry, it's auto-approved by default.
TYPE:
|
description_prefix
|
The prefix to use when constructing action requests.
This is used to provide context about the tool call and the action being
requested. Not used if a tool has a
TYPE:
|
Bases: AgentMiddleware
Uses an LLM to select relevant tools before calling the main model.
When an agent has many tools available, this middleware filters them down to only the most relevant ones for the user's query. This reduces token usage and helps the main model focus on the right tools.
Examples:
Limit to 3 tools:
from langchain.agents.middleware import LLMToolSelectorMiddleware
middleware = LLMToolSelectorMiddleware(max_tools=3)
agent = create_agent(
model="openai:gpt-4o",
tools=[tool1, tool2, tool3, tool4, tool5],
middleware=[middleware],
)
Use a smaller model for selection:
class-attribute
instance-attribute
¶
state_schema: type[StateT] = cast('type[StateT]', AgentState)
The schema for state passed to the middleware nodes.
property
¶
name: str
The name of the middleware instance.
Defaults to the class name, but can be overridden for custom naming.
Logic to run before the agent execution starts.
async
¶
Async logic to run before the agent execution starts.
Logic to run before the model is called.
async
¶
Async logic to run before the model is called.
Logic to run after the model is called.
async
¶
Async logic to run after the model is called.
Logic to run after the agent execution completes.
async
¶
Async logic to run after the agent execution completes.
wrap_tool_call(
request: ToolCallRequest,
handler: Callable[[ToolCallRequest], ToolMessage | Command],
) -> ToolMessage | Command
Intercept tool execution for retries, monitoring, or modification.
Multiple middleware compose automatically (first defined = outermost).
Exceptions propagate unless handle_tool_errors is configured on ToolNode.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Tool call request with call
TYPE:
|
handler
|
Callable to execute the tool (can be called multiple times).
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ToolMessage | Command
|
|
The handler callable can be invoked multiple times for retry logic. Each call to handler is independent and stateless.
Examples:
Modify request before execution:
def wrap_tool_call(self, request, handler):
request.tool_call["args"]["value"] *= 2
return handler(request)
Retry on error (call handler multiple times):
def wrap_tool_call(self, request, handler):
for attempt in range(3):
try:
result = handler(request)
if is_valid(result):
return result
except Exception:
if attempt == 2:
raise
return result
Conditional retry based on response:
async
¶
awrap_tool_call(
request: ToolCallRequest,
handler: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command]],
) -> ToolMessage | Command
Intercept and control async tool execution via handler callback.
The handler callback executes the tool call and returns a ToolMessage or
Command. Middleware can call the handler multiple times for retry logic, skip
calling it to short-circuit, or modify the request/response. Multiple middleware
compose with first in list as outermost layer.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Tool call request with call
TYPE:
|
handler
|
Async callable to execute the tool and returns
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ToolMessage | Command
|
|
The handler callable can be invoked multiple times for retry logic. Each call to handler is independent and stateless.
Examples:
Async retry on error:
__init__(
*,
model: str | BaseChatModel | None = None,
system_prompt: str = DEFAULT_SYSTEM_PROMPT,
max_tools: int | None = None,
always_include: list[str] | None = None
) -> None
Initialize the tool selector.
| PARAMETER | DESCRIPTION |
|---|---|
model
|
Model to use for selection. If not provided, uses the agent's main model. Can be a model identifier string or BaseChatModel instance.
TYPE:
|
system_prompt
|
Instructions for the selection model.
TYPE:
|
max_tools
|
Maximum number of tools to select. If the model selects more, only the first max_tools will be used. No limit if not specified.
TYPE:
|
always_include
|
Tool names to always include regardless of selection. These do not count against the max_tools limit. |
wrap_model_call(
request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]
) -> ModelCallResult
Filter tools based on LLM selection before invoking the model via handler.
async
¶
awrap_model_call(
request: ModelRequest, handler: Callable[[ModelRequest], Awaitable[ModelResponse]]
) -> ModelCallResult
Filter tools based on LLM selection before invoking the model via handler.
Bases: AgentMiddleware
Middleware that emulates specified tools using an LLM instead of executing them.
This middleware allows selective emulation of tools for testing purposes. By default (when tools=None), all tools are emulated. You can specify which tools to emulate by passing a list of tool names or BaseTool instances.
Examples:
Emulate all tools (default behavior):
from langchain.agents.middleware import LLMToolEmulator
middleware = LLMToolEmulator()
agent = create_agent(
model="openai:gpt-4o",
tools=[get_weather, get_user_location, calculator],
middleware=[middleware],
)
Emulate specific tools by name:
Use a custom model for emulation:
Emulate specific tools by passing tool instances:
class-attribute
instance-attribute
¶
state_schema: type[StateT] = cast('type[StateT]', AgentState)
The schema for state passed to the middleware nodes.
property
¶
name: str
The name of the middleware instance.
Defaults to the class name, but can be overridden for custom naming.
Logic to run before the agent execution starts.
async
¶
Async logic to run before the agent execution starts.
Logic to run before the model is called.
async
¶
Async logic to run before the model is called.
Logic to run after the model is called.
async
¶
Async logic to run after the model is called.
wrap_model_call(
request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]
) -> ModelCallResult
Intercept and control model execution via handler callback.
The handler callback executes the model request and returns a ModelResponse.
Middleware can call the handler multiple times for retry logic, skip calling
it to short-circuit, or modify the request/response. Multiple middleware
compose with first in list as outermost layer.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Model request to execute (includes state and runtime).
TYPE:
|
handler
|
Callback that executes the model request and returns
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ModelCallResult
|
|
Examples:
Retry on error:
def wrap_model_call(self, request, handler):
for attempt in range(3):
try:
return handler(request)
except Exception:
if attempt == 2:
raise
Rewrite response:
def wrap_model_call(self, request, handler):
response = handler(request)
ai_msg = response.result[0]
return ModelResponse(
result=[AIMessage(content=f"[{ai_msg.content}]")],
structured_response=response.structured_response,
)
Error to fallback:
def wrap_model_call(self, request, handler):
try:
return handler(request)
except Exception:
return ModelResponse(result=[AIMessage(content="Service unavailable")])
Cache/short-circuit:
def wrap_model_call(self, request, handler):
if cached := get_cache(request):
return cached # Short-circuit with cached result
response = handler(request)
save_cache(request, response)
return response
Simple AIMessage return (converted automatically):
async
¶
awrap_model_call(
request: ModelRequest, handler: Callable[[ModelRequest], Awaitable[ModelResponse]]
) -> ModelCallResult
Intercept and control async model execution via handler callback.
The handler callback executes the model request and returns a ModelResponse.
Middleware can call the handler multiple times for retry logic, skip calling
it to short-circuit, or modify the request/response. Multiple middleware
compose with first in list as outermost layer.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Model request to execute (includes state and runtime).
TYPE:
|
handler
|
Async callback that executes the model request and returns
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ModelCallResult
|
ModelCallResult |
Examples:
Retry on error:
Logic to run after the agent execution completes.
async
¶
Async logic to run after the agent execution completes.
__init__(
*,
tools: list[str | BaseTool] | None = None,
model: str | BaseChatModel | None = None
) -> None
Initialize the tool emulator.
| PARAMETER | DESCRIPTION |
|---|---|
tools
|
List of tool names (str) or BaseTool instances to emulate. If None (default), ALL tools will be emulated. If empty list, no tools will be emulated. |
model
|
Model to use for emulation. Defaults to "anthropic:claude-3-5-sonnet-latest". Can be a model identifier string or BaseChatModel instance.
TYPE:
|
wrap_tool_call(
request: ToolCallRequest,
handler: Callable[[ToolCallRequest], ToolMessage | Command],
) -> ToolMessage | Command
Emulate tool execution using LLM if tool should be emulated.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Tool call request to potentially emulate.
TYPE:
|
handler
|
Callback to execute the tool (can be called multiple times).
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ToolMessage | Command
|
ToolMessage with emulated response if tool should be emulated, |
ToolMessage | Command
|
otherwise calls handler for normal execution. |
async
¶
awrap_tool_call(
request: ToolCallRequest,
handler: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command]],
) -> ToolMessage | Command
Async version of wrap_tool_call.
Emulate tool execution using LLM if tool should be emulated.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Tool call request to potentially emulate.
TYPE:
|
handler
|
Async callback to execute the tool (can be called multiple times).
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ToolMessage | Command
|
ToolMessage with emulated response if tool should be emulated, |
ToolMessage | Command
|
otherwise calls handler for normal execution. |
Bases: AgentMiddleware[ModelCallLimitState, Any]
Middleware that tracks model call counts and enforces limits.
This middleware monitors the number of model calls made during agent execution and can terminate the agent when specified limits are reached. It supports both thread-level and run-level call counting with configurable exit behaviors.
Thread-level: The middleware tracks the number of model calls and persists call count across multiple runs (invocations) of the agent.
Run-level: The middleware tracks the number of model calls made during a single run (invocation) of the agent.
Example
from langchain.agents.middleware.call_tracking import ModelCallLimitMiddleware
from langchain.agents import create_agent
# Create middleware with limits
call_tracker = ModelCallLimitMiddleware(thread_limit=10, run_limit=5, exit_behavior="end")
agent = create_agent("openai:gpt-4o", middleware=[call_tracker])
# Agent will automatically jump to end when limits are exceeded
result = await agent.invoke({"messages": [HumanMessage("Help me with a task")]})
property
¶
name: str
The name of the middleware instance.
Defaults to the class name, but can be overridden for custom naming.
Logic to run before the agent execution starts.
async
¶
Async logic to run before the agent execution starts.
async
¶
Async logic to run before the model is called.
async
¶
Async logic to run after the model is called.
wrap_model_call(
request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]
) -> ModelCallResult
Intercept and control model execution via handler callback.
The handler callback executes the model request and returns a ModelResponse.
Middleware can call the handler multiple times for retry logic, skip calling
it to short-circuit, or modify the request/response. Multiple middleware
compose with first in list as outermost layer.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Model request to execute (includes state and runtime).
TYPE:
|
handler
|
Callback that executes the model request and returns
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ModelCallResult
|
|
Examples:
Retry on error:
def wrap_model_call(self, request, handler):
for attempt in range(3):
try:
return handler(request)
except Exception:
if attempt == 2:
raise
Rewrite response:
def wrap_model_call(self, request, handler):
response = handler(request)
ai_msg = response.result[0]
return ModelResponse(
result=[AIMessage(content=f"[{ai_msg.content}]")],
structured_response=response.structured_response,
)
Error to fallback:
def wrap_model_call(self, request, handler):
try:
return handler(request)
except Exception:
return ModelResponse(result=[AIMessage(content="Service unavailable")])
Cache/short-circuit:
def wrap_model_call(self, request, handler):
if cached := get_cache(request):
return cached # Short-circuit with cached result
response = handler(request)
save_cache(request, response)
return response
Simple AIMessage return (converted automatically):
async
¶
awrap_model_call(
request: ModelRequest, handler: Callable[[ModelRequest], Awaitable[ModelResponse]]
) -> ModelCallResult
Intercept and control async model execution via handler callback.
The handler callback executes the model request and returns a ModelResponse.
Middleware can call the handler multiple times for retry logic, skip calling
it to short-circuit, or modify the request/response. Multiple middleware
compose with first in list as outermost layer.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Model request to execute (includes state and runtime).
TYPE:
|
handler
|
Async callback that executes the model request and returns
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ModelCallResult
|
ModelCallResult |
Examples:
Retry on error:
Logic to run after the agent execution completes.
async
¶
Async logic to run after the agent execution completes.
wrap_tool_call(
request: ToolCallRequest,
handler: Callable[[ToolCallRequest], ToolMessage | Command],
) -> ToolMessage | Command
Intercept tool execution for retries, monitoring, or modification.
Multiple middleware compose automatically (first defined = outermost).
Exceptions propagate unless handle_tool_errors is configured on ToolNode.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Tool call request with call
TYPE:
|
handler
|
Callable to execute the tool (can be called multiple times).
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ToolMessage | Command
|
|
The handler callable can be invoked multiple times for retry logic. Each call to handler is independent and stateless.
Examples:
Modify request before execution:
def wrap_tool_call(self, request, handler):
request.tool_call["args"]["value"] *= 2
return handler(request)
Retry on error (call handler multiple times):
def wrap_tool_call(self, request, handler):
for attempt in range(3):
try:
result = handler(request)
if is_valid(result):
return result
except Exception:
if attempt == 2:
raise
return result
Conditional retry based on response:
async
¶
awrap_tool_call(
request: ToolCallRequest,
handler: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command]],
) -> ToolMessage | Command
Intercept and control async tool execution via handler callback.
The handler callback executes the tool call and returns a ToolMessage or
Command. Middleware can call the handler multiple times for retry logic, skip
calling it to short-circuit, or modify the request/response. Multiple middleware
compose with first in list as outermost layer.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Tool call request with call
TYPE:
|
handler
|
Async callable to execute the tool and returns
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ToolMessage | Command
|
|
The handler callable can be invoked multiple times for retry logic. Each call to handler is independent and stateless.
Examples:
Async retry on error:
class-attribute
instance-attribute
¶
The schema for state passed to the middleware nodes.
__init__(
*,
thread_limit: int | None = None,
run_limit: int | None = None,
exit_behavior: Literal["end", "error"] = "end"
) -> None
Initialize the call tracking middleware.
| PARAMETER | DESCRIPTION |
|---|---|
thread_limit
|
Maximum number of model calls allowed per thread. None means no limit.
TYPE:
|
run_limit
|
Maximum number of model calls allowed per run. None means no limit.
TYPE:
|
exit_behavior
|
What to do when limits are exceeded.
- "end": Jump to the end of the agent execution and
inject an artificial AI message indicating that the limit was exceeded.
- "error": Raise a
TYPE:
|
| RAISES | DESCRIPTION |
|---|---|
ValueError
|
If both limits are |
Check model call limits before making a model call.
| PARAMETER | DESCRIPTION |
|---|---|
state
|
The current agent state containing call counts.
TYPE:
|
runtime
|
The langgraph runtime.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
dict[str, Any] | None
|
If limits are exceeded and exit_behavior is "end", returns |
dict[str, Any] | None
|
a Command to jump to the end with a limit exceeded message. Otherwise returns None. |
| RAISES | DESCRIPTION |
|---|---|
ModelCallLimitExceededError
|
If limits are exceeded and exit_behavior is "error". |
Bases: AgentMiddleware
Automatic fallback to alternative models on errors.
Retries failed model calls with alternative models in sequence until success or all models exhausted. Primary model specified in create_agent().
Example
from langchain.agents.middleware.model_fallback import ModelFallbackMiddleware
from langchain.agents import create_agent
fallback = ModelFallbackMiddleware(
"openai:gpt-4o-mini", # Try first on error
"anthropic:claude-3-5-sonnet-20241022", # Then this
)
agent = create_agent(
model="openai:gpt-4o", # Primary model
middleware=[fallback],
)
# If primary fails: tries gpt-4o-mini, then claude-3-5-sonnet
result = await agent.invoke({"messages": [HumanMessage("Hello")]})
class-attribute
instance-attribute
¶
state_schema: type[StateT] = cast('type[StateT]', AgentState)
The schema for state passed to the middleware nodes.
property
¶
name: str
The name of the middleware instance.
Defaults to the class name, but can be overridden for custom naming.
Logic to run before the agent execution starts.
async
¶
Async logic to run before the agent execution starts.
Logic to run before the model is called.
async
¶
Async logic to run before the model is called.
Logic to run after the model is called.
async
¶
Async logic to run after the model is called.
Logic to run after the agent execution completes.
async
¶
Async logic to run after the agent execution completes.
wrap_tool_call(
request: ToolCallRequest,
handler: Callable[[ToolCallRequest], ToolMessage | Command],
) -> ToolMessage | Command
Intercept tool execution for retries, monitoring, or modification.
Multiple middleware compose automatically (first defined = outermost).
Exceptions propagate unless handle_tool_errors is configured on ToolNode.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Tool call request with call
TYPE:
|
handler
|
Callable to execute the tool (can be called multiple times).
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ToolMessage | Command
|
|
The handler callable can be invoked multiple times for retry logic. Each call to handler is independent and stateless.
Examples:
Modify request before execution:
def wrap_tool_call(self, request, handler):
request.tool_call["args"]["value"] *= 2
return handler(request)
Retry on error (call handler multiple times):
def wrap_tool_call(self, request, handler):
for attempt in range(3):
try:
result = handler(request)
if is_valid(result):
return result
except Exception:
if attempt == 2:
raise
return result
Conditional retry based on response:
async
¶
awrap_tool_call(
request: ToolCallRequest,
handler: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command]],
) -> ToolMessage | Command
Intercept and control async tool execution via handler callback.
The handler callback executes the tool call and returns a ToolMessage or
Command. Middleware can call the handler multiple times for retry logic, skip
calling it to short-circuit, or modify the request/response. Multiple middleware
compose with first in list as outermost layer.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Tool call request with call
TYPE:
|
handler
|
Async callable to execute the tool and returns
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ToolMessage | Command
|
|
The handler callable can be invoked multiple times for retry logic. Each call to handler is independent and stateless.
Examples:
Async retry on error:
__init__(
first_model: str | BaseChatModel, *additional_models: str | BaseChatModel
) -> None
Initialize model fallback middleware.
| PARAMETER | DESCRIPTION |
|---|---|
first_model
|
First fallback model (string name or instance).
TYPE:
|
*additional_models
|
Additional fallbacks in order.
TYPE:
|
wrap_model_call(
request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]
) -> ModelCallResult
Try fallback models in sequence on errors.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Initial model request.
TYPE:
|
handler
|
Callback to execute the model.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ModelCallResult
|
AIMessage from successful model call. |
| RAISES | DESCRIPTION |
|---|---|
Exception
|
If all models fail, re-raises last exception. |
async
¶
awrap_model_call(
request: ModelRequest, handler: Callable[[ModelRequest], Awaitable[ModelResponse]]
) -> ModelCallResult
Try fallback models in sequence on errors (async version).
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Initial model request.
TYPE:
|
handler
|
Async callback to execute the model.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ModelCallResult
|
AIMessage from successful model call. |
| RAISES | DESCRIPTION |
|---|---|
Exception
|
If all models fail, re-raises last exception. |
Bases: AgentMiddleware
Detect and handle Personally Identifiable Information (PII) in agent conversations.
This middleware detects common PII types and applies configurable strategies to handle them. It can detect emails, credit cards, IP addresses, MAC addresses, and URLs in both user input and agent output.
Built-in PII types
email: Email addressescredit_card: Credit card numbers (validated with Luhn algorithm)ip: IP addresses (validated with stdlib)mac_address: MAC addressesurl: URLs (bothhttp/httpsand bare URLs)
Strategies
block: Raise an exception when PII is detectedredact: Replace PII with[REDACTED_TYPE]placeholdersmask: Partially mask PII (e.g.,****-****-****-1234for credit card)hash: Replace PII with deterministic hash (e.g.,<email_hash:a1b2c3d4>)
Strategy Selection Guide:
| Strategy | Preserves Identity? | Best For |
|---|---|---|
block |
N/A | Avoid PII completely |
redact |
No | General compliance, log sanitization |
mask |
No | Human readability, customer service UIs |
hash |
Yes (pseudonymous) | Analytics, debugging |
Example
from langchain.agents.middleware import PIIMiddleware
from langchain.agents import create_agent
# Redact all emails in user input
agent = create_agent(
"openai:gpt-5",
middleware=[
PIIMiddleware("email", strategy="redact"),
],
)
# Use different strategies for different PII types
agent = create_agent(
"openai:gpt-4o",
middleware=[
PIIMiddleware("credit_card", strategy="mask"),
PIIMiddleware("url", strategy="redact"),
PIIMiddleware("ip", strategy="hash"),
],
)
# Custom PII type with regex
agent = create_agent(
"openai:gpt-5",
middleware=[
PIIMiddleware("api_key", detector=r"sk-[a-zA-Z0-9]{32}", strategy="block"),
],
)
class-attribute
instance-attribute
¶
state_schema: type[StateT] = cast('type[StateT]', AgentState)
The schema for state passed to the middleware nodes.
Logic to run before the agent execution starts.
async
¶
Async logic to run before the agent execution starts.
async
¶
Async logic to run before the model is called.
async
¶
Async logic to run after the model is called.
wrap_model_call(
request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]
) -> ModelCallResult
Intercept and control model execution via handler callback.
The handler callback executes the model request and returns a ModelResponse.
Middleware can call the handler multiple times for retry logic, skip calling
it to short-circuit, or modify the request/response. Multiple middleware
compose with first in list as outermost layer.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Model request to execute (includes state and runtime).
TYPE:
|
handler
|
Callback that executes the model request and returns
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ModelCallResult
|
|
Examples:
Retry on error:
def wrap_model_call(self, request, handler):
for attempt in range(3):
try:
return handler(request)
except Exception:
if attempt == 2:
raise
Rewrite response:
def wrap_model_call(self, request, handler):
response = handler(request)
ai_msg = response.result[0]
return ModelResponse(
result=[AIMessage(content=f"[{ai_msg.content}]")],
structured_response=response.structured_response,
)
Error to fallback:
def wrap_model_call(self, request, handler):
try:
return handler(request)
except Exception:
return ModelResponse(result=[AIMessage(content="Service unavailable")])
Cache/short-circuit:
def wrap_model_call(self, request, handler):
if cached := get_cache(request):
return cached # Short-circuit with cached result
response = handler(request)
save_cache(request, response)
return response
Simple AIMessage return (converted automatically):
async
¶
awrap_model_call(
request: ModelRequest, handler: Callable[[ModelRequest], Awaitable[ModelResponse]]
) -> ModelCallResult
Intercept and control async model execution via handler callback.
The handler callback executes the model request and returns a ModelResponse.
Middleware can call the handler multiple times for retry logic, skip calling
it to short-circuit, or modify the request/response. Multiple middleware
compose with first in list as outermost layer.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Model request to execute (includes state and runtime).
TYPE:
|
handler
|
Async callback that executes the model request and returns
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ModelCallResult
|
ModelCallResult |
Examples:
Retry on error:
Logic to run after the agent execution completes.
async
¶
Async logic to run after the agent execution completes.
wrap_tool_call(
request: ToolCallRequest,
handler: Callable[[ToolCallRequest], ToolMessage | Command],
) -> ToolMessage | Command
Intercept tool execution for retries, monitoring, or modification.
Multiple middleware compose automatically (first defined = outermost).
Exceptions propagate unless handle_tool_errors is configured on ToolNode.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Tool call request with call
TYPE:
|
handler
|
Callable to execute the tool (can be called multiple times).
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ToolMessage | Command
|
|
The handler callable can be invoked multiple times for retry logic. Each call to handler is independent and stateless.
Examples:
Modify request before execution:
def wrap_tool_call(self, request, handler):
request.tool_call["args"]["value"] *= 2
return handler(request)
Retry on error (call handler multiple times):
def wrap_tool_call(self, request, handler):
for attempt in range(3):
try:
result = handler(request)
if is_valid(result):
return result
except Exception:
if attempt == 2:
raise
return result
Conditional retry based on response:
async
¶
awrap_tool_call(
request: ToolCallRequest,
handler: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command]],
) -> ToolMessage | Command
Intercept and control async tool execution via handler callback.
The handler callback executes the tool call and returns a ToolMessage or
Command. Middleware can call the handler multiple times for retry logic, skip
calling it to short-circuit, or modify the request/response. Multiple middleware
compose with first in list as outermost layer.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Tool call request with call
TYPE:
|
handler
|
Async callable to execute the tool and returns
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ToolMessage | Command
|
|
The handler callable can be invoked multiple times for retry logic. Each call to handler is independent and stateless.
Examples:
Async retry on error:
__init__(
pii_type: Literal["email", "credit_card", "ip", "mac_address", "url"] | str,
*,
strategy: Literal["block", "redact", "mask", "hash"] = "redact",
detector: Callable[[str], list[PIIMatch]] | str | None = None,
apply_to_input: bool = True,
apply_to_output: bool = False,
apply_to_tool_results: bool = False
) -> None
Initialize the PII detection middleware.
| PARAMETER | DESCRIPTION |
|---|---|
pii_type
|
Type of PII to detect. Can be a built-in type
(
TYPE:
|
strategy
|
How to handle detected PII:
TYPE:
|
detector
|
Custom detector function or regex pattern.
TYPE:
|
apply_to_input
|
Whether to check user messages before model call.
TYPE:
|
apply_to_output
|
Whether to check AI messages after model call.
TYPE:
|
apply_to_tool_results
|
Whether to check tool result messages after tool execution.
TYPE:
|
| RAISES | DESCRIPTION |
|---|---|
ValueError
|
If pii_type is not built-in and no detector is provided. |
before_model(state: AgentState, runtime: Runtime) -> dict[str, Any] | None
Check user messages and tool results for PII before model invocation.
| PARAMETER | DESCRIPTION |
|---|---|
state
|
The current agent state.
TYPE:
|
runtime
|
The langgraph runtime.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
dict[str, Any] | None
|
Updated state with PII handled according to strategy, or None if no PII detected. |
| RAISES | DESCRIPTION |
|---|---|
PIIDetectionError
|
If PII is detected and strategy is "block". |
after_model(state: AgentState, runtime: Runtime) -> dict[str, Any] | None
Check AI messages for PII after model invocation.
| PARAMETER | DESCRIPTION |
|---|---|
state
|
The current agent state.
TYPE:
|
runtime
|
The langgraph runtime.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
dict[str, Any] | None
|
Updated state with PII handled according to strategy, or None if no PII detected. |
| RAISES | DESCRIPTION |
|---|---|
PIIDetectionError
|
If PII is detected and strategy is "block". |
Bases: AgentMiddleware
Middleware that summarizes conversation history when token limits are approached.
This middleware monitors message token counts and automatically summarizes older messages when a threshold is reached, preserving recent messages and maintaining context continuity by ensuring AI/Tool message pairs remain together.
class-attribute
instance-attribute
¶
state_schema: type[StateT] = cast('type[StateT]', AgentState)
The schema for state passed to the middleware nodes.
property
¶
name: str
The name of the middleware instance.
Defaults to the class name, but can be overridden for custom naming.
Logic to run before the agent execution starts.
async
¶
Async logic to run before the agent execution starts.
async
¶
Async logic to run before the model is called.
Logic to run after the model is called.
async
¶
Async logic to run after the model is called.
wrap_model_call(
request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]
) -> ModelCallResult
Intercept and control model execution via handler callback.
The handler callback executes the model request and returns a ModelResponse.
Middleware can call the handler multiple times for retry logic, skip calling
it to short-circuit, or modify the request/response. Multiple middleware
compose with first in list as outermost layer.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Model request to execute (includes state and runtime).
TYPE:
|
handler
|
Callback that executes the model request and returns
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ModelCallResult
|
|
Examples:
Retry on error:
def wrap_model_call(self, request, handler):
for attempt in range(3):
try:
return handler(request)
except Exception:
if attempt == 2:
raise
Rewrite response:
def wrap_model_call(self, request, handler):
response = handler(request)
ai_msg = response.result[0]
return ModelResponse(
result=[AIMessage(content=f"[{ai_msg.content}]")],
structured_response=response.structured_response,
)
Error to fallback:
def wrap_model_call(self, request, handler):
try:
return handler(request)
except Exception:
return ModelResponse(result=[AIMessage(content="Service unavailable")])
Cache/short-circuit:
def wrap_model_call(self, request, handler):
if cached := get_cache(request):
return cached # Short-circuit with cached result
response = handler(request)
save_cache(request, response)
return response
Simple AIMessage return (converted automatically):
async
¶
awrap_model_call(
request: ModelRequest, handler: Callable[[ModelRequest], Awaitable[ModelResponse]]
) -> ModelCallResult
Intercept and control async model execution via handler callback.
The handler callback executes the model request and returns a ModelResponse.
Middleware can call the handler multiple times for retry logic, skip calling
it to short-circuit, or modify the request/response. Multiple middleware
compose with first in list as outermost layer.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Model request to execute (includes state and runtime).
TYPE:
|
handler
|
Async callback that executes the model request and returns
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ModelCallResult
|
ModelCallResult |
Examples:
Retry on error:
Logic to run after the agent execution completes.
async
¶
Async logic to run after the agent execution completes.
wrap_tool_call(
request: ToolCallRequest,
handler: Callable[[ToolCallRequest], ToolMessage | Command],
) -> ToolMessage | Command
Intercept tool execution for retries, monitoring, or modification.
Multiple middleware compose automatically (first defined = outermost).
Exceptions propagate unless handle_tool_errors is configured on ToolNode.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Tool call request with call
TYPE:
|
handler
|
Callable to execute the tool (can be called multiple times).
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ToolMessage | Command
|
|
The handler callable can be invoked multiple times for retry logic. Each call to handler is independent and stateless.
Examples:
Modify request before execution:
def wrap_tool_call(self, request, handler):
request.tool_call["args"]["value"] *= 2
return handler(request)
Retry on error (call handler multiple times):
def wrap_tool_call(self, request, handler):
for attempt in range(3):
try:
result = handler(request)
if is_valid(result):
return result
except Exception:
if attempt == 2:
raise
return result
Conditional retry based on response:
async
¶
awrap_tool_call(
request: ToolCallRequest,
handler: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command]],
) -> ToolMessage | Command
Intercept and control async tool execution via handler callback.
The handler callback executes the tool call and returns a ToolMessage or
Command. Middleware can call the handler multiple times for retry logic, skip
calling it to short-circuit, or modify the request/response. Multiple middleware
compose with first in list as outermost layer.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Tool call request with call
TYPE:
|
handler
|
Async callable to execute the tool and returns
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ToolMessage | Command
|
|
The handler callable can be invoked multiple times for retry logic. Each call to handler is independent and stateless.
Examples:
Async retry on error:
__init__(
model: str | BaseChatModel,
max_tokens_before_summary: int | None = None,
messages_to_keep: int = _DEFAULT_MESSAGES_TO_KEEP,
token_counter: TokenCounter = count_tokens_approximately,
summary_prompt: str = DEFAULT_SUMMARY_PROMPT,
summary_prefix: str = SUMMARY_PREFIX,
) -> None
Initialize the summarization middleware.
| PARAMETER | DESCRIPTION |
|---|---|
model
|
The language model to use for generating summaries.
TYPE:
|
max_tokens_before_summary
|
Token threshold to trigger summarization.
If
TYPE:
|
messages_to_keep
|
Number of recent messages to preserve after summarization.
TYPE:
|
token_counter
|
Function to count tokens in messages.
TYPE:
|
summary_prompt
|
Prompt template for generating summaries.
TYPE:
|
summary_prefix
|
Prefix added to system message when including summary.
TYPE:
|
Bases: AgentMiddleware[ToolCallLimitState, Any]
Middleware that tracks tool call counts and enforces limits.
This middleware monitors the number of tool calls made during agent execution and can terminate the agent when specified limits are reached. It supports both thread-level and run-level call counting with configurable exit behaviors.
Thread-level: The middleware tracks the total number of tool calls and persists call count across multiple runs (invocations) of the agent.
Run-level: The middleware tracks the number of tool calls made during a single run (invocation) of the agent.
Example
from langchain.agents.middleware.tool_call_limit import ToolCallLimitMiddleware
from langchain.agents import create_agent
# Limit all tool calls globally
global_limiter = ToolCallLimitMiddleware(thread_limit=20, run_limit=10, exit_behavior="end")
# Limit a specific tool
search_limiter = ToolCallLimitMiddleware(
tool_name="search", thread_limit=5, run_limit=3, exit_behavior="end"
)
# Use both in the same agent
agent = create_agent("openai:gpt-4o", middleware=[global_limiter, search_limiter])
result = await agent.invoke({"messages": [HumanMessage("Help me with a task")]})
Logic to run before the agent execution starts.
async
¶
Async logic to run before the agent execution starts.
async
¶
Async logic to run before the model is called.
async
¶
Async logic to run after the model is called.
wrap_model_call(
request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]
) -> ModelCallResult
Intercept and control model execution via handler callback.
The handler callback executes the model request and returns a ModelResponse.
Middleware can call the handler multiple times for retry logic, skip calling
it to short-circuit, or modify the request/response. Multiple middleware
compose with first in list as outermost layer.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Model request to execute (includes state and runtime).
TYPE:
|
handler
|
Callback that executes the model request and returns
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ModelCallResult
|
|
Examples:
Retry on error:
def wrap_model_call(self, request, handler):
for attempt in range(3):
try:
return handler(request)
except Exception:
if attempt == 2:
raise
Rewrite response:
def wrap_model_call(self, request, handler):
response = handler(request)
ai_msg = response.result[0]
return ModelResponse(
result=[AIMessage(content=f"[{ai_msg.content}]")],
structured_response=response.structured_response,
)
Error to fallback:
def wrap_model_call(self, request, handler):
try:
return handler(request)
except Exception:
return ModelResponse(result=[AIMessage(content="Service unavailable")])
Cache/short-circuit:
def wrap_model_call(self, request, handler):
if cached := get_cache(request):
return cached # Short-circuit with cached result
response = handler(request)
save_cache(request, response)
return response
Simple AIMessage return (converted automatically):
async
¶
awrap_model_call(
request: ModelRequest, handler: Callable[[ModelRequest], Awaitable[ModelResponse]]
) -> ModelCallResult
Intercept and control async model execution via handler callback.
The handler callback executes the model request and returns a ModelResponse.
Middleware can call the handler multiple times for retry logic, skip calling
it to short-circuit, or modify the request/response. Multiple middleware
compose with first in list as outermost layer.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Model request to execute (includes state and runtime).
TYPE:
|
handler
|
Async callback that executes the model request and returns
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ModelCallResult
|
ModelCallResult |
Examples:
Retry on error:
Logic to run after the agent execution completes.
async
¶
Async logic to run after the agent execution completes.
wrap_tool_call(
request: ToolCallRequest,
handler: Callable[[ToolCallRequest], ToolMessage | Command],
) -> ToolMessage | Command
Intercept tool execution for retries, monitoring, or modification.
Multiple middleware compose automatically (first defined = outermost).
Exceptions propagate unless handle_tool_errors is configured on ToolNode.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Tool call request with call
TYPE:
|
handler
|
Callable to execute the tool (can be called multiple times).
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ToolMessage | Command
|
|
The handler callable can be invoked multiple times for retry logic. Each call to handler is independent and stateless.
Examples:
Modify request before execution:
def wrap_tool_call(self, request, handler):
request.tool_call["args"]["value"] *= 2
return handler(request)
Retry on error (call handler multiple times):
def wrap_tool_call(self, request, handler):
for attempt in range(3):
try:
result = handler(request)
if is_valid(result):
return result
except Exception:
if attempt == 2:
raise
return result
Conditional retry based on response:
async
¶
awrap_tool_call(
request: ToolCallRequest,
handler: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command]],
) -> ToolMessage | Command
Intercept and control async tool execution via handler callback.
The handler callback executes the tool call and returns a ToolMessage or
Command. Middleware can call the handler multiple times for retry logic, skip
calling it to short-circuit, or modify the request/response. Multiple middleware
compose with first in list as outermost layer.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Tool call request with call
TYPE:
|
handler
|
Async callable to execute the tool and returns
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ToolMessage | Command
|
|
The handler callable can be invoked multiple times for retry logic. Each call to handler is independent and stateless.
Examples:
Async retry on error:
class-attribute
instance-attribute
¶
The schema for state passed to the middleware nodes.
__init__(
*,
tool_name: str | None = None,
thread_limit: int | None = None,
run_limit: int | None = None,
exit_behavior: Literal["end", "error"] = "end"
) -> None
Initialize the tool call limit middleware.
| PARAMETER | DESCRIPTION |
|---|---|
tool_name
|
Name of the specific tool to limit. If
TYPE:
|
thread_limit
|
Maximum number of tool calls allowed per thread.
None means no limit. Defaults to
TYPE:
|
run_limit
|
Maximum number of tool calls allowed per run.
None means no limit. Defaults to
TYPE:
|
exit_behavior
|
What to do when limits are exceeded. - "end": Jump to the end of the agent execution and inject an artificial AI message indicating that the limit was exceeded. - "error": Raise a ToolCallLimitExceededError Defaults to "end".
TYPE:
|
| RAISES | DESCRIPTION |
|---|---|
ValueError
|
If both limits are |
property
¶
name: str
The name of the middleware instance.
Includes the tool name if specified to allow multiple instances of this middleware with different tool names.
Check tool call limits before making a model call.
| PARAMETER | DESCRIPTION |
|---|---|
state
|
The current agent state containing tool call counts.
TYPE:
|
runtime
|
The langgraph runtime.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
dict[str, Any] | None
|
If limits are exceeded and exit_behavior is "end", returns |
dict[str, Any] | None
|
a Command to jump to the end with a limit exceeded message. Otherwise returns None. |
| RAISES | DESCRIPTION |
|---|---|
ToolCallLimitExceededError
|
If limits are exceeded and exit_behavior is "error". |
Increment tool call counts after a model call (when tool calls are made).
| PARAMETER | DESCRIPTION |
|---|---|
state
|
The current agent state.
TYPE:
|
runtime
|
The langgraph runtime.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
dict[str, Any] | None
|
State updates with incremented tool call counts if tool calls were made. |
Bases: Generic[StateT, ContextT]
Base middleware class for an agent.
Subclass this and implement any of the defined methods to customize agent behavior between steps in the main agent loop.
class-attribute
instance-attribute
¶
state_schema: type[StateT] = cast('type[StateT]', AgentState)
The schema for state passed to the middleware nodes.
property
¶
name: str
The name of the middleware instance.
Defaults to the class name, but can be overridden for custom naming.
Logic to run before the agent execution starts.
async
¶
Async logic to run before the agent execution starts.
Logic to run before the model is called.
async
¶
Async logic to run before the model is called.
Logic to run after the model is called.
async
¶
Async logic to run after the model is called.
wrap_model_call(
request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]
) -> ModelCallResult
Intercept and control model execution via handler callback.
The handler callback executes the model request and returns a ModelResponse.
Middleware can call the handler multiple times for retry logic, skip calling
it to short-circuit, or modify the request/response. Multiple middleware
compose with first in list as outermost layer.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Model request to execute (includes state and runtime).
TYPE:
|
handler
|
Callback that executes the model request and returns
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ModelCallResult
|
|
Examples:
Retry on error:
def wrap_model_call(self, request, handler):
for attempt in range(3):
try:
return handler(request)
except Exception:
if attempt == 2:
raise
Rewrite response:
def wrap_model_call(self, request, handler):
response = handler(request)
ai_msg = response.result[0]
return ModelResponse(
result=[AIMessage(content=f"[{ai_msg.content}]")],
structured_response=response.structured_response,
)
Error to fallback:
def wrap_model_call(self, request, handler):
try:
return handler(request)
except Exception:
return ModelResponse(result=[AIMessage(content="Service unavailable")])
Cache/short-circuit:
def wrap_model_call(self, request, handler):
if cached := get_cache(request):
return cached # Short-circuit with cached result
response = handler(request)
save_cache(request, response)
return response
Simple AIMessage return (converted automatically):
async
¶
awrap_model_call(
request: ModelRequest, handler: Callable[[ModelRequest], Awaitable[ModelResponse]]
) -> ModelCallResult
Intercept and control async model execution via handler callback.
The handler callback executes the model request and returns a ModelResponse.
Middleware can call the handler multiple times for retry logic, skip calling
it to short-circuit, or modify the request/response. Multiple middleware
compose with first in list as outermost layer.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Model request to execute (includes state and runtime).
TYPE:
|
handler
|
Async callback that executes the model request and returns
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ModelCallResult
|
ModelCallResult |
Examples:
Retry on error:
Logic to run after the agent execution completes.
async
¶
Async logic to run after the agent execution completes.
wrap_tool_call(
request: ToolCallRequest,
handler: Callable[[ToolCallRequest], ToolMessage | Command],
) -> ToolMessage | Command
Intercept tool execution for retries, monitoring, or modification.
Multiple middleware compose automatically (first defined = outermost).
Exceptions propagate unless handle_tool_errors is configured on ToolNode.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Tool call request with call
TYPE:
|
handler
|
Callable to execute the tool (can be called multiple times).
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ToolMessage | Command
|
|
The handler callable can be invoked multiple times for retry logic. Each call to handler is independent and stateless.
Examples:
Modify request before execution:
def wrap_tool_call(self, request, handler):
request.tool_call["args"]["value"] *= 2
return handler(request)
Retry on error (call handler multiple times):
def wrap_tool_call(self, request, handler):
for attempt in range(3):
try:
result = handler(request)
if is_valid(result):
return result
except Exception:
if attempt == 2:
raise
return result
Conditional retry based on response:
async
¶
awrap_tool_call(
request: ToolCallRequest,
handler: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command]],
) -> ToolMessage | Command
Intercept and control async tool execution via handler callback.
The handler callback executes the tool call and returns a ToolMessage or
Command. Middleware can call the handler multiple times for retry logic, skip
calling it to short-circuit, or modify the request/response. Multiple middleware
compose with first in list as outermost layer.
| PARAMETER | DESCRIPTION |
|---|---|
request
|
Tool call request with call
TYPE:
|
handler
|
Async callable to execute the tool and returns
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ToolMessage | Command
|
|
The handler callable can be invoked multiple times for retry logic. Each call to handler is independent and stateless.
Examples:
Async retry on error:
dataclass
¶
Bases: ContextEdit
Configuration for clearing tool outputs when token limits are exceeded.
class-attribute
instance-attribute
¶
trigger: int = 100000
Token count that triggers the edit.
class-attribute
instance-attribute
¶
clear_at_least: int = 0
Minimum number of tokens to reclaim when the edit runs.
class-attribute
instance-attribute
¶
keep: int = 3
Number of most recent tool results that must be preserved.
class-attribute
instance-attribute
¶
clear_tool_inputs: bool = False
Whether to clear the originating tool call parameters on the AI message.
class-attribute
instance-attribute
¶
List of tool names to exclude from clearing.
class-attribute
instance-attribute
¶
placeholder: str = DEFAULT_TOOL_PLACEHOLDER
Placeholder text inserted for cleared tool outputs.
apply(messages: list[AnyMessage], *, count_tokens: TokenCounter) -> None
Apply the clear-tool-uses strategy.
Bases: TypedDict
Configuration for an action requiring human in the loop.
This is the configuration format used in the HumanInTheLoopMiddleware.__init__
method.
instance-attribute
¶
allowed_decisions: list[DecisionType]
The decisions that are allowed for this action.
instance-attribute
¶
description: NotRequired[str | _DescriptionFactory]
The description attached to the request for human input.
Can be either:
- A static string describing the approval request
- A callable that dynamically generates the description based on agent state, runtime, and tool call information
Example
# Static string description
config = ToolConfig(
allowed_decisions=["approve", "reject"],
description="Please review this tool execution"
)
# Dynamic callable description
def format_tool_description(
tool_call: ToolCall,
state: AgentState,
runtime: Runtime
) -> str:
import json
return (
f"Tool: {tool_call['name']}\n"
f"Arguments:\n{json.dumps(tool_call['args'], indent=2)}"
)
config = InterruptOnConfig(
allowed_decisions=["approve", "edit", "reject"],
description=format_tool_description
)
instance-attribute
¶
args_schema: NotRequired[dict[str, Any]]
JSON schema for the args associated with the action, if edits are allowed.
dataclass
¶
Model request information for the agent.
override(**overrides: Unpack[_ModelRequestOverrides]) -> ModelRequest
Replace the request with a new request with the given overrides.
Returns a new ModelRequest instance with the specified attributes replaced.
This follows an immutable pattern, leaving the original request unchanged.
| PARAMETER | DESCRIPTION |
|---|---|
**overrides
|
Keyword arguments for attributes to override. Supported keys: - model: BaseChatModel instance - system_prompt: Optional system prompt string - messages: List of messages - tool_choice: Tool choice configuration - tools: List of available tools - response_format: Response format specification - model_settings: Additional model settings
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ModelRequest
|
New ModelRequest instance with specified overrides applied. |
Examples:
dataclass
¶
Response from model execution including messages and optional structured output.
The result will usually contain a single AIMessage, but may include an additional ToolMessage if the model used a tool for structured output.
before_model(
func: _CallableWithStateAndRuntime[StateT, ContextT] | None = None,
*,
state_schema: type[StateT] | None = None,
tools: list[BaseTool] | None = None,
can_jump_to: list[JumpTo] | None = None,
name: str | None = None
) -> (
Callable[
[_CallableWithStateAndRuntime[StateT, ContextT]],
AgentMiddleware[StateT, ContextT],
]
| AgentMiddleware[StateT, ContextT]
)
Decorator used to dynamically create a middleware with the before_model hook.
| PARAMETER | DESCRIPTION |
|---|---|
func
|
The function to be decorated. Must accept:
TYPE:
|
state_schema
|
Optional custom state schema type. If not provided, uses the default
TYPE:
|
tools
|
Optional list of additional tools to register with this middleware. |
can_jump_to
|
Optional list of valid jump destinations for conditional edges.
Valid values are:
TYPE:
|
name
|
Optional name for the generated middleware class. If not provided, uses the decorated function's name.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Callable[[_CallableWithStateAndRuntime[StateT, ContextT]], AgentMiddleware[StateT, ContextT]] | AgentMiddleware[StateT, ContextT]
|
Either an |
Callable[[_CallableWithStateAndRuntime[StateT, ContextT]], AgentMiddleware[StateT, ContextT]] | AgentMiddleware[StateT, ContextT]
|
decorator function that can be applied to a function it is wrapping. |
The decorated function should return
dict[str, Any]- State updates to merge into the agent stateCommand- A command to control flow (e.g., jump to different node)None- No state updates or flow control
Examples:
Basic usage:
@before_model
def log_before_model(state: AgentState, runtime: Runtime) -> None:
print(f"About to call model with {len(state['messages'])} messages")
With conditional jumping:
@before_model(can_jump_to=["end"])
def conditional_before_model(state: AgentState, runtime: Runtime) -> dict[str, Any] | None:
if some_condition(state):
return {"jump_to": "end"}
return None
With custom state schema:
after_model(
func: _CallableWithStateAndRuntime[StateT, ContextT] | None = None,
*,
state_schema: type[StateT] | None = None,
tools: list[BaseTool] | None = None,
can_jump_to: list[JumpTo] | None = None,
name: str | None = None
) -> (
Callable[
[_CallableWithStateAndRuntime[StateT, ContextT]],
AgentMiddleware[StateT, ContextT],
]
| AgentMiddleware[StateT, ContextT]
)
Decorator used to dynamically create a middleware with the after_model hook.
| PARAMETER | DESCRIPTION |
|---|---|
func
|
The function to be decorated. Must accept:
TYPE:
|
state_schema
|
Optional custom state schema type. If not provided, uses the
default
TYPE:
|
tools
|
Optional list of additional tools to register with this middleware. |
can_jump_to
|
Optional list of valid jump destinations for conditional edges.
Valid values are:
TYPE:
|
name
|
Optional name for the generated middleware class. If not provided, uses the decorated function's name.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Callable[[_CallableWithStateAndRuntime[StateT, ContextT]], AgentMiddleware[StateT, ContextT]] | AgentMiddleware[StateT, ContextT]
|
Either an |
Callable[[_CallableWithStateAndRuntime[StateT, ContextT]], AgentMiddleware[StateT, ContextT]] | AgentMiddleware[StateT, ContextT]
|
function that can be applied to a function. |
The decorated function should return
dict[str, Any]- State updates to merge into the agent stateCommand- A command to control flow (e.g., jump to different node)None- No state updates or flow control
Examples:
Basic usage for logging model responses:
@after_model
def log_latest_message(state: AgentState, runtime: Runtime) -> None:
print(state["messages"][-1].content)
With custom state schema:
wrap_model_call(
func: _CallableReturningModelResponse[StateT, ContextT] | None = None,
*,
state_schema: type[StateT] | None = None,
tools: list[BaseTool] | None = None,
name: str | None = None
) -> (
Callable[
[_CallableReturningModelResponse[StateT, ContextT]],
AgentMiddleware[StateT, ContextT],
]
| AgentMiddleware[StateT, ContextT]
)
Create middleware with wrap_model_call hook from a function.
Converts a function with handler callback into middleware that can intercept model calls, implement retry logic, handle errors, and rewrite responses.
| PARAMETER | DESCRIPTION |
|---|---|
func
|
Function accepting (request, handler) that calls handler(request)
to execute the model and returns
TYPE:
|
state_schema
|
Custom state schema. Defaults to
TYPE:
|
tools
|
Additional tools to register with this middleware. |
name
|
Middleware class name. Defaults to function name.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Callable[[_CallableReturningModelResponse[StateT, ContextT]], AgentMiddleware[StateT, ContextT]] | AgentMiddleware[StateT, ContextT]
|
|
Examples:
Basic retry logic:
@wrap_model_call
def retry_on_error(request, handler):
max_retries = 3
for attempt in range(max_retries):
try:
return handler(request)
except Exception:
if attempt == max_retries - 1:
raise
Model fallback:
@wrap_model_call
def fallback_model(request, handler):
# Try primary model
try:
return handler(request)
except Exception:
pass
# Try fallback model
request.model = fallback_model_instance
return handler(request)
Rewrite response content (full ModelResponse):
@wrap_model_call
def uppercase_responses(request, handler):
response = handler(request)
ai_msg = response.result[0]
return ModelResponse(
result=[AIMessage(content=ai_msg.content.upper())],
structured_response=response.structured_response,
)
Simple AIMessage return (converted automatically):
wrap_tool_call(
func: _CallableReturningToolResponse | None = None,
*,
tools: list[BaseTool] | None = None,
name: str | None = None
) -> Callable[[_CallableReturningToolResponse], AgentMiddleware] | AgentMiddleware
Create middleware with wrap_tool_call hook from a function.
Converts a function with handler callback into middleware that can intercept tool calls, implement retry logic, monitor execution, and modify responses.
| PARAMETER | DESCRIPTION |
|---|---|
func
|
Function accepting (request, handler) that calls
handler(request) to execute the tool and returns final
TYPE:
|
tools
|
Additional tools to register with this middleware. |
name
|
Middleware class name. Defaults to function name.
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
Callable[[_CallableReturningToolResponse], AgentMiddleware] | AgentMiddleware
|
|
Examples:
Retry logic:
@wrap_tool_call
def retry_on_error(request, handler):
max_retries = 3
for attempt in range(max_retries):
try:
return handler(request)
except Exception:
if attempt == max_retries - 1:
raise
Async retry logic:
@wrap_tool_call
async def async_retry(request, handler):
for attempt in range(3):
try:
return await handler(request)
except Exception:
if attempt == 2:
raise
Modify request:
@wrap_tool_call
def modify_args(request, handler):
request.tool_call["args"]["value"] *= 2
return handler(request)
Short-circuit with cached result:
dataclass
¶
Model request information for the agent.
override(**overrides: Unpack[_ModelRequestOverrides]) -> ModelRequest
Replace the request with a new request with the given overrides.
Returns a new ModelRequest instance with the specified attributes replaced.
This follows an immutable pattern, leaving the original request unchanged.
| PARAMETER | DESCRIPTION |
|---|---|
**overrides
|
Keyword arguments for attributes to override. Supported keys: - model: BaseChatModel instance - system_prompt: Optional system prompt string - messages: List of messages - tool_choice: Tool choice configuration - tools: List of available tools - response_format: Response format specification - model_settings: Additional model settings
TYPE:
|
| RETURNS | DESCRIPTION |
|---|---|
ModelRequest
|
New ModelRequest instance with specified overrides applied. |
Examples: