Prebuilt middleware

on_error Callable[[Exception, ToolCallRequest], str | list[ContentBlock] | None]

Sync handler called for each exception raised by tool execution. Return content (a str or list of content blocks) to convert the exception into a ToolMessage(status="error"). Return None or omit a return statement to let the exception propagate. Used on the sync path and, unless aon_error is given, on the async path.

aon_error Callable[[Exception, ToolCallRequest], Awaitable[str | list[ContentBlock] | None]]

Optional async handler, used on the async execution path. Falls back to on_error when not provided.

tools list[BaseTool | str]

Optional list of tools or tool names to apply error handling to. If None, applies to all tools.

The on_error handler receives the exception and the ToolCallRequest (which includes the tool call dict with name, args, and call ID). Return None for exceptions you do not want to handle, and they will propagate normally.

from langchain.agents import create_agent
from langchain.agents.middleware import ToolErrorMiddleware, ToolRetryMiddleware


def on_error(exc: Exception, request: ToolCallRequest) -> str | None:
    # Surface ValueError to the model so it can correct the input
    if isinstance(exc, ValueError):
        return f"`{request.tool_call['name']}` failed: {type(exc).__name__}. Fix the input and retry."
    # Let all other exceptions propagate (halts the run)
    return None


# Async-only usage
async def aon_error(exc: Exception, request: ToolCallRequest) -> str | None:
    if isinstance(exc, ConnectionError):
        return f"Tool `{request.tool_call['name']}` encountered a connection error."
    return None


agent = create_agent(
    model="gpt-5.5",
    tools=[search_tool, database_tool],
    middleware=[
        # Place retry inner so exceptions reach ToolErrorMiddleware after retries are exhausted
        ToolErrorMiddleware(on_error=on_error, tools=["search_tool"]),
        ToolRetryMiddleware(max_retries=3, on_failure="error"),
    ],
)

# Async-only: pass aon_error alone (do not pass on_error)
async_agent = create_agent(
    model="gpt-5.5",
    tools=[api_tool],
    middleware=[ToolErrorMiddleware(aon_error=aon_error)],
)

max_retries number default: 2

Maximum number of retry attempts after the initial call (3 total attempts with default)

tools list[BaseTool | str]

Optional list of tools or tool names to apply retry logic to. If None, applies to all tools.

retry_on tuple[type[Exception], ...] | callable default: default_retry_on

Either a tuple of exception types to retry on, or a callable that takes an exception and returns True if it should be retried. With langchain>=1.3.16, the default retries retryable model errors and all unclassified exceptions, and no longer retries model errors marked non-retryable.

on_failure string | callable default: continue

Behavior when all retries are exhausted. Options:

• 'continue' (default) - Return a ToolMessage with error details, allowing the LLM to handle the failure
• 'error' - Re-raise the exception, stopping agent execution
• Custom callable - Function that takes the exception and returns a string for the ToolMessage content

Deprecated values: 'return_message' (use 'continue' instead) and 'raise' (use 'error' instead).

backoff_factor number default: 2.0

Multiplier for exponential backoff. Each retry waits initial_delay * (backoff_factor ** retry_number) seconds. Set to 0.0 for constant delay.

initial_delay number default: 1.0

Initial delay in seconds before first retry

max_delay number default: 60.0

Maximum delay in seconds between retries (caps exponential backoff growth)

jitter boolean default: true

Whether to add random jitter (±25%) to delay to avoid thundering herd

The middleware automatically retries failed tool calls with exponential backoff.

Key configuration:

• max_retries - Number of retry attempts (default: 2)
• backoff_factor - Multiplier for exponential backoff (default: 2.0)
• initial_delay - Starting delay in seconds (default: 1.0)
• max_delay - Cap on delay growth (default: 60.0)
• jitter - Add random variation (default: True)

Failure handling:

• on_failure='continue' (default) - Return error message
• on_failure='error' - Re-raise exception
• Custom function - Function returning error message

from langchain.agents import create_agent
from langchain.agents.middleware import ToolRetryMiddleware


agent = create_agent(
    model="gpt-5.5",
    tools=[search_tool, database_tool, api_tool],
    middleware=[
        ToolRetryMiddleware(
            max_retries=3,
            backoff_factor=2.0,
            initial_delay=1.0,
            max_delay=60.0,
            jitter=True,
            tools=["api_tool"],
            retry_on=(ConnectionError, TimeoutError),
            on_failure="continue",
        ),
    ],
)

max_retries number default: 2

Maximum number of retry attempts after the initial call (3 total attempts with default)

retry_on tuple[type[Exception], ...] | callable default: default_retry_on

Either a tuple of exception types to retry on, or a callable that takes an exception and returns True if it should be retried. With langchain>=1.3.16, the default retries retryable model errors and all unclassified exceptions, and no longer retries model errors marked non-retryable.

on_failure string | callable default: continue

Behavior when all retries are exhausted. Options:

• 'continue' (default) - Return an AIMessage with error details, allowing the agent to potentially handle the failure gracefully
• 'error' - Re-raise the exception (stops agent execution)
• Custom callable - Function that takes the exception and returns a string for the AIMessage content

backoff_factor number default: 2.0

Multiplier for exponential backoff. Each retry waits initial_delay * (backoff_factor ** retry_number) seconds. Set to 0.0 for constant delay.

initial_delay number default: 1.0

Initial delay in seconds before first retry

max_delay number default: 60.0

Maximum delay in seconds between retries (caps exponential backoff growth)

jitter boolean default: true

Whether to add random jitter (±25%) to delay to avoid thundering herd

The middleware automatically retries failed model calls with exponential backoff.

from langchain.agents import create_agent
from langchain.agents.middleware import ModelRetryMiddleware


# Basic usage with default settings (2 retries, exponential backoff)
agent = create_agent(
    model="gpt-5.5",
    tools=[search_tool],
    middleware=[ModelRetryMiddleware()],
)

# Custom exception filtering
class TimeoutError(Exception):
    """Custom exception for timeout errors."""
    pass

class ConnectionError(Exception):
    """Custom exception for connection errors."""
    pass

# Retry specific exceptions only
retry = ModelRetryMiddleware(
    max_retries=4,
    retry_on=(TimeoutError, ConnectionError),
    backoff_factor=1.5,
)


def should_retry(error: Exception) -> bool:
    # Only retry on rate limit errors
    if isinstance(error, TimeoutError):
        return True
    # Or check for specific HTTP status codes
    if hasattr(error, "status_code"):
        return error.status_code in (429, 503)
    return False

retry_with_filter = ModelRetryMiddleware(
    max_retries=3,
    retry_on=should_retry,
)

# Return error message instead of raising
retry_continue = ModelRetryMiddleware(
    max_retries=4,
    on_failure="continue",  # Return AIMessage with error instead of raising
)

# Custom error message formatting
def format_error(error: Exception) -> str:
    return f"Model call failed: {error}. Please try again later."

retry_with_formatter = ModelRetryMiddleware(
    max_retries=4,
    on_failure=format_error,
)

# Constant backoff (no exponential growth)
constant_backoff = ModelRetryMiddleware(
    max_retries=5,
    backoff_factor=0.0,  # No exponential growth
    initial_delay=2.0,  # Always wait 2 seconds
)

# Raise exception on failure
strict_retry = ModelRetryMiddleware(
    max_retries=2,
    on_failure="error",  # Re-raise exception instead of returning message
)

from langchain.chat_models import init_chat_model

custom_profile = {
    "max_input_tokens": 100_000,
    # ...
}
model = init_chat_model("gpt-5.5", profile=custom_profile)

model string | BaseChatModel required

Model for generating summaries. Can be a model identifier string (e.g., 'openai:gpt-5.4-mini') or a BaseChatModel instance. See init_chat_model for more information.

trigger ContextSize | TriggerClause | list[ContextSize | TriggerClause] | None

Condition(s) for triggering summarization. Can be:

• A single ContextSize tuple (the specified threshold must be met)
• A single TriggerClause dict (all specified thresholds must be met - AND logic)
• A list mixing either form (any item must be met - OR logic)

Supported thresholds are:

• fraction (float): Fraction of model's context size (0-1)
• tokens (int): Absolute token count
• messages (int): Message count

A ContextSize tuple expresses exactly one threshold. A TriggerClause dict can include one or more thresholds, e.g. {"tokens": 4000, "messages": 10}, and all thresholds in the dict must be met (AND).

Each TriggerClause dict must specify at least one threshold. If trigger is not provided, summarization will not trigger automatically.

See the API reference for ContextSize and TriggerClause for more information.

keep ContextSize default: ('messages', 20)

How much context to preserve after summarization. Specify exactly one of:

• fraction (float): Fraction of model's context size to keep (0-1)
• tokens (int): Absolute token count to keep
• messages (int): Number of recent messages to keep

See the API reference for ContextSize for more information.

token_counter function

Custom token counting function. Defaults to character-based counting.

summary_prompt string

Custom prompt template for summarization. Uses built-in template if not specified. The template should include {messages} placeholder where conversation history will be inserted.

trim_tokens_to_summarize number default: 4000

Maximum number of tokens to include when generating the summary. Messages will be trimmed to fit this limit before summarization.

summary_prefix string

Deprecated: Use summary_prompt to provide the full prompt instead.

max_tokens_before_summary number

Deprecated: Use trigger: ("tokens", value) instead. Token threshold for triggering summarization.

messages_to_keep number

Deprecated: Use keep: ("messages", value) instead. Recent messages to preserve.

The summarization middleware monitors message token counts and automatically summarizes older messages when thresholds are reached.

Trigger conditions control when summarization runs:

• A single threshold triggers when that threshold is met
• A trigger clause with multiple thresholds triggers only when all thresholds are met (AND logic)
• A list of trigger conditions triggers when any item is met (OR logic)
• Each threshold can use fraction (of model's context size), tokens (absolute count), or messages (message count)

Keep condition control how much context to preserve (specify exactly one):

• fraction - Fraction of model's context size to keep
• tokens - Absolute token count to keep
• messages - Number of recent messages to keep

from langchain.agents import create_agent
from langchain.agents.middleware import SummarizationMiddleware


# Single condition: trigger if tokens >= 4000
agent = create_agent(
    model="gpt-5.5",
    tools=[your_weather_tool, your_calculator_tool],
    middleware=[
        SummarizationMiddleware(
            model="gpt-5.4-mini",
            trigger=("tokens", 4000),
            keep=("messages", 20),
        ),
    ],
)

# Multiple conditions: trigger if number of tokens >= 3000 OR messages >= 6
agent2 = create_agent(
    model="gpt-5.5",
    tools=[your_weather_tool, your_calculator_tool],
    middleware=[
        SummarizationMiddleware(
            model="gpt-5.4-mini",
            trigger=[
                ("tokens", 3000),
                ("messages", 6),
            ],
            keep=("messages", 20),
        ),
    ],
)

# AND logic: trigger only when tokens >= 4000 AND messages >= 10
agent3 = create_agent(
    model="gpt-5.5",
    tools=[your_weather_tool, your_calculator_tool],
    middleware=[
        SummarizationMiddleware(
            model="gpt-5.4-mini",
            trigger={"tokens": 4000, "messages": 10},
            keep=("messages", 20),
        ),
    ],
)

# Combine AND and OR: trigger if (tokens >= 5000 AND messages >= 3)
# OR (tokens >= 3000 AND messages >= 6)
agent4 = create_agent(
    model="gpt-5.5",
    tools=[your_weather_tool, your_calculator_tool],
    middleware=[
        SummarizationMiddleware(
            model="gpt-5.4-mini",
            trigger=[
                {"tokens": 5000, "messages": 3},
                {"tokens": 3000, "messages": 6},
            ],
            keep=("messages", 20),
        ),
    ],
)

# Using fractional limits
agent5 = create_agent(
    model="gpt-5.5",
    tools=[your_weather_tool, your_calculator_tool],
    middleware=[
        SummarizationMiddleware(
            model="gpt-5.4-mini",
            trigger=("fraction", 0.8),
            keep=("fraction", 0.3),
        ),
    ],
)

tool_name string

Name of specific tool to limit. If not provided, limits apply to all tools globally.

thread_limit number

Maximum tool calls across all runs in a thread (conversation). Persists across multiple invocations with the same thread ID. Requires a checkpointer to maintain state. None means no thread limit.

run_limit number

Maximum tool calls per single invocation (one user message → response cycle). Resets with each new user message. None means no run limit.

Note: At least one of thread_limit or run_limit must be specified.

exit_behavior string default: continue

Behavior when limit is reached:

• 'continue' (default) - Block exceeded tool calls with error messages, let other tools and the model continue. The model decides when to end based on the error messages.
• 'error' - Raise a ToolCallLimitExceededError exception, stopping execution immediately
• 'end' - Stop execution immediately with a ToolMessage and AI message for the exceeded tool call. Only works when limiting a single tool; raises NotImplementedError if other tools have pending calls.

Specify limits with:

• Thread limit - Max calls across all runs in a conversation (requires checkpointer)
• Run limit - Max calls per single invocation (resets each turn)

Exit behaviors:

• 'continue' (default) - Block exceeded calls with error messages, agent continues
• 'error' - Raise exception immediately
• 'end' - Stop with ToolMessage + AI message (single-tool scenarios only)

from langchain.agents import create_agent
from langchain.agents.middleware import ToolCallLimitMiddleware


global_limiter = ToolCallLimitMiddleware(thread_limit=20, run_limit=10)
search_limiter = ToolCallLimitMiddleware(tool_name="search", thread_limit=5, run_limit=3)
database_limiter = ToolCallLimitMiddleware(tool_name="query_database", thread_limit=10)
strict_limiter = ToolCallLimitMiddleware(tool_name="scrape_webpage", run_limit=2, exit_behavior="error")

agent = create_agent(
    model="gpt-5.5",
    tools=[search_tool, database_tool, scraper_tool],
    middleware=[global_limiter, search_limiter, database_limiter, strict_limiter],
)

searchable_tools list[str | BaseTool]

Tools to defer behind the provider's tool search, given by name or instance. Deferred tools are withheld from the model until its search surfaces them. Tools constructed with extras={"defer_loading": True} are deferred regardless of this option; if searchable_tools is omitted, only those pre-marked tools are deferred.

The middleware opts-in all tools included in searchable_tools for deferral and search. A tool can also opt into deferral at construction time by setting extras={"defer_loading": True}.

from langchain.agents import create_agent
from langchain.agents.middleware import ProviderToolSearchMiddleware
from langchain.tools import tool


# Marked `defer_loading` at construction, so it's deferred on its own —
# no need to list it in `searchable_tools`.
@tool(extras={"defer_loading": True})
def send_email(to: str) -> str:
    """Send an email."""
    return "sent"


agent = create_agent(
    model="anthropic:claude-opus-4-8",
    tools=[send_email],
    middleware=[ProviderToolSearchMiddleware()],
)

workspace_root str | Path | None

Base directory for the shell session. If omitted, a temporary directory is created when the agent starts and removed when it ends.

startup_commands tuple[str, ...] | list[str] | str | None

Optional commands executed sequentially after the session starts

shutdown_commands tuple[str, ...] | list[str] | str | None

Optional commands executed before the session shuts down

execution_policy BaseExecutionPolicy | None

Execution policy controlling timeouts, output limits, and resource configuration. Options:

• HostExecutionPolicy - Full host access (default); best for trusted environments where the agent already runs inside a container or VM
• DockerExecutionPolicy - Launches a separate Docker container for each agent run, providing harder isolation
• CodexSandboxExecutionPolicy - Reuses the Codex CLI sandbox for additional syscall/filesystem restrictions

redaction_rules tuple[RedactionRule, ...] | list[RedactionRule] | None

Optional redaction rules to sanitize command output before returning it to the model.

tool_description str | None

Optional override for the registered shell tool description

shell_command Sequence[str] | str | None

Optional shell executable (string) or argument sequence used to launch the persistent session. Defaults to /bin/bash.

env Mapping[str, Any] | None

Optional environment variables to supply to the shell session. Values are coerced to strings before command execution.

The middleware provides a single persistent shell session that agents can use to execute commands sequentially.

Execution policies:

• HostExecutionPolicy (default) - Native execution with full host access
• DockerExecutionPolicy - Isolated Docker container execution
• CodexSandboxExecutionPolicy - Sandboxed execution via Codex CLI

from langchain.agents import create_agent
from langchain.agents.middleware import (
    ShellToolMiddleware,
    HostExecutionPolicy,
    DockerExecutionPolicy,
    RedactionRule,
)


# Basic shell tool with host execution
agent = create_agent(
    model="gpt-5.5",
    tools=[search_tool],
    middleware=[
        ShellToolMiddleware(
            workspace_root="/workspace",
            execution_policy=HostExecutionPolicy(),
        ),
    ],
)

# Docker isolation with startup commands
agent_docker = create_agent(
    model="gpt-5.5",
    tools=[],
    middleware=[
        ShellToolMiddleware(
            workspace_root="/workspace",
            startup_commands=["pip install requests", "export PYTHONPATH=/workspace"],
            execution_policy=DockerExecutionPolicy(
                image="python:3.11-slim",
                command_timeout=60.0,
            ),
        ),
    ],
)

# With output redaction (applied post execution)
agent_redacted = create_agent(
    model="gpt-5.5",
    tools=[],
    middleware=[
        ShellToolMiddleware(
            workspace_root="/workspace",
            redaction_rules=[
                RedactionRule(pii_type="api_key", detector=r"sk-[a-zA-Z0-9]{32}"),
            ],
        ),
    ],
)

root_path str required

Root directory to search. All file operations are relative to this path.

use_ripgrep bool default: True

Whether to use ripgrep for search. Falls back to Python regex if ripgrep is unavailable.

max_file_size_mb int default: 10

Maximum file size to search in MB. Files larger than this are skipped.

The middleware adds two search tools to agents:

Glob tool - Fast file pattern matching:

• Supports patterns like **/*.py, src/**/*.ts
• Returns matching file paths sorted by modification time

Grep tool - Content search with regex:

• Full regex syntax support
• Filter by file patterns with include parameter
• Three output modes: files_with_matches, content, count

from langchain.agents import create_agent
from langchain.agents.middleware import FilesystemFileSearchMiddleware
from langchain.messages import HumanMessage


agent = create_agent(
    model="gpt-5.5",
    tools=[],
    middleware=[
        FilesystemFileSearchMiddleware(
            root_path="/workspace",
            use_ripgrep=True,
            max_file_size_mb=10,
        ),
    ],
)

# Agent can now use glob_search and grep_search tools
result = agent.invoke({
    "messages": [HumanMessage("Find all Python files containing 'async def'")]
})

# The agent will use:
# 1. glob_search(pattern="**/*.py") to find Python files
# 2. grep_search(pattern="async def", include="*.py") to find async functions

edits list[ContextEdit] default: [ClearToolUsesEdit()]

List of ContextEdit strategies to apply

token_count_method string default: approximate

Token counting method. Options: 'approximate' or 'model'

ClearToolUsesEdit options:

trigger number default: 100000

Token count that triggers the edit. When the conversation exceeds this token count, older tool outputs will be cleared.

clear_at_least number default: 0

Minimum number of tokens to reclaim when the edit runs. If set to 0, clears as much as needed.

keep number default: 3

Number of most recent tool results that must be preserved. These will never be cleared.

clear_tool_inputs boolean default: False

Whether to clear the originating tool call parameters on the AI message. When True, tool call arguments are replaced with empty objects.

exclude_tools list[string] default: ()

List of tool names to exclude from clearing. These tools will never have their outputs cleared.

placeholder string default: [cleared]

Placeholder text inserted for cleared tool outputs. This replaces the original tool message content.

The middleware applies context editing strategies when token limits are reached. The most common strategy is ClearToolUsesEdit, which clears older tool results while preserving recent ones.

How it works:

1. Monitor token count in conversation
2. When threshold is reached, clear older tool outputs
3. Keep most recent N tool results
4. Optionally preserve tool call arguments for context

from langchain.agents import create_agent
from langchain.agents.middleware import ContextEditingMiddleware, ClearToolUsesEdit


agent = create_agent(
    model="gpt-5.5",
    tools=[search_tool, your_calculator_tool, database_tool],
    middleware=[
        ContextEditingMiddleware(
            edits=[
                ClearToolUsesEdit(
                    trigger=2000,
                    keep=3,
                    clear_tool_inputs=False,
                    exclude_tools=[],
                    placeholder="[cleared]",
                ),
            ],
        ),
    ],
)

tools list[str | BaseTool]

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. If array with tool names/instances, only those tools will be emulated.

model string | BaseChatModel

Model to use for generating emulated tool responses. Can be a model identifier string (e.g., 'google_genai:gemini-3.6-flash') or a BaseChatModel instance. Defaults to the agent's model if not specified. See init_chat_model for more information.

The middleware uses an LLM to generate plausible responses for tool calls instead of executing the actual tools.

from langchain.agents import create_agent
from langchain.agents.middleware import LLMToolEmulator
from langchain.tools import tool


@tool
def get_weather(location: str) -> str:
    """Get the current weather for a location."""
    return f"Weather in {location}"

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """Send an email."""
    return "Email sent"


# Emulate all tools (default behavior)
agent = create_agent(
    model="gpt-5.5",
    tools=[get_weather, send_email],
    middleware=[LLMToolEmulator()],
)

# Emulate specific tools only
agent2 = create_agent(
    model="gpt-5.5",
    tools=[get_weather, send_email],
    middleware=[LLMToolEmulator(tools=["get_weather"])],
)

# Use custom model for emulation
agent4 = create_agent(
    model="gpt-5.5",
    tools=[get_weather, send_email],
    middleware=[LLMToolEmulator(model="claude-sonnet-4-6")],
)