agents/middleware

224 notes in this subject, read out of this brain and free to use. This is page 2 of 4.

onCreated callback example

stream.submit( { messages: [{ type: "human", content: "What is quantum computing?" }] }, { onCreated(run) { console.log("Run created:", run.runId); // Chain a follow-up stream.submit({ messages: [{ type: "human", content: "Give me a simple analogy." }], }); }, } );

Message queue best practices

Best practices for message queuing include: limiting queue size to avoid degrading user experience (warn at threshold like 10 items), showing queue position by numbering items, preserving input focus after submission, animating transitions smoothly from queue to message list, handling errors gracefully without blocking subsequent entries, and debouncing rapid programmatic submissions with small delays between messages.

Message queue use cases

Message queuing solves friction in several scenarios: batch questions (asking five related questions at once), follow-up chains (submitting clarifications while agent is working), automated testing sequences (programmatically sending prompts to validate agent behavior), and data entry workflows (feeding structured inputs one after another).

Message queue is an agent UX primitive

Message queuing is an agent UX primitive rather than a cosmetic chat feature. The SDK keeps track of the queue as part of the stream controller, so your UI can show pending work, cancel stale requests, and keep the composer active while the current run continues.

Detecting interrupts in checkpoint timeline

To detect if a checkpoint contains human-in-the-loop interrupts, check if any task in checkpoint.tasks has interrupts: checkpoint.tasks?.some((t) => t.interrupts && t.interrupts.length > 0). Checkpoints with interrupts represent moments where the agent stopped and waited for human input.

Time travel combined with human-in-the-loop

Time travel is especially powerful when combined with human-in-the-loop patterns. If a human reviewer rejects an agent's action at an interrupt, they can resume from the checkpoint before the action was taken and provide corrective input.

Human-in-the-Loop middleware overview

The Human-in-the-Loop (HITL) middleware lets you add human oversight to agent tool calls. When a model proposes an action that might require review (such as writing to a file or executing SQL), the middleware can pause execution and wait for a decision. It checks each tool call against a configurable policy and issues an interrupt to halt execution if intervention is needed. The graph state is saved using LangGraph's persistence layer, so execution can pause safely and resume later.

HITL decision types

The HITL middleware defines four built-in ways a human can respond to an interrupt: approve (approve the tool call as-is and execute it without changes), edit (modify the tool call before execution), reject (deny the requested action with optional feedback), and respond (respond directly as the tool, used for 'ask user' style tools).

HITL configuration with interrupt_on mapping

To use HITL, add the HumanInTheLoopMiddleware to the agent's middleware list when creating the agent. Configure it with a mapping of tool actions to the decision types allowed for each action. Values can be: True (interrupt with default config), False (auto-approve), or an InterruptOnConfig object. The middleware will interrupt execution when a tool call matches an action in the mapping.

HITL requires checkpointing

Human-in-the-loop requires checkpointing to handle interrupts and persist graph state across pauses. In production, use a persistent checkpointer like AsyncPostgresSaver or MongoDBSaver. For testing or prototyping, use InMemorySaver. When invoking the agent, pass a config that includes the thread ID to associate execution with a conversation thread.

HITL reject vs respond distinction

Use reject when denying the requested action. Use respond only when the human is acting as the tool, such as answering an ask_user prompt. Do not use respond to deny side-effecting tools, because its message is treated as a successful tool result.

HITL configuration parameters

Python HumanInTheLoopMiddleware configuration: interrupt_on (dict, required) - Mapping of tool names to approval configs; description_prefix (string, default 'Tool execution requires approval') - Prefix for action request descriptions. InterruptOnConfig options: allowed_decisions (list[string]) - List of allowed decisions: 'approve', 'edit', 'reject', or 'respond'; description (string | callable) - Static string or callable function for custom description; when (callable) - Optional predicate that receives ToolCallRequest and returns True to interrupt or False to auto-approve. JavaScript configuration: interruptOn (object, required) - Mapping of tool names to approval configs; Tool approval config: allowAccept (boolean, default false) - Whether approval is allowed; allowEdit (boolean, default false) - Whether editing is allowed; allowRespond (boolean, default false) - Whether responding/rejection is allowed.

Conditional interrupts with when predicate

To pause only some tool calls, add a when predicate to a tool's InterruptOnConfig. The predicate receives a ToolCallRequest and returns True to interrupt or False to auto-approve. When the when predicate returns False, the call runs without interrupting. When it returns True, or when when is omitted, the call pauses as usual. Calls that evaluate to False are never added to the interrupt batch, so a reviewer only sees the actions that need a decision. Conditional interrupts require langchain>=1.3.3.

HITL invoke with version and thread_id

When invoking an agent with HITL, use version='v2' for GraphOutput with interrupts attribute. Provide a config with thread_id to associate execution with a conversation thread for pausing and resuming. Example: config = {"configurable": {"thread_id": "some_id"}}. Call agent.invoke() with version='v2' parameter.

HITL interrupt response structure

An interrupt is a GraphOutput containing an interrupts attribute with Interrupt objects. Each Interrupt contains value with action_requests (list of actions requiring review, each with name, arguments, and description) and review_configs (list of review configurations, each with action_name and allowed_decisions).

HITL approve decision example

To approve a tool call as-is: agent.invoke(Command(resume={"decisions": [{"type": "approve"}]}), config=config, version="v2") in Python or await agent.invoke(new Command({resume: {decisions: [{type: "approve"}]}}), config) in JavaScript.

HITL edit decision with edited_action

To modify a tool call before execution, use type 'edit' with edited_action containing name (tool name to call) and args (arguments to pass). Example: {"type": "edit", "edited_action": {"name": "new_tool_name", "args": {"key1": "new_value"}}}.

HITL reject decision with message

To deny a tool call, use type 'reject' with optional message parameter explaining why the action was rejected. The message is added to the conversation as feedback. When omitted, the middleware uses a default rejection message. For side-effecting tools, provide domain-specific feedback about whether the agent should abandon the action, ask a follow-up question, or try a safer alternative.

HITL respond decision for ask_user tools

Use respond type when the tool is a placeholder for human input, such as an ask_user tool. The message content is returned directly as the tool result; the tool itself is not executed. The message becomes a successful ToolMessage to the agent. Do not use respond to deny a proposed action, because it tells the model the tool completed successfully.

HITL multiple decisions ordering

When multiple actions are under review, provide a decision for each action in the same order as they appear in the interrupt request. Decisions must match the order of actions, with one decision per action.

HITL execution lifecycle

The HITL middleware defines an after_model hook that runs after the model generates a response but before any tool calls are executed: (1) The agent invokes the model to generate a response. (2) The middleware inspects the response for tool calls. (3) If any calls require human input, the middleware builds a HITLRequest with action_requests and review_configs and calls interrupt. (4) The agent waits for human decisions. (5) Based on HITLResponse decisions, the middleware executes approved or edited calls, synthesizes ToolMessages for rejected calls, returns human replies directly as ToolMessages for respond decisions, and resumes execution.

HITL editing guidelines

When editing tool arguments, make changes conservatively. Significant modifications to the original arguments may cause the model to re-evaluate its approach and potentially execute the tool multiple times or take unexpected actions.

HITL interrupt with Python example

Example Python code for HITL configuration: from langchain.agents import create_agent from langchain.agents.middleware import HumanInTheLoopMiddleware from langgraph.checkpoint.memory import InMemorySaver agent = create_agent( model="gpt-5.5", tools=[write_file, execute_sql, read_data], middleware=[ HumanInTheLoopMiddleware( interrupt_on={ "write_file": True, "execute_sql": {"allowed_decisions": ["approve", "reject"]}, "read_data": False, }, description_prefix="Tool execution pending approval", ), ], checkpointer=InMemorySaver(), )

HITL interrupt with JavaScript example

Example JavaScript code for HITL configuration: import { createAgent, humanInTheLoopMiddleware } from "langchain"; import { MemorySaver } from "@langchain/langgraph"; const agent = createAgent({ model: "gpt-5.5", tools: [writeFileTool, executeSQLTool, readDataTool], middleware: [ humanInTheLoopMiddleware({ interruptOn: { write_file: true, execute_sql: { allowedDecisions: ["approve", "reject"], description: "🚨 SQL execution requires DBA approval", }, read_data: false, }, descriptionPrefix: "Tool execution pending approval", }), ], checkpointer: new MemorySaver(), });

HITL conditional interrupts Python example

Example Python code for conditional interrupts: from langchain.agents import create_agent from langchain.agents.middleware import HumanInTheLoopMiddleware, ToolCallRequest from langgraph.checkpoint.memory import InMemorySaver def writes_outside_workspace(request: ToolCallRequest) -> bool: path = request.tool_call["args"].get("path", "") return not path.startswith("/workspace/") def is_write_query(request: ToolCallRequest) -> bool: query = request.tool_call["args"].get("query", "") return not query.lstrip().upper().startswith("SELECT") agent = create_agent( model="gpt-5.5", tools=[write_file, execute_sql, read_data], middleware=[ HumanInTheLoopMiddleware( interrupt_on={ "write_file": { "allowed_decisions": ["approve", "edit", "reject"], "when": writes_outside_workspace, }, "execute_sql": { "allowed_decisions": ["approve", "reject"], "when": is_write_query, }, }, ), ], checkpointer=InMemorySaver(), )

HITL invoke and interrupt response Python example

Example Python code for invoking agent with HITL and responding to interrupts: from langgraph.types import Command config = {"configurable": {"thread_id": "some_id"}} result = agent.invoke( {"messages": [{"role": "user", "content": "Delete old records from the database"}]}, config=config, version="v2", ) print(result.interrupts) agent.invoke( Command(resume={"decisions": [{"type": "approve"}]}), config=config, version="v2", )

HITL invoke and interrupt response JavaScript example

Example JavaScript code for invoking agent with HITL and responding to interrupts: import { HumanMessage } from "@langchain/core/messages"; import { Command } from "@langchain/langgraph"; const config = { configurable: { thread_id: "some_id" } }; const result = await agent.invoke( { messages: [new HumanMessage("Delete old records from the database")] }, config ); console.log(result.__interrupt__); await agent.invoke( new Command({ resume: { decisions: [{ type: "approve" }] } }), config );

HITL multiple decisions Python example

Example Python code for providing multiple decisions: {"decisions": [{"type": "approve"}, {"type": "edit", "edited_action": {"name": "tool_name", "args": {"param": "new_value"}}}, {"type": "reject", "message": "This action is not allowed"}]}

HITL multiple decisions JavaScript example

Example JavaScript code for providing multiple decisions: {decisions: [{type: "approve"}, {type: "edit", editedAction: {name: "tool_name", args: {param: "new_value"}}}, {type: "reject", message: "This action is not allowed"}]}

Example: Accessing store in MCP interceptor

```python from dataclasses import dataclass from langchain_mcp_adapters.client import MultiServerMCPClient from langchain_mcp_adapters.interceptors import MCPToolCallRequest from langchain.agents import create_agent from langgraph.store.memory import InMemoryStore @dataclass class Context: user_id: str async def personalize_search(request: MCPToolCallRequest, handler): """Personalize MCP tool calls using stored preferences.""" runtime = request.runtime user_id = runtime.context.user_id store = runtime.store prefs = store.get(("preferences",), user_id) if prefs and request.name == "search": modified_args = { **request.args, "language": prefs.value.get("language", "en"), "limit": prefs.value.get("result_limit", 10), } request = request.override(args=modified_args) return await handler(request) client = MultiServerMCPClient({...}, tool_interceptors=[personalize_search]) tools = await client.get_tools() agent = create_agent( "gpt-5.5", tools, context_schema=Context, store=InMemoryStore() ) ``` This example shows how to access the store from an interceptor to read user preferences and personalize tool calls.

MCP tool interceptors purpose

Interceptors bridge the gap between MCP servers (which run as separate processes) and LangGraph runtime information like the store, context, and agent state. Interceptors also provide middleware-like control over tool calls: you can modify requests, implement retries, add headers dynamically, or short-circuit execution entirely.

Accessing runtime context in MCP interceptors

When MCP tools are used within a LangChain agent, interceptors receive access to ToolRuntime context via request.runtime. This provides access to tool_call_id, state, config, and store, enabling patterns for accessing user data, persisting information, and controlling agent behavior.

MCP interceptor state updates with Command

Interceptors can return Command objects to update agent state or control graph execution flow. Use Command(update={...}, goto="node_name") to update state and switch nodes, or Command(update={...}, goto="__end__") to end execution early.

MCP interceptor composition pattern

Multiple interceptors compose in 'onion' order—the first interceptor in the list is the outermost layer. Execution flows: first interceptor before -> ... -> inner interceptor before -> tool execution -> inner interceptor after -> ... -> first interceptor after.

MCP interceptor request modification

Use request.override() to create a modified request in an interceptor. This follows an immutable pattern, leaving the original request unchanged. You can override args (tool arguments) or headers.

MCP interceptor error handling and retry logic

Interceptors can catch exceptions from tool execution (transport, session, or runtime failures) and add retry logic. Tool execution errors (CallToolResult(isError=True)) do not raise by default; to catch those as exceptions, set handle_tool_errors=False. Use try/except with exponential backoff for retry patterns.

Example: Injecting user context into MCP tool calls

```python from dataclasses import dataclass from langchain_mcp_adapters.client import MultiServerMCPClient from langchain_mcp_adapters.interceptors import MCPToolCallRequest from langchain.agents import create_agent @dataclass class Context: user_id: str api_key: str async def inject_user_context(request: MCPToolCallRequest, handler): """Inject user credentials into MCP tool calls.""" runtime = request.runtime user_id = runtime.context.user_id api_key = runtime.context.api_key modified_request = request.override( args={**request.args, "user_id": user_id} ) return await handler(modified_request) client = MultiServerMCPClient({...}, tool_interceptors=[inject_user_context]) tools = await client.get_tools() agent = create_agent("gpt-5.5", tools, context_schema=Context) result = await agent.ainvoke( {"messages": [{"role": "user", "content": "Search my orders"}]}, context={"user_id": "user_123", "api_key": "sk-..."} ) ``` This example shows how to use interceptors to inject user-specific context from the runtime into MCP tool calls.

Example: Retry interceptor with exponential backoff

```python import asyncio from langchain_mcp_adapters.client import MultiServerMCPClient from langchain_mcp_adapters.interceptors import MCPToolCallRequest async def retry_interceptor(request: MCPToolCallRequest, handler, max_retries: int = 3, delay: float = 1.0): """Retry failed tool calls with exponential backoff.""" last_error = None for attempt in range(max_retries): try: return await handler(request) except Exception as e: last_error = e if attempt < max_retries - 1: wait_time = delay * (2 ** attempt) print(f"Tool {request.name} failed (attempt {attempt + 1}), retrying in {wait_time}s...") await asyncio.sleep(wait_time) raise last_error client = MultiServerMCPClient({...}, tool_interceptors=[retry_interceptor]) ``` This example shows a retry interceptor that implements exponential backoff for failed tool calls.

Middleware stream transformers registration

Middleware can register stream transformer factories that project events from the live agent stream onto typed extension channels. At compile time, middleware-registered factories merge with anything the caller passes directly. Final ordering rules keep the built-in ToolCallTransformer in front and let caller-supplied entries land last. This is useful for surfacing counters, side-channel artifacts, partial outputs, or wire-level redaction without coupling to the framework's built-in projections.

Node-style middleware hooks in Python

Node-style hooks run sequentially at specific execution points in agent execution. Python provides four node-style hooks: before_agent (runs once before agent starts), before_model (runs before each model call), after_model (runs after each model response), and after_agent (runs once after agent completes). These hooks are useful for logging, validation, and state updates.

Node-style middleware hooks in JavaScript

JavaScript node-style hooks run sequentially at specific execution points. The four node-style hooks are: beforeAgent (runs once before agent starts), beforeModel (runs before each model call), afterModel (runs after each model response), and afterAgent (runs once after agent completes).

Wrap-style middleware hooks in Python

Wrap-style hooks run around each call, giving control over execution. Python provides two wrap-style hooks: wrap_model_call (wraps each model call) and wrap_tool_call (wraps each tool call). These are useful for retries, caching, and transformation logic.

Wrap-style middleware hooks in JavaScript

Wrap-style hooks run around each call. JavaScript provides two wrap-style hooks: wrapModelCall (wraps each model call) and wrapToolCall (wraps each tool call). You can control whether the handler is called zero times (short-circuit), once (normal flow), or multiple times (retry logic).

Node-style hooks return state updates directly

Node-style hooks return a dict directly to merge updates into agent state. The dict keys map to state fields. The dict is applied to the agent state using the graph's reducers. Returning None or an empty dict applies no changes.

Wrap-style hooks in Python use ExtendedModelResponse for state updates

Wrap-style hooks in Python use ExtendedModelResponse with a Command to inject state updates. For model calls, return ExtendedModelResponse with a Command containing update dict. For tool calls, return a Command directly. Use this when tracking state based on logic during the model or tool call, such as summarization triggers, usage metadata, or custom fields from the request or response.

Wrap-style hooks in JavaScript use Command for state updates

Wrap-style hooks in JavaScript return a Command directly to inject state updates from the model or tool call layer. The Command flows through the graph's reducers, so updates are applied correctly and messages are additive rather than replacing existing state.

Multiple wrap-style hooks composition behavior in Python

When multiple middleware layers return ExtendedModelResponse, their commands compose as follows: Commands are applied through reducers, so each Command becomes a separate state update (messages are additive). For non-reducer state fields, commands are applied inner-first then outer, so the outermost middleware's value takes precedence on conflicting keys. Retry logic is retry-safe: if outer middleware implements retry logic resulting in multiple calls to handler(), commands from earlier calls are discarded.

Multiple wrap-style hooks composition behavior in JavaScript

When multiple middleware layers return responses in JavaScript, the framework passes on the last AIMessages produced. Each middleware's handler() receives the AIMessage from the previous layer. When middleware returns an AIMessage, that becomes input to the next middleware's handler. If middleware returns a Command whose state update does not touch messages, it is treated as a no-op for message flow and the next middleware's handler receives the AIMessage from the middleware before the one that returned the Command. Commands still apply through reducers (messages additive, outer wins on conflicts). Retry logic discards commands from earlier calls.

Creating middleware in Python: decorator-based approach

Decorator-based middleware in Python uses decorators on individual functions for single-hook middleware. Available decorators: @before_agent (runs before agent starts once per invocation), @before_model (runs before each model call), @after_model (runs after each model response), @after_agent (runs after agent completes once per invocation), @wrap_model_call (wraps each model call), @wrap_tool_call (wraps each tool call), and @dynamic_prompt (generates dynamic system prompts). This approach is quick and simple, best for single hook needed with no complex configuration.

Creating middleware in Python: class-based approach

Class-based middleware in Python subclasses AgentMiddleware. Can declare three class attributes: state_schema (extend agent state with custom fields), tools (register additional tools that ship with middleware), transformers (register scope-aware stream transformer factories). Class-based approach is more powerful for complex middleware with multiple hooks or configuration. Use when defining both sync and async implementations for the same hook, combining multiple hooks in single middleware, or complex configuration required. An AgentMiddleware subclass can define sync methods (before_model, after_model, before_agent, after_agent, wrap_model_call, wrap_tool_call) and async counterparts (abefore_model, aafter_model, abefore_agent, aafter_agent).

Creating middleware in JavaScript

Use the createMiddleware function to define custom middleware in JavaScript. createMiddleware accepts three configuration fields picked up at compile time: stateSchema (extend agent state with custom fields), tools (register additional tools that ship with middleware), streamTransformers (register scope-aware stream transformer factories). The name field is required to identify the middleware.

Custom middleware state schema in Python

Middleware can extend agent state with custom properties by defining a state schema. Create a class extending AgentState with NotRequired fields for optional state properties. Pass this to decorators using state_schema parameter or set state_schema class attribute on AgentMiddleware subclass. This enables tracking state across execution, sharing data between hooks, implementing cross-cutting concerns like rate limiting or audit logging, and making conditional decisions using accumulated state.

Custom middleware state schema in JavaScript

Middleware can extend agent state with custom properties by passing a stateSchema to createMiddleware using Zod schema definition. State fields can be public or private (start with underscore). Private fields (with leading underscore) are excluded from agent result and not returned to caller. Public fields are included in invoke result. This is useful for storing internal middleware state that shouldn't be exposed to caller, such as temporary tracking variables or internal flags.

Custom context schema in JavaScript middleware

Middleware can define a custom context schema to access per-invocation metadata. Define a context schema using Zod and access it via runtime.context in middleware hooks. Unlike state, context is read-only and not persisted between invocations. Required fields in the context schema are enforced at TypeScript level, ensuring you must provide them when calling agent.invoke(). Context is ideal for user information, configuration overrides, tenant/workspace context, and request metadata.

Middleware execution order for before hooks

When using multiple middleware, before_* hooks (before_agent, before_model) run in order from first to last middleware in the list.

Middleware execution order for after hooks

When using multiple middleware, after_* hooks (after_model, after_agent) run in reverse order from last to first middleware in the list.

Middleware execution order for wrap hooks

When using multiple middleware, wrap_* hooks (wrap_model_call, wrap_tool_call) nest like function calls. The first middleware wraps all others, creating nested execution where middleware1 wraps middleware2 which wraps middleware3, which then calls the actual model or tool.

Agent jumps with middleware

To exit early from middleware, return a dictionary with 'jump_to' key. Available jump targets are 'end' (jump to end of agent execution or first after_agent hook), 'tools' (jump to tools node), and 'model' (jump to model node or first before_model hook). When using decorators, use @hook_config(can_jump_to=[...]) to declare which targets a hook can jump to.

ModelRequest.system_message is always SystemMessage object

In middleware, ModelRequest.system_message is always a SystemMessage object, even if the agent was created with system_prompt as a string. Use SystemMessage.content_blocks to access content as a list of blocks regardless of whether original content was string or list. When modifying system messages, use content_blocks and append new blocks to preserve existing structure.

Dynamic prompt middleware example in Python decorator style

Example shows wrap_model_call decorator modifying system prompt by appending user context to content_blocks. The pattern is: get current system_message.content_blocks, append new content blocks with context information, create new SystemMessage with modified content, and call handler with request.override(system_message=new_system_message).

Give your agent this brain