Middleware that removes incomplete tool calls from the messages sent to the model.

class langgraph_agent_toolkit.agents.components.middlewares.sanitize_history.SanitizeHistoryMiddleware

Bases: AgentMiddleware

Repair broken tool-call/tool-result pairing in the messages before each model call.

Mirrors the custom create_react_agent’s sanitize_chat_history, fixing both directions that providers reject: an AIMessage tool call with no ToolMessage (call without result) has the dangling call stripped, and a ToolMessage with no requesting AIMessage tool call (orphaned result, e.g. after trimming/summarization) is dropped. Only the messages sent to the model are sanitized; persisted state is left untouched.

wrap_model_call(request, handler)

Intercept and control model execution via handler callback.

Async version is awrap_model_call

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.

Parameters:

• request (ModelRequest) – Model request to execute (includes state and runtime).

• handler (Callable[[ModelRequest], ModelResponse]) –

  Callback that executes the model request and returns ModelResponse.

  Call this to execute the model.

  Can be called multiple times for retry logic.

  Can skip calling it to short-circuit.

Returns:

The model call result.

Return type:

ModelResponse

Examples

!!! example “Retry on error”

```python def wrap_model_call(self, request, handler):

for attempt in range(3):

try:

return handler(request)

except Exception:

if attempt == 2:

raise

```

!!! example “Rewrite response”

```python 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,

)

```

!!! example “Error to fallback”

```python def wrap_model_call(self, request, handler):

try:

return handler(request)

except Exception:

return ModelResponse(result=[AIMessage(content=”Service unavailable”)])

```

!!! example “Cache/short-circuit”

```python 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

```

!!! example “Simple AIMessage return (converted automatically)”

```python def wrap_model_call(self, request, handler):

response = handler(request) # Can return AIMessage directly for simple cases return AIMessage(content=”Simplified response”)

```

async awrap_model_call(request, handler)

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.

Parameters:

• request (ModelRequest) – Model request to execute (includes state and runtime).

• handler (Callable[[ModelRequest], Awaitable[ModelResponse]]) –

  Async callback that executes the model request and returns ModelResponse.

  Call this to execute the model.

  Can be called multiple times for retry logic.

  Can skip calling it to short-circuit.

Returns:

The model call result.

Return type:

ModelResponse

Examples

!!! example “Retry on error”

```python async def awrap_model_call(self, request, handler):

for attempt in range(3):

try:

return await handler(request)

except Exception:

if attempt == 2:

raise

```

async aafter_agent(state, runtime)

Async logic to run after the agent execution completes.

Parameters:

• state (StateT) – The current agent state.

• runtime (Runtime[ContextT]) – The runtime context.

Returns:

Agent state updates to apply after agent execution.

Return type:

dict[str, Any] | None

async aafter_model(state, runtime)

Async logic to run after the model is called.

Parameters:

• state (StateT) – The current agent state.

• runtime (Runtime[ContextT]) – The runtime context.

Returns:

Agent state updates to apply after model call.

Return type:

dict[str, Any] | None

async abefore_agent(state, runtime)

Async logic to run before the agent execution starts.

Parameters:

• state (StateT) – The current agent state.

• runtime (Runtime[ContextT]) – The runtime context.

Returns:

Agent state updates to apply before agent execution.

Return type:

dict[str, Any] | None

async abefore_model(state, runtime)

Async logic to run before the model is called.

Parameters:

• state (StateT) – The agent state.

• runtime (Runtime[ContextT]) – The runtime context.

Returns:

Agent state updates to apply before model call.

Return type:

dict[str, Any] | None

after_agent(state, runtime)

Logic to run after the agent execution completes.

Parameters:

• state (StateT) – The current agent state.

• runtime (Runtime[ContextT]) – The runtime context.

Returns:

Agent state updates to apply after agent execution.

Return type:

dict[str, Any] | None

after_model(state, runtime)

Logic to run after the model is called.

Parameters:

• state (StateT) – The current agent state.

• runtime (Runtime[ContextT]) – The runtime context.

Returns:

Agent state updates to apply after model call.

Return type:

dict[str, Any] | None

async awrap_tool_call(request, handler)

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.

Parameters:

• request (ToolCallRequest) –

  Tool call request with call dict, BaseTool, state, and runtime.

  Access state via request.state and runtime via request.runtime.

• handler (Callable[[ToolCallRequest], Awaitable[ToolMessage | Command[Any]]]) –

  Async callable to execute the tool and returns ToolMessage or Command.

  Call this to execute the tool.

  Can be called multiple times for retry logic.

  Can skip calling it to short-circuit.

Returns:

ToolMessage or Command (the final result).

Return type:

ToolMessage | Command[Any]

The handler Callable can be invoked multiple times for retry logic.

Each call to handler is independent and stateless.

Examples

!!! example “Async retry on error”

```python async def awrap_tool_call(self, request, handler):

for attempt in range(3):

try:

result = await handler(request) if is_valid(result):

return result

except Exception:

if attempt == 2:

raise

return result

```

```python async def awrap_tool_call(self, request, handler):

if cached := await get_cache_async(request):

return ToolMessage(content=cached, tool_call_id=request.tool_call[“id”])

result = await handler(request) await save_cache_async(request, result) return result

```

before_agent(state, runtime)

Logic to run before the agent execution starts.

Parameters:

• state (StateT) – The current agent state.

• runtime (Runtime[ContextT]) – The runtime context.

Returns:

Agent state updates to apply before agent execution.

Return type:

dict[str, Any] | None

before_model(state, runtime)

Logic to run before the model is called.

Parameters:

• state (StateT) – The current agent state.

• runtime (Runtime[ContextT]) – The runtime context.

Returns:

Agent state updates to apply before model call.

Return type:

dict[str, Any] | None

property name: str

The name of the middleware instance.

Defaults to the class name, but can be overridden for custom naming.

state_schema

alias of _DefaultAgentState

transformers = ()

Stream transformer factories registered by the middleware.

Each entry is a scope-aware factory invoked as factory(scope) so every invocation receives a fresh instance. Factories are merged with the transformers argument of [create_agent][langchain.agents.create_agent] at graph compile time, after the ToolCallTransformer and before any user-supplied entries.

wrap_tool_call(request, handler)

Intercept tool execution for retries, monitoring, or modification.

Async version is awrap_tool_call

Multiple middleware compose automatically (first defined = outermost).

Exceptions propagate unless handle_tool_errors is configured on ToolNode.

Parameters:

• request (ToolCallRequest) –

  Tool call request with call dict, BaseTool, state, and runtime.

  Access state via request.state and runtime via request.runtime.

• handler (Callable[[ToolCallRequest], ToolMessage | Command[Any]]) – Callable to execute the tool (can be called multiple times).

Returns:

ToolMessage or Command (the final result).

Return type:

ToolMessage | Command[Any]

The handler Callable can be invoked multiple times for retry logic.

Each call to handler is independent and stateless.

Examples

!!! example “Modify request before execution”

```python def wrap_tool_call(self, request, handler):

modified_call = {

**request.tool_call, “args”: {

**request.tool_call[“args”], “value”: request.tool_call[“args”][“value”] * 2,

},

} request = request.override(tool_call=modified_call) return handler(request)

```

!!! example “Retry on error (call handler multiple times)”

```python 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

```

!!! example “Conditional retry based on response”

```python def wrap_tool_call(self, request, handler):

for attempt in range(3):

result = handler(request) if isinstance(result, ToolMessage) and result.status != “error”:

return result

if attempt < 2:

continue

return result

```

tools

Additional tools registered by the middleware.