Middleware that drops repeated/intermediate tool calls from earlier turns to save tokens.

class langgraph_agent_toolkit.agents.components.middlewares.clear_intermediate_tool_calls.ClearIntermediateToolCallsMiddleware(by=None, keep_last_n=None, exclude_tools=None, enabled=None)

Bases: AgentMiddleware

Reduce tokens by keeping only the most recent result(s) of each tool from previous turns.

Ports the _clean_intermediate_tool_calls history-stripping used in downstream agents: the current turn (everything from the latest human message onward) is preserved in full, while in earlier turns each tool keeps only its last keep_last_n call/result pair(s).

Defaults come from Settings (CLEAR_INTERMEDIATE_TOOL_CALLS*) and can be overridden per agent. With by="name_args" (the default) only genuinely identical repeat calls are collapsed, so distinct-argument calls are preserved; by="name" is the aggressive variant that collapses all repeats of a tool regardless of arguments. exclude_tools are never deduped — use it for stateful tools whose every call matters. Only the messages sent to the model are affected; persisted state is left untouched.

Parameters:

• by (Literal['name_args', 'name'] | None)

• keep_last_n (int | None)

• exclude_tools (Collection[str] | None)

• enabled (bool | None)

__init__(by=None, keep_last_n=None, exclude_tools=None, enabled=None)

Parameters:

• by (Literal['name_args', 'name'] | None)

• keep_last_n (int | None)

• exclude_tools (Collection[str] | None)

• enabled (bool | None)

Return type:

None

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.