Middleware that trims the model’s input to a bounded token budget (view-only).

class langgraph_agent_toolkit.agents.components.middlewares.token_trim.TokenTrimMiddleware(max_tokens=None, token_counter=count_tokens_approximately)

Bases: AgentMiddleware

Bound the model’s input to a token budget, non-destructively.

The token-counting companion to TrimMessagesMiddleware (which bounds by message count). Each model call sees at most max_tokens tokens of recent history (trimmed to start on a human turn and to keep the system prompt), while the full history stays in state. Compose the two to cap both message count and token size, as the custom create_react_agent’s pre_model_hook did.

Parameters:

• max_tokens (int | None) – Token budget for the model’s message view. Defaults to settings.DEFAULT_MAX_TOKENS_HISTORY_LENGTH. None (the toolkit default) disables trimming — set it, or pass max_tokens, to a budget appropriate for your model.

• token_counter (Callable[[list[BaseMessage]], int] | Callable[[BaseMessage], int] | BaseLanguageModel) – How to count tokens — a per-message or per-list callable, or a chat model. Defaults to count_tokens_approximately (fast, model-agnostic, no extra dependency). Pass a tiktoken-backed counter (or the model itself) for an exact count.

Note

On create_agent the system prompt is never part of request.messages (it lives on request.system_message and is prepended at call time), so it is always preserved regardless of the budget.

__init__(max_tokens=None, token_counter=count_tokens_approximately)

Parameters:

• max_tokens (int | None)

• token_counter (Callable[[list[BaseMessage]], int] | Callable[[BaseMessage], int] | BaseLanguageModel)

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.