langgraph_agent_toolkit.agents.components.middlewares package

Reusable agent middleware that brings custom create_react_agent features to native create_agent.

class langgraph_agent_toolkit.agents.components.middlewares.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.

class langgraph_agent_toolkit.agents.components.middlewares.ImmediateGenerationMiddleware(model_call_limit=None, instruction=None)

Bases: AgentMiddleware

Force a final, tool-free answer on the last allowed model call of a run.

Mirrors the custom create_react_agent’s immediate_generation router: rather than running out of steps mid-tool-loop (returning “need more steps” or hitting the recursion limit), the model is asked to synthesize a direct answer from what it already has once it reaches the budget. The tool results are always kept — only tools and the system message are changed — so the model answers from the data it already collected.

The budget is the number of model calls already made in the current run (AI messages since the latest human message), so the middleware needs no extra state. It should fire only as a graceful fallback near the recursion limit, not cut off legitimate multi-step work, so model_call_limit defaults to about half the recursion limit (a model→tools cycle is ~2 steps): with the default recursion limit of 64 that is 32 model calls.

Parameters:

• model_call_limit (int | None)

• instruction (str | None)

__init__(model_call_limit=None, instruction=None)

Parameters:

• model_call_limit (int | None)

• instruction (str | 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.

class langgraph_agent_toolkit.agents.components.middlewares.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.

class langgraph_agent_toolkit.agents.components.middlewares.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.

class langgraph_agent_toolkit.agents.components.middlewares.TrimMessagesMiddleware(max_messages=None)

Bases: AgentMiddleware

Bound the model’s input to the last max_messages messages, non-destructively.

Ports the custom create_react_agent’s pre_model_hook trimming to native create_agent: each model call sees at most max_messages recent messages (trimmed to start on a human turn and to keep the system prompt), while the full history stays in state. Use it to bound text growth in long conversations — ContextEditingMiddleware only bounds tool output.

max_messages defaults to settings.DEFAULT_MAX_MESSAGE_HISTORY_LENGTH and can be overridden per agent. It counts messages (not tokens). Best for conversational / text-heavy agents; for a single turn that makes very many tool calls, prefer a larger value (or compose with SanitizeHistoryMiddleware) so the current turn’s tool results are not cut.

Parameters:

max_messages (int | None)

__init__(max_messages=None)

Parameters:

max_messages (int | 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.

Submodules

• ClearIntermediateToolCallsMiddleware
  • ClearIntermediateToolCallsMiddleware.__init__()
  • ClearIntermediateToolCallsMiddleware.wrap_model_call()
  • ClearIntermediateToolCallsMiddleware.awrap_model_call()
  • ClearIntermediateToolCallsMiddleware.aafter_agent()
  • ClearIntermediateToolCallsMiddleware.aafter_model()
  • ClearIntermediateToolCallsMiddleware.abefore_agent()
  • ClearIntermediateToolCallsMiddleware.abefore_model()
  • ClearIntermediateToolCallsMiddleware.after_agent()
  • ClearIntermediateToolCallsMiddleware.after_model()
  • ClearIntermediateToolCallsMiddleware.awrap_tool_call()
  • ClearIntermediateToolCallsMiddleware.before_agent()
  • ClearIntermediateToolCallsMiddleware.before_model()
  • ClearIntermediateToolCallsMiddleware.name
  • ClearIntermediateToolCallsMiddleware.state_schema
  • ClearIntermediateToolCallsMiddleware.transformers
  • ClearIntermediateToolCallsMiddleware.wrap_tool_call()
  • ClearIntermediateToolCallsMiddleware.tools
• ImmediateGenerationMiddleware
  • ImmediateGenerationMiddleware.__init__()
  • ImmediateGenerationMiddleware.wrap_model_call()
  • ImmediateGenerationMiddleware.awrap_model_call()
  • ImmediateGenerationMiddleware.aafter_agent()
  • ImmediateGenerationMiddleware.aafter_model()
  • ImmediateGenerationMiddleware.abefore_agent()
  • ImmediateGenerationMiddleware.abefore_model()
  • ImmediateGenerationMiddleware.after_agent()
  • ImmediateGenerationMiddleware.after_model()
  • ImmediateGenerationMiddleware.awrap_tool_call()
  • ImmediateGenerationMiddleware.before_agent()
  • ImmediateGenerationMiddleware.before_model()
  • ImmediateGenerationMiddleware.name
  • ImmediateGenerationMiddleware.state_schema
  • ImmediateGenerationMiddleware.transformers
  • ImmediateGenerationMiddleware.wrap_tool_call()
  • ImmediateGenerationMiddleware.tools
• SanitizeHistoryMiddleware
  • SanitizeHistoryMiddleware.wrap_model_call()
  • SanitizeHistoryMiddleware.awrap_model_call()
  • SanitizeHistoryMiddleware.aafter_agent()
  • SanitizeHistoryMiddleware.aafter_model()
  • SanitizeHistoryMiddleware.abefore_agent()
  • SanitizeHistoryMiddleware.abefore_model()
  • SanitizeHistoryMiddleware.after_agent()
  • SanitizeHistoryMiddleware.after_model()
  • SanitizeHistoryMiddleware.awrap_tool_call()
  • SanitizeHistoryMiddleware.before_agent()
  • SanitizeHistoryMiddleware.before_model()
  • SanitizeHistoryMiddleware.name
  • SanitizeHistoryMiddleware.state_schema
  • SanitizeHistoryMiddleware.transformers
  • SanitizeHistoryMiddleware.wrap_tool_call()
  • SanitizeHistoryMiddleware.tools
• TokenTrimMiddleware
  • TokenTrimMiddleware.__init__()
  • TokenTrimMiddleware.wrap_model_call()
  • TokenTrimMiddleware.awrap_model_call()
  • TokenTrimMiddleware.aafter_agent()
  • TokenTrimMiddleware.aafter_model()
  • TokenTrimMiddleware.abefore_agent()
  • TokenTrimMiddleware.abefore_model()
  • TokenTrimMiddleware.after_agent()
  • TokenTrimMiddleware.after_model()
  • TokenTrimMiddleware.awrap_tool_call()
  • TokenTrimMiddleware.before_agent()
  • TokenTrimMiddleware.before_model()
  • TokenTrimMiddleware.name
  • TokenTrimMiddleware.state_schema
  • TokenTrimMiddleware.transformers
  • TokenTrimMiddleware.wrap_tool_call()
  • TokenTrimMiddleware.tools
• TrimMessagesMiddleware
  • TrimMessagesMiddleware.__init__()
  • TrimMessagesMiddleware.wrap_model_call()
  • TrimMessagesMiddleware.awrap_model_call()
  • TrimMessagesMiddleware.aafter_agent()
  • TrimMessagesMiddleware.aafter_model()
  • TrimMessagesMiddleware.abefore_agent()
  • TrimMessagesMiddleware.abefore_model()
  • TrimMessagesMiddleware.after_agent()
  • TrimMessagesMiddleware.after_model()
  • TrimMessagesMiddleware.awrap_tool_call()
  • TrimMessagesMiddleware.before_agent()
  • TrimMessagesMiddleware.before_model()
  • TrimMessagesMiddleware.name
  • TrimMessagesMiddleware.state_schema
  • TrimMessagesMiddleware.transformers
  • TrimMessagesMiddleware.wrap_tool_call()
  • TrimMessagesMiddleware.tools