Hardening a Python LangChain Agent with Seven Built-In Middlewares

Getting a LangChain agent to answer questions in a notebook takes minutes. Getting one you would trust in production is harder: it must not loop until it drains your API budget, forward a customer's card number to the model provider, or send an email nobody approved. LangChain 1.0 addresses these operational concerns with middleware, a layer of hooks around the agent loop that you configure as a simple list. This guide builds the mental model, then applies seven middlewares to a Gemini-backed Python agent and closes with a custom one, so you can see exactly what each hook changes and when to reach for it.

Checkpoints around the agent loop

Think of an airport. The goal is to fly from one city to another, yet around that single action sit a series of checkpoints: check-in confirms identity, security scans luggage, the gate verifies boarding passes and baggage claim takes over after landing. None of them flies the plane, and the pilot does not screen bags. Each layer does one job before or after the core action.

Middleware applies the same idea to agents. The core of an agent is a loop: call the model, let it choose tools, run them and repeat until the model returns a final answer. Middleware inserts checkpoints around that loop without editing it:

• Stripping card numbers before text reaches the LLM is the security scanner.
• Requiring a person to approve an outgoing email is the boarding gate.
• Halting after ten model calls to cap spending is a circuit breaker.

If you work in TypeScript, the companion piece on LangChain guardrails and middleware covers the same ideas from the JavaScript side; this guide stays in Python and focuses on the concrete built-in classes.

The hooks middleware can use

The loop exposes hooks at each stage, and a middleware attaches to one or more of them:

• before_agent and after_agent run once, at the very start and the very end of an invocation.
• before_model and after_model fire each time the loop is about to call, or has just called, the model.
• wrap_model_call and wrap_tool_call wrap the call itself, so they can retry it, substitute it or serve it from a cache.

That is the whole mental model. You attach middleware by passing a list to create_agent, as in this example combining email redaction, a call cap and tool retries:

agent = create_agent(
    model=model,
    tools=[my_tool],
    middleware=[
        PIIMiddleware("email", strategy="redact"),
        ModelCallLimitMiddleware(run_limit=5),
        ToolRetryMiddleware(max_retries=3),
    ],
)

The order of the list matters. Middlewares are applied in sequence, like the layers of an onion wrapped around the agent, so a redaction step listed first sees raw input before anything else does.

Setup and a baseline agent

Middleware needs LangChain 1.0 or later, so install with the -U flag to upgrade any older copy. The examples use Gemini through its free tier; you can create a key in Google AI Studio.

!pip install -qU langchain langchain-google-genai

Import os and the Gemini chat model class:

import os
from langchain_google_genai import ChatGoogleGenerativeAI

Then configure the model. The snippet sets the key inline for demonstration only; in real code, export GOOGLE_API_KEY in your environment or load it from a secrets manager instead of placing it in source. Temperature zero keeps the outputs repeatable:

os.environ["GOOGLE_API_KEY"] = "YOUR_GEMINI_API_KEY_HERE"
model = ChatGoogleGenerativeAI(model="gemini-3.5-flash-lite", temperature=0)

The subject for every experiment is a minimal agent with no middleware. It needs create_agent and the tool decorator:

from langchain.agents import create_agent
from langchain_core.tools import tool

It has one fake weather tool that always reports sunshine, and it is invoked with a single user message:

@tool
def get_weather(city: str) -> str:
    """Get the current weather for a city."""
    return f"The weather in {city} is 31°C and sunny."agent = create_agent(model=model, tools=[get_weather])result = agent.invoke(
    {"messages": [{"role": "user", "content": "What's the weather in Bengaluru?"}]}
)
print(result["messages"][-1].text)

Each section below wraps one more checkpoint around this agent.

1. SummarizationMiddleware for bounded memory

In long conversations the message history keeps growing until it exceeds the context window, and every extra token is billed. SummarizationMiddleware checks the size of the history in before_model; when it passes a threshold, it condenses older messages into a summary and keeps only the most recent ones verbatim.

from langchain.agents.middleware import SummarizationMiddleware

The configuration below uses the same model to write summaries, triggers at 10 messages and keeps the latest 4 untouched. The test builds a fake history of six cities (twelve messages), then asks which city came first:

agent = create_agent(
    model=model,
    tools=[get_weather],
    middleware=[
        SummarizationMiddleware(
            model=model,               # which LLM writes the summary
            trigger=("messages", 10),  # summarize when history hits 10 messages
            keep=("messages", 4),      # keep the 4 most recent messages intact
        ),
    ],
)# Simulate a long conversation
long_history = []
for city in ["Delhi", "Mumbai", "Chennai", "Kolkata", "Pune", "Jaipur"]:
    long_history.append({"role": "user", "content": f"What's the weather in {city}?"})
    long_history.append({"role": "assistant", "content": f"The weather in {city} is sunny."})
long_history.append({"role": "user", "content": "Which city did I ask about first?"})print("Messages passed IN:", len(long_history))   # 13result = agent.invoke({"messages": long_history})
print("Final answer:", result["messages"][-1].text)
print("Messages now in state:", len(result["messages"]))   # 6

Thirteen messages go in and six remain afterwards: the summary, the four preserved messages and the new reply. The model still answers "Delhi", because that fact was carried into the summary. Counting messages makes the behavior easy to see in a demo, but in production a token-based trigger such as ("tokens", 3000) or a fraction of the context window such as ("fraction", 0.8) tracks actual cost and limits far better. Be aware that summaries are lossy: exact figures or identifiers mentioned early in a conversation may not survive.

2. Call limits as cost circuit breakers

The most expensive agent failure is the runaway loop, where the model and tools keep calling each other and spend money for minutes before anyone notices. Two middlewares put a hard ceiling on this:

from langchain.agents.middleware import ModelCallLimitMiddleware, ToolCallLimitMiddleware

Here the model is capped at three calls per run, with exit_behavior="end" so the agent stops cleanly instead of raising an exception, and tools are capped at two calls. The prompt deliberately asks for six cities one by one:

agent = create_agent(
    model=model,
    tools=[get_weather],
    middleware=[
        # "end" = stop gracefully instead of raising an error
        ModelCallLimitMiddleware(run_limit=3, exit_behavior="end"),
        ToolCallLimitMiddleware(run_limit=2),
    ],
)result = agent.invoke(
    {"messages": [{"role": "user", "content":
        "Get the weather for Delhi, Mumbai, Chennai, Kolkata, Pune and Jaipur one by one."}]}
)
print(result["messages"][-1].text)

The agent wants six lookups, reaches its limits and finishes gracefully with the partial results it has. A thread_limit parameter is also available to cap calls across a whole conversation thread rather than a single run. These two lines are cheap insurance; choose limits comfortably above what legitimate requests need so they only fire on genuine runaways.

3. ToolRetryMiddleware for flaky dependencies

Real tools fail: HTTP calls time out and connections drop. ToolRetryMiddleware uses wrap_tool_call to catch failures and retry them with exponential backoff.

from langchain.agents.middleware import ToolRetryMiddleware

To demonstrate it, a stock price tool counts its attempts and raises ConnectionError on the first two before succeeding. The middleware allows up to three retries, starting with a one-second delay and doubling each time:

attempt_counter = {"count": 0}@tool
def flaky_stock_price(symbol: str) -> str:
    """Get the current stock price for a ticker symbol."""
    attempt_counter["count"] += 1
    print(f"  [tool called — attempt #{attempt_counter['count']}]")
    if attempt_counter["count"] < 3:
        raise ConnectionError("API timeout — please retry")
    return f"{symbol} is trading at ₹2,845.50"agent = create_agent(
    model=model,
    tools=[flaky_stock_price],
    middleware=[
        ToolRetryMiddleware(
            max_retries=3,       # retry a failed tool up to 3 times
            initial_delay=1.0,   # wait 1s before first retry
            backoff_factor=2.0,  # double the wait each time: 1s, 2s, 4s
        ),
    ],
)result = agent.invoke(
    {"messages": [{"role": "user", "content": "What's the price of RELIANCE stock?"}]}
)
print(result["messages"][-1].text)

The tool fails twice, the middleware waits and tries again, and the third attempt succeeds. From the agent's perspective nothing went wrong. Retries are only safe for idempotent operations such as reads; retrying a tool that charges a card or sends a message could repeat the side effect.

4. ModelFallbackMiddleware for provider outages

The same resilience idea can protect the model call. When the primary model fails, because of rate limiting or an outage for example, this middleware retries the request against backup models in the order you list them:

from langchain.agents.middleware import ModelFallbackMiddleware

The example keeps the lighter Gemini model as primary and adds a second Gemini model as backup:

backup_model = ChatGoogleGenerativeAI(model="gemini-3.5-flash", temperature=0)agent = create_agent(
    model=model,  # primary: gemini-3.5-flash-lite
    tools=[get_weather],
    middleware=[ModelFallbackMiddleware(backup_model)],
)

While the primary is healthy you will notice nothing, which is the intent. The value appears only on the day your provider has trouble. For stronger protection, consider a backup from a different provider, since an outage often affects every model behind the same API.

5. PIIMiddleware for sensitive data

Frequently you do not want emails, card numbers or IP addresses sent to the model provider at all. PIIMiddleware scans text in before_model, before the model sees it, and applies one of four strategies: redact, mask, hash or block.

from langchain.agents.middleware import PIIMiddleware

This agent has no tools. It replaces email addresses entirely with a placeholder and masks card numbers so that only the last four digits remain, with both rules applied to user input:

agent = create_agent(
    model=model,
    tools=[],
    middleware=[
        # Replace emails entirely with [REDACTED_EMAIL]
        PIIMiddleware("email", strategy="redact", apply_to_input=True),
        # Mask credit cards — keeps last 4 digits
        PIIMiddleware("credit_card", strategy="mask", apply_to_input=True),
    ],
)result = agent.invoke(
    {"messages": [{"role": "user", "content":
        "Draft a support reply to priya.sharma@example.com confirming her card "
        "4111-1111-1111-1234 was not charged."}]}
)
print(result["messages"][-1].text)

The model drafts its reply without ever receiving the real address or the full card number. You can also register a custom PII type with your own regular expression, for instance to block any text that looks like an internal API key:

PIIMiddleware("api_key", detector=r"sk-[a-zA-Z0-9]{32}", strategy="block")

Pattern-based detection catches well-formed values, not every creative spelling, so treat it as one layer of defense rather than a compliance guarantee.

6. HumanInTheLoopMiddleware for approval gates

Some actions are too consequential for full autonomy: sending emails, deleting records or making payments. HumanInTheLoopMiddleware halts the agent immediately before a sensitive tool runs, waits for a person's decision and then continues.

This works differently from the previous middlewares. A paused agent is not terminated. A checkpointer saves its full state, and the thread ID acts as the key for finding and resuming that run later. The reviewer can approve the call, edit its arguments or reject it.

The imports bring in the middleware, an in-memory checkpointer from LangGraph and the Command type used to resume:

from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

The agent below has a send_email tool, marks it for interruption and stores paused state in InMemorySaver. The first invocation, on thread demo-1, asks the agent to email a manager:

@tool
def send_email(to: str, subject: str, body: str) -> str:
    """Send an email to the given recipient."""
    return f"Email sent to {to} with subject '{subject}'"agent = create_agent(
    model=model,
    tools=[send_email],
    middleware=[
        # Pause and ask a human whenever the agent wants to call send_email
        HumanInTheLoopMiddleware(interrupt_on={"send_email": True}),
    ],
    checkpointer=InMemorySaver(),   # where the paused state is saved
)config = {"configurable": {"thread_id": "demo-1"}}# Step 1: run — the agent PAUSES before sending
result = agent.invoke(
    {"messages": [{"role": "user", "content":
        "Send an email to boss@company.com saying the report is ready."}]},
    config,
)
print("Agent paused! It wants to run:")
print(result["__interrupt__"])

Execution halts before the tool runs. The __interrupt__ entry in the result shows the pending call with its recipient, subject and body, and nothing has been sent. Approval is sent as a Command on the same thread, which resumes the paused run:

# Step 2: approve and resume
result = agent.invoke(
    Command(resume={"decisions": [{"type": "approve"}]}),
    config,  # same thread_id -> resumes the paused run
)
print(result["messages"][-1].text)

Instead of approve, you can send reject with a reason or edit with changed arguments. In a real application, the pause is where you would render an approval screen. Two practical notes: InMemorySaver loses state when the process restarts, so production systems need a persistent checkpointer; and the resume payload format has changed between LangChain versions, so confirm it in the middleware reference for the version you run.

7. Writing your own middleware with a decorator

When nothing built in fits, a custom middleware is small, because every hook has a matching decorator. The imports are the before_model decorator and the AgentState type:

from langchain.agents.middleware import before_model, AgentState

This example logs how many messages are about to be sent on each model call and is added to the list like any prebuilt middleware:

@before_model
def log_before_model(state: AgentState, runtime) -> None:
    print(f"  [middleware] Calling model with {len(state['messages'])} messages")
    # Returning None = observe only.
    # Returning a dict would UPDATE the agent's state (e.g., trim messages)
    return Noneagent = create_agent(
    model=model,
    tools=[get_weather],
    middleware=[log_before_model],  # plugs in like any prebuilt middleware
)

The return value is the important design choice. Returning None means the middleware only observes. Returning a dictionary updates the agent state, which is how you would trim messages, inject context or enforce a custom guardrail. Decorators exist for the other hooks too: @before_agent, @after_model, @wrap_model_call, @wrap_tool_call, plus @dynamic_prompt for building system prompts at runtime.

Key takeaways

• Middleware separates operational concerns from agent logic: the loop stays the same while checkpoints are layered around it as a plain list passed to create_agent.
• Order the list deliberately; redaction should run before anything that forwards text elsewhere.
• Summarization and call limits control cost, tool retries and model fallback control reliability, PII handling controls data exposure and human-in-the-loop controls irreversible actions.
• Each has limits worth remembering: summaries lose detail, retries are unsafe for non-idempotent tools, regex detection is incomplete and in-memory checkpoints vanish on restart.
• The wider catalog, including TodoListMiddleware, LLMToolSelectorMiddleware and ContextEditingMiddleware, is documented in the official reference.