bitwikiorg """Middleware for providing subagents to an agent via a task tool."""
from collections.abc import Awaitable, Callable, Sequence from typing import Any, NotRequired, TypedDict, cast
from langchain.agents import create_agent from langchain.agents.middleware import HumanInTheLoopMiddleware, InterruptOnConfig from langchain.agents.middleware.types import AgentMiddleware, ModelRequest, ModelResponse from langchain.tools import BaseTool, ToolRuntime from langchain_core.language_models import BaseChatModel from langchain_core.messages import HumanMessage, ToolMessage from langchain_core.runnables import Runnable from langchain_core.tools import StructuredTool from langgraph.types import Command
from deepagents.middleware._utils import append_to_system_message
class SubAgent(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: 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."""
class CompiledSubAgent(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: 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.
"""
DEFAULT_SUBAGENT_PROMPT = "In order to complete the objective that the user asks of you, you have access to a number of standard tools."
State keys that are excluded when passing state to subagents and when returning
updates from subagents.
When returning updates:
1. The messages key is handled explicitly to ensure only the final message is included
2. The todos and structured_response keys are excluded as they do not have a defined reducer
and no clear meaning for returning them from a subagent to the main agent.
_EXCLUDED_STATE_KEYS = {"messages", "todos", "structured_response"}
TASK_TOOL_DESCRIPTION = """Launch an ephemeral subagent to handle complex, multi-step independent tasks with isolated context windows.
Available agent types and the tools they have access to: {available_agents}
When using the Task tool, you must specify a subagent_type parameter to select which agent type to use.
Usage notes:
- Launch multiple agents concurrently whenever possible, to maximize performance; to do that, use a single message with multiple tool uses
- When the agent is done, it will return a single message back to you. The result returned by the agent is not visible to the user. To show the user the result, you should send a text message back to the user with a concise summary of the result.
- Each agent invocation is stateless. You will not be able to send additional messages to the agent, nor will the agent be able to communicate with you outside of its final report. Therefore, your prompt should contain a highly detailed task description for the agent to perform autonomously and you should specify exactly what information the agent should return back to you in its final and only message to you.
- The agent's outputs should generally be trusted
- Clearly tell the agent whether you expect it to create content, perform analysis, or just do research (search, file reads, web fetches, etc.), since it is not aware of the user's intent
- If the agent description mentions that it should be used proactively, then you should try your best to use it without the user having to ask for it first. Use your judgement.
- When only the general-purpose agent is provided, you should use it for all tasks. It is great for isolating context and token usage, and completing specific, complex tasks, as it has all the same capabilities as the main agent.
Example usage of the general-purpose agent:
Example usage with custom agents:
TASK_SYSTEM_PROMPT = """## task (subagent spawner)
You have access to a task tool to launch short-lived subagents that handle isolated tasks. These agents are ephemeral — they live only for the duration of the task and return a single result.
When to use the task tool:
- When a task is complex and multi-step, and can be fully delegated in isolation
- When a task is independent of other tasks and can run in parallel
- When a task requires focused reasoning or heavy token/context usage that would bloat the orchestrator thread
- When sandboxing improves reliability (e.g. code execution, structured searches, data formatting)
- When you only care about the output of the subagent, and not the intermediate steps (ex. performing a lot of research and then returned a synthesized report, performing a series of computations or lookups to achieve a concise, relevant answer.)
Subagent lifecycle:
- Spawn → Provide clear role, instructions, and expected output
- Run → The subagent completes the task autonomously
- Return → The subagent provides a single structured result
- Reconcile → Incorporate or synthesize the result into the main thread
When NOT to use the task tool:
- If you need to see the intermediate reasoning or steps after the subagent has completed (the task tool hides them)
- If the task is trivial (a few tool calls or simple lookup)
- If delegating does not reduce token usage, complexity, or context switching
- If splitting would add latency without benefit
Important Task Tool Usage Notes to Remember
- Whenever possible, parallelize the work that you do. This is true for both tool_calls, and for tasks. Whenever you have independent steps to complete - make tool_calls, or kick off tasks (subagents) in parallel to accomplish them faster. This saves time for the user, which is incredibly important.
- Remember to use the
tasktool to silo independent tasks within a multi-part objective. - You should use the
tasktool whenever you have a complex task that will take multiple steps, and is independent from other tasks that the agent needs to complete. These agents are highly competent and efficient.""" # noqa: E501
DEFAULT_GENERAL_PURPOSE_DESCRIPTION = "General-purpose agent for researching complex questions, searching for files and content, and executing multi-step tasks. When you are searching for a keyword or file and are not confident that you will find the right match in the first few tries use this agent to perform the search for you. This agent has access to all tools as the main agent." # noqa: E501
def _get_subagents( *, default_model: str | BaseChatModel, default_tools: Sequence[BaseTool | Callable | dict[str, Any]], default_middleware: list[AgentMiddleware] | None, default_interrupt_on: dict[str, bool | InterruptOnConfig] | None, subagents: list[SubAgent | CompiledSubAgent], general_purpose_agent: bool, ) -> tuple[dict[str, Any], list[str]]: """Create subagent instances from specifications.
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
def _create_task_tool( *, default_model: str | BaseChatModel, default_tools: Sequence[BaseTool | Callable | dict[str, Any]], default_middleware: list[AgentMiddleware] | None, default_interrupt_on: dict[str, bool | InterruptOnConfig] | None, subagents: list[SubAgent | CompiledSubAgent], general_purpose_agent: bool, task_description: str | None = None, ) -> BaseTool: """Create a task tool for invoking subagents.
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,
)
class SubAgentMiddleware(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.
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)