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: str
"""Unique identifier for the subagent."""

description: str
"""What this subagent does. The main agent uses this to decide when to delegate."""

system_prompt: str
"""Instructions for the subagent."""

tools: Sequence[BaseTool | Callable | dict[str, Any]]
"""Tools the subagent can use."""

model: NotRequired[str | BaseChatModel]
"""Override the main agent's model. Use `'provider:model-name'` format."""

middleware: NotRequired[list[AgentMiddleware]]
"""Additional middleware for custom behavior."""

interrupt_on: NotRequired[dict[str, bool | InterruptOnConfig]]
"""Configure human-in-the-loop for specific tools."""

!!! 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: str
"""Unique identifier for the subagent."""

description: str
"""What this subagent does."""

runnable: Runnable
"""A custom agent implementation.

Create a custom agent using either:

1. LangChain's [`create_agent()`](https://docs.langchain.com/oss/python/langchain/quickstart)
2. A custom graph using [`langgraph`](https://docs.langchain.com/oss/python/langgraph/quickstart)

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.
"""

Args:
    default_model: Default model for subagents that don't specify one.
    default_tools: Default tools for subagents that don't specify tools.
    default_middleware: Middleware to apply to all subagents. If `None`,
        no default middleware is applied.
    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.
    subagents: List of agent specifications or pre-compiled agents.
    general_purpose_agent: Whether to include a general-purpose subagent.

Returns:
    Tuple of (agent_dict, description_list) where agent_dict maps agent names
    to runnable instances and description_list contains formatted descriptions.
"""
# Use empty list if None (no default middleware)
default_subagent_middleware = default_middleware or []

agents: dict[str, Any] = {}
subagent_descriptions = []

# Create general-purpose agent if enabled
if general_purpose_agent:
    general_purpose_middleware = [*default_subagent_middleware]
    if default_interrupt_on:
        general_purpose_middleware.append(HumanInTheLoopMiddleware(interrupt_on=default_interrupt_on))
    general_purpose_subagent = create_agent(
        default_model,
        system_prompt=DEFAULT_SUBAGENT_PROMPT,
        tools=default_tools,
        middleware=general_purpose_middleware,
        name="general-purpose",
    )
    agents["general-purpose"] = general_purpose_subagent
    subagent_descriptions.append(f"- general-purpose: {DEFAULT_GENERAL_PURPOSE_DESCRIPTION}")

# Process custom subagents
for agent_ in subagents:
    subagent_descriptions.append(f"- {agent_['name']}: {agent_['description']}")
    if "runnable" in agent_:
        custom_agent = cast("CompiledSubAgent", agent_)
        agents[custom_agent["name"]] = custom_agent["runnable"]
        continue
    _tools = agent_.get("tools", list(default_tools))

    subagent_model = agent_.get("model", default_model)

    _middleware = [*default_subagent_middleware, *agent_["middleware"]] if "middleware" in agent_ else [*default_subagent_middleware]

    interrupt_on = agent_.get("interrupt_on", default_interrupt_on)
    if interrupt_on:
        _middleware.append(HumanInTheLoopMiddleware(interrupt_on=interrupt_on))

    agents[agent_["name"]] = create_agent(
        subagent_model,
        system_prompt=agent_["system_prompt"],
        tools=_tools,
        middleware=_middleware,
        name=agent_["name"],
    )
return agents, subagent_descriptions

Args:
    default_model: Default model for subagents.
    default_tools: Default tools for subagents.
    default_middleware: Middleware to apply to all subagents.
    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.
    subagents: List of subagent specifications.
    general_purpose_agent: Whether to include general-purpose agent.
    task_description: Custom description for the task tool. If `None`,
        uses default template. Supports `{available_agents}` placeholder.

Returns:
    A StructuredTool that can invoke subagents by type.
"""
subagent_graphs, subagent_descriptions = _get_subagents(
    default_model=default_model,
    default_tools=default_tools,
    default_middleware=default_middleware,
    default_interrupt_on=default_interrupt_on,
    subagents=subagents,
    general_purpose_agent=general_purpose_agent,
)
subagent_description_str = "\n".join(subagent_descriptions)

def _return_command_with_state_update(result: dict, tool_call_id: str) -> Command:
    # Validate that the result contains a 'messages' key
    if "messages" not in result:
        error_msg = (
            "CompiledSubAgent must return a state containing a 'messages' key. "
            "Custom StateGraphs used with CompiledSubAgent should include 'messages' "
            "in their state schema to communicate results back to the main agent."
        )
        raise ValueError(error_msg)

    state_update = {k: v for k, v in result.items() if k not in _EXCLUDED_STATE_KEYS}
    # Strip trailing whitespace to prevent API errors with Anthropic
    message_text = result["messages"][-1].text.rstrip() if result["messages"][-1].text else ""
    return Command(
        update={
            **state_update,
            "messages": [ToolMessage(message_text, tool_call_id=tool_call_id)],
        }
    )

def _validate_and_prepare_state(subagent_type: str, description: str, runtime: ToolRuntime) -> tuple[Runnable, dict]:
    """Prepare state for invocation."""
    subagent = subagent_graphs[subagent_type]
    # Create a new state dict to avoid mutating the original
    subagent_state = {k: v for k, v in runtime.state.items() if k not in _EXCLUDED_STATE_KEYS}
    subagent_state["messages"] = [HumanMessage(content=description)]
    return subagent, subagent_state

# Use custom description if provided, otherwise use default template
if task_description is None:
    task_description = TASK_TOOL_DESCRIPTION.format(available_agents=subagent_description_str)
elif "{available_agents}" in task_description:
    # If custom description has placeholder, format with agent descriptions
    task_description = task_description.format(available_agents=subagent_description_str)

def task(
    description: str,
    subagent_type: str,
    runtime: ToolRuntime,
) -> str | Command:
    if subagent_type not in subagent_graphs:
        allowed_types = ", ".join([f"`{k}`" for k in subagent_graphs])
        return f"We cannot invoke subagent {subagent_type} because it does not exist, the only allowed types are {allowed_types}"
    subagent, subagent_state = _validate_and_prepare_state(subagent_type, description, runtime)
    result = subagent.invoke(subagent_state)
    if not runtime.tool_call_id:
        value_error_msg = "Tool call ID is required for subagent invocation"
        raise ValueError(value_error_msg)
    return _return_command_with_state_update(result, runtime.tool_call_id)

async def atask(
    description: str,
    subagent_type: str,
    runtime: ToolRuntime,
) -> str | Command:
    if subagent_type not in subagent_graphs:
        allowed_types = ", ".join([f"`{k}`" for k in subagent_graphs])
        return f"We cannot invoke subagent {subagent_type} because it does not exist, the only allowed types are {allowed_types}"
    subagent, subagent_state = _validate_and_prepare_state(subagent_type, description, runtime)
    result = await subagent.ainvoke(subagent_state)
    if not runtime.tool_call_id:
        value_error_msg = "Tool call ID is required for subagent invocation"
        raise ValueError(value_error_msg)
    return _return_command_with_state_update(result, runtime.tool_call_id)

return StructuredTool.from_function(
    name="task",
    func=task,
    coroutine=atask,
    description=task_description,
)

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.

Args:
    default_model: The model to use for subagents.

        Can be a `LanguageModelLike` or a dict for `init_chat_model`.
    default_tools: The tools to use for the default general-purpose subagent.
    default_middleware: Default middleware to apply to all subagents.

        If `None`, no default middleware is applied.

        Pass a list to specify custom middleware.
    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.
    subagents: A list of additional subagents to provide to the agent.
    system_prompt: Full system prompt override. When provided, completely replaces
        the agent's system prompt.
    general_purpose_agent: Whether to include the general-purpose agent.
    task_description: Custom description for the task tool.

        If `None`, uses the default description template.

Example:
    ```python
    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=[],
            )
        ],
    )
    ```
"""

def __init__(
    self,
    *,
    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`."""
    super().__init__()
    self.system_prompt = system_prompt
    task_tool = _create_task_tool(
        default_model=default_model,
        default_tools=default_tools or [],
        default_middleware=default_middleware,
        default_interrupt_on=default_interrupt_on,
        subagents=subagents or [],
        general_purpose_agent=general_purpose_agent,
        task_description=task_description,
    )
    self.tools = [task_tool]

def wrap_model_call(
    self,
    request: ModelRequest,
    handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
    """Update the system message to include instructions on using subagents."""
    if self.system_prompt is not None:
        new_system_message = append_to_system_message(request.system_message, self.system_prompt)
        return handler(request.override(system_message=new_system_message))
    return handler(request)

async def awrap_model_call(
    self,
    request: ModelRequest,
    handler: Callable[[ModelRequest], Awaitable[ModelResponse]],
) -> ModelResponse:
    """(async) Update the system message to include instructions on using subagents."""
    if self.system_prompt is not None:
        new_system_message = append_to_system_message(request.system_message, self.system_prompt)
        return await handler(request.override(system_message=new_system_message))
    return await handler(request)