SubAgentMiddleware

SubAgentMiddleware

Bases: AgentMiddleware

Middleware for providing subagents to an agent via a task tool.

This middleware adds a task tool to the agent that can be used to invoke subagents. Subagents are useful for handling complex tasks that require multiple steps, or tasks that require a lot of context to resolve.

A chief benefit of subagents is that they can handle multi-step tasks, and then return a clean, concise response to the main agent.

Subagents are also great for different domains of expertise that require a narrower subset of tools and focus.

This middleware comes with a default general-purpose subagent that can be used to handle the same tasks as the main agent, but with isolated context.

PARAMETER	DESCRIPTION
default_model	The model to use for subagents. Can be a LanguageModelLike or a dict for init_chat_model. TYPE: str | BaseChatModel
default_tools	The tools to use for the default general-purpose subagent. TYPE: Sequence[BaseTool | Callable | dict[str, Any]] | None DEFAULT: None
default_middleware	Default middleware to apply to all subagents. If None, no default middleware is applied. Pass a list to specify custom middleware. TYPE: list[AgentMiddleware] | None DEFAULT: None
default_interrupt_on	The tool configs to use for the default general-purpose subagent. These are also the fallback for any subagents that don't specify their own tool configs. TYPE: dict[str, bool | InterruptOnConfig] | None DEFAULT: None
subagents	A list of additional subagents to provide to the agent. TYPE: list[SubAgent | CompiledSubAgent] | None DEFAULT: None
system_prompt	Full system prompt override. When provided, completely replaces the agent's system prompt. TYPE: str | None DEFAULT: TASK_SYSTEM_PROMPT
general_purpose_agent	Whether to include the general-purpose agent. TYPE: bool DEFAULT: True
task_description	Custom description for the task tool. If None, uses the default description template. TYPE: str | None DEFAULT: None

Example

from langchain.agents.middleware.subagents import SubAgentMiddleware
from langchain.agents import create_agent

# Basic usage with defaults (no default middleware)
agent = create_agent(
    "openai:gpt-4o",
    middleware=[
        SubAgentMiddleware(
            default_model="openai:gpt-4o",
            subagents=[],
        )
    ],
)

# Add custom middleware to subagents
agent = create_agent(
    "openai:gpt-4o",
    middleware=[
        SubAgentMiddleware(
            default_model="openai:gpt-4o",
            default_middleware=[TodoListMiddleware()],
            subagents=[],
        )
    ],
)

METHOD	DESCRIPTION
__init__	Initialize the SubAgentMiddleware.
wrap_model_call	Update the system message to include instructions on using subagents.
awrap_model_call	(async) Update the system message to include instructions on using subagents.
before_agent	Logic to run before the agent execution starts.
abefore_agent	Async logic to run before the agent execution starts.
before_model	Logic to run before the model is called.
abefore_model	Async logic to run before the model is called.
after_model	Logic to run after the model is called.
aafter_model	Async logic to run after the model is called.
after_agent	Logic to run after the agent execution completes.
aafter_agent	Async logic to run after the agent execution completes.
wrap_tool_call	Intercept tool execution for retries, monitoring, or modification.
awrap_tool_call	Intercept and control async tool execution via handler callback.

tools instance-attribute

tools = [task_tool]

Additional tools registered by the middleware.

state_schema class-attribute instance-attribute

state_schema: type[StateT] = cast('type[StateT]', _DefaultAgentState)

The schema for state passed to the middleware nodes.

name property

name: str

The name of the middleware instance.

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

__init__

__init__(
    *,
    default_model: str | BaseChatModel,
    default_tools: Sequence[BaseTool | Callable | dict[str, Any]] | None = None,
    default_middleware: list[AgentMiddleware] | None = None,
    default_interrupt_on: dict[str, bool | InterruptOnConfig] | None = None,
    subagents: list[SubAgent | CompiledSubAgent] | None = None,
    system_prompt: str | None = TASK_SYSTEM_PROMPT,
    general_purpose_agent: bool = True,
    task_description: str | None = None,
) -> None

Initialize the SubAgentMiddleware.

wrap_model_call

wrap_model_call(
    request: ModelRequest, handler: Callable[[ModelRequest], ModelResponse]
) -> ModelResponse

Update the system message to include instructions on using subagents.

awrap_model_call async

awrap_model_call(
    request: ModelRequest, handler: Callable[[ModelRequest], Awaitable[ModelResponse]]
) -> ModelResponse

(async) Update the system message to include instructions on using subagents.

before_agent

before_agent(state: StateT, runtime: Runtime[ContextT]) -> dict[str, Any] | None

Logic to run before the agent execution starts.

PARAMETER	DESCRIPTION
state	The current agent state. TYPE: StateT
runtime	The runtime context. TYPE: Runtime[ContextT]

RETURNS	DESCRIPTION
dict[str, Any] | None	Agent state updates to apply before agent execution.

abefore_agent async

abefore_agent(state: StateT, runtime: Runtime[ContextT]) -> dict[str, Any] | None

Async logic to run before the agent execution starts.

PARAMETER	DESCRIPTION
state	The current agent state. TYPE: StateT
runtime	The runtime context. TYPE: Runtime[ContextT]

RETURNS	DESCRIPTION
dict[str, Any] | None	Agent state updates to apply before agent execution.

before_model

before_model(state: StateT, runtime: Runtime[ContextT]) -> dict[str, Any] | None

Logic to run before the model is called.

PARAMETER	DESCRIPTION
state	The current agent state. TYPE: StateT
runtime	The runtime context. TYPE: Runtime[ContextT]

RETURNS	DESCRIPTION
dict[str, Any] | None	Agent state updates to apply before model call.

abefore_model async

abefore_model(state: StateT, runtime: Runtime[ContextT]) -> dict[str, Any] | None

Async logic to run before the model is called.

PARAMETER	DESCRIPTION
state	The agent state. TYPE: StateT
runtime	The runtime context. TYPE: Runtime[ContextT]

RETURNS	DESCRIPTION
dict[str, Any] | None	Agent state updates to apply before model call.

after_model

after_model(state: StateT, runtime: Runtime[ContextT]) -> dict[str, Any] | None

Logic to run after the model is called.

PARAMETER	DESCRIPTION
state	The current agent state. TYPE: StateT
runtime	The runtime context. TYPE: Runtime[ContextT]

RETURNS	DESCRIPTION
dict[str, Any] | None	Agent state updates to apply after model call.

aafter_model async

aafter_model(state: StateT, runtime: Runtime[ContextT]) -> dict[str, Any] | None

Async logic to run after the model is called.

PARAMETER	DESCRIPTION
state	The current agent state. TYPE: StateT
runtime	The runtime context. TYPE: Runtime[ContextT]

RETURNS	DESCRIPTION
dict[str, Any] | None	Agent state updates to apply after model call.

after_agent

after_agent(state: StateT, runtime: Runtime[ContextT]) -> dict[str, Any] | None

Logic to run after the agent execution completes.

PARAMETER	DESCRIPTION
state	The current agent state. TYPE: StateT
runtime	The runtime context. TYPE: Runtime[ContextT]

RETURNS	DESCRIPTION
dict[str, Any] | None	Agent state updates to apply after agent execution.

aafter_agent async

aafter_agent(state: StateT, runtime: Runtime[ContextT]) -> dict[str, Any] | None

Async logic to run after the agent execution completes.

PARAMETER	DESCRIPTION
state	The current agent state. TYPE: StateT
runtime	The runtime context. TYPE: Runtime[ContextT]

RETURNS	DESCRIPTION
dict[str, Any] | None	Agent state updates to apply after agent execution.

wrap_tool_call

wrap_tool_call(
    request: ToolCallRequest,
    handler: Callable[[ToolCallRequest], ToolMessage | Command[Any]],
) -> ToolMessage | Command[Any]

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.

PARAMETER	DESCRIPTION
request	Tool call request with call dict, BaseTool, state, and runtime. Access state via request.state and runtime via request.runtime. TYPE: ToolCallRequest
handler	Callable to execute the tool (can be called multiple times). TYPE: Callable[[ToolCallRequest], ToolMessage | Command[Any]]

RETURNS	DESCRIPTION
ToolMessage | Command[Any]	ToolMessage or Command (the final result).

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

Each call to handler is independent and stateless.

Examples:

Modify request before execution

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)

Retry on error (call handler multiple times)

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

Conditional retry based on response

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

awrap_tool_call async

awrap_tool_call(
    request: ToolCallRequest,
    handler: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command[Any]]],
) -> ToolMessage | Command[Any]

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.

PARAMETER	DESCRIPTION
request	Tool call request with call dict, BaseTool, state, and runtime. Access state via request.state and runtime via request.runtime. TYPE: ToolCallRequest
handler	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. TYPE: Callable[[ToolCallRequest], Awaitable[ToolMessage | Command[Any]]]

RETURNS	DESCRIPTION
ToolMessage | Command[Any]	ToolMessage or Command (the final result).

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

Each call to handler is independent and stateless.

Examples:

Async retry on error

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

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

SubAgent

Bases: TypedDict

Specification for an agent.

When specifying custom agents, the default_middleware from SubAgentMiddleware will be applied first, followed by any middleware specified in this spec. To use only custom middleware without the defaults, pass default_middleware=[] to SubAgentMiddleware.

Required fields

name: Unique identifier for the subagent.

The main agent uses this name when calling the `task()` tool.

description: What this subagent does.

Be specific and action-oriented. The main agent uses this to decide when to delegate.

system_prompt: Instructions for the subagent.

Include tool usage guidance and output format requirements.

tools: Tools the subagent can use.

Keep this minimal and include only what's needed.

Optional fields

model: Override the main agent's model.

Use the format `'provider:model-name'` (e.g., `'openai:gpt-4o'`).

middleware: Additional middleware for custom behavior, logging, or rate limiting. interrupt_on: Configure human-in-the-loop for specific tools.

Requires a checkpointer.

name instance-attribute

name: str

Unique identifier for the subagent.

description instance-attribute

description: str

What this subagent does. The main agent uses this to decide when to delegate.

system_prompt instance-attribute

system_prompt: str

Instructions for the subagent.

tools instance-attribute

tools: Sequence[BaseTool | Callable | dict[str, Any]]

Tools the subagent can use.

model instance-attribute

model: NotRequired[str | BaseChatModel]

Override the main agent's model. Use 'provider:model-name' format.

middleware instance-attribute

middleware: NotRequired[list[AgentMiddleware]]

Additional middleware for custom behavior.

interrupt_on instance-attribute

interrupt_on: NotRequired[dict[str, bool | InterruptOnConfig]]

Configure human-in-the-loop for specific tools.

CompiledSubAgent

Bases: TypedDict

A pre-compiled agent spec.

Note

The runnable's state schema must include a 'messages' key.

This is required for the subagent to communicate results back to the main agent.

When the subagent completes, the final message in the 'messages' list will be extracted and returned as a ToolMessage to the parent agent.

name instance-attribute

name: str

Unique identifier for the subagent.

description instance-attribute

description: str

What this subagent does.

runnable instance-attribute

runnable: Runnable

A custom agent implementation.

Create a custom agent using either:

1. LangChain's create_agent()
2. A custom graph using langgraph

If you're creating a custom graph, make sure the state schema includes a 'messages' key. This is required for the subagent to communicate results back to the main agent.