deep-agents-orchestration

• SubAgentMiddleware: Delegate work via task tool to specialized agents

• TodoListMiddleware: Plan and track tasks via write_todos tool

• HumanInTheLoopMiddleware: Require approval before sensitive operations

All three are automatically included in create_deep_agent().

Subagents (Task Delegation)

Use Subagents When

Use Main Agent When

Task needs specialized tools

General-purpose tools sufficient

Want to isolate complex work

Single-step operation

Need clean context for main agent

Context bloat acceptable

Default subagent: "general-purpose" - automatically available with same tools/config as main agent.

from deepagents import create_deep_agent

from langchain.tools import tool

@tool

def search_papers(query: str) -> str:

    """Search academic papers."""

    return f"Found 10 papers about {query}"

agent = create_deep_agent(

    subagents=[

        {

            "name": "researcher",

            "description": "Conduct web research and compile findings",

            "system_prompt": "Search thoroughly, return concise summary",

            "tools": [search_papers],

        }

    ]

)

# Main agent delegates: task(agent="researcher", instruction="Research AI trends")

import { createDeepAgent } from "deepagents";

import { tool } from "@langchain/core/tools";

import { z } from "zod";

const searchPapers = tool(

  async ({ query }) => `Found 10 papers about ${query}`,

  { name: "search_papers", description: "Search papers", schema: z.object({ query: z.string() }) }

);

const agent = await createDeepAgent({

  subagents: [

    {

      name: "researcher",

      description: "Conduct web research and compile findings",

      systemPrompt: "Search thoroughly, return concise summary",

      tools: [searchPapers],

    }

  ]

});

// Main agent delegates: task(agent="researcher", instruction="Research AI trends")

from deepagents import create_deep_agent

from langgraph.checkpoint.memory import MemorySaver

agent = create_deep_agent(

    subagents=[

        {

            "name": "code-deployer",

            "description": "Deploy code to production",

            "system_prompt": "You deploy code after tests pass.",

            "tools": [run_tests, deploy_to_prod],

            "interrupt_on": {"deploy_to_prod": True},  # Require approval

        }

    ],

    checkpointer=MemorySaver()  # Required for interrupts

)

# WRONG: Subagents don't remember previous calls

# task(agent='research', instruction='Find data')

# task(agent='research', instruction='What did you find?')  # Starts fresh!

# CORRECT: Complete instructions upfront

# task(agent='research', instruction='Find data on AI, save to /research/, return summary')

// WRONG: Subagents don't remember previous calls

// task research: Find data

// task research: What did you find?  // Starts fresh!

// CORRECT: Complete instructions upfront

// task research: Find data on AI, save to /research/, return summary

# WRONG: Custom subagent won't have main agent's skills

agent = create_deep_agent(

    skills=["/main-skills/"],

    subagents=[{"name": "helper", ...}]  # No skills inherited

)

# CORRECT: Provide skills explicitly (general-purpose subagent DOES inherit)

agent = create_deep_agent(

    skills=["/main-skills/"],

    subagents=[{"name": "helper", "skills": ["/helper-skills/"], ...}]

)

TodoList (Task Planning)

Use TodoList When

Skip TodoList When

Complex multi-step tasks

Simple single-action tasks

Long-running operations

Quick operations (< 3 steps)

write_todos(todos: list[dict]) -> None

Each todo item has:

• content: Description of the task

• status: One of "pending", "in_progress", "completed"

from deepagents import create_deep_agent

agent = create_deep_agent()  # TodoListMiddleware included by default

result = agent.invoke({

    "messages": [{"role": "user", "content": "Create a REST API: design models, implement CRUD, add auth, write tests"}]

}, config={"configurable": {"thread_id": "session-1"}})

# Agent's planning via write_todos:

# [

#   {"content": "Design data models", "status": "in_progress"},

#   {"content": "Implement CRUD endpoints", "status": "pending"},

#   {"content": "Add authentication", "status": "pending"},

#   {"content": "Write tests", "status": "pending"}

# ]

import { createDeepAgent } from "deepagents";

const agent = await createDeepAgent();  // TodoListMiddleware included

const result = await agent.invoke({

  messages: [{ role: "user", content: "Create a REST API: design models, implement CRUD, add auth, write tests" }]

}, { configurable: { thread_id: "session-1" } });

result = agent.invoke({...}, config={"configurable": {"thread_id": "session-1"}})

# Access todo list from final state

todos = result.get("todos", [])

for todo in todos:

    print(f"[{todo['status']}] {todo['content']}")

# WRONG: Fresh state each time without thread_id

agent.invoke({"messages": [...]})

# CORRECT: Use thread_id

config = {"configurable": {"thread_id": "user-session"}}

agent.invoke({"messages": [...]}, config=config)  # Todos preserved

Human-in-the-Loop (Approval Workflows)

Use HITL When

Skip HITL When

High-stakes operations (DB writes, deployments)

Read-only operations

Compliance requires human oversight

Fully automated workflows

from deepagents import create_deep_agent

from langgraph.checkpoint.memory import MemorySaver

agent = create_deep_agent(

    interrupt_on={

        "write_file": True,  # All decisions allowed

        "execute_sql": {"allowed_decisions": ["approve", "reject"]},

        "read_file": False,  # No interrupts

    },

    checkpointer=MemorySaver()  # REQUIRED for interrupts

)

import { createDeepAgent } from "deepagents";

import { MemorySaver } from "@langchain/langgraph";

const agent = await createDeepAgent({

  interruptOn: {

    write_file: true,

    execute_sql: { allowedDecisions: ["approve", "reject"] },

    read_file: false,

  },

  checkpointer: new MemorySaver()  // REQUIRED

});

from deepagents import create_deep_agent

from langgraph.checkpoint.memory import MemorySaver

from langgraph.types import Command

agent = create_deep_agent(

    interrupt_on={"write_file": True},

    checkpointer=MemorySaver()

)

config = {"configurable": {"thread_id": "session-1"}}

# Step 1: Agent proposes write_file - execution pauses

result = agent.invoke({

    "messages": [{"role": "user", "content": "Write config to /prod.yaml"}]

}, config=config)

# Step 2: Check for interrupts

state = agent.get_state(config)

if state.next:

    print(f"Pending action")

# Step 3: Approve and resume

result = agent.invoke(Command(resume={"decisions": [{"type": "approve"}]}), config=config)

import { createDeepAgent } from "deepagents";

import { MemorySaver, Command } from "@langchain/langgraph";

const agent = await createDeepAgent({

  interruptOn: { write_file: true },

  checkpointer: new MemorySaver()

});

const config = { configurable: { thread_id: "session-1" } };

// Step 1: Agent proposes write_file - execution pauses

let result = await agent.invoke({

  messages: [{ role: "user", content: "Write config to /prod.yaml" }]

}, config);

// Step 2: Check for interrupts

const state = await agent.getState(config);

if (state.next) {

  console.log("Pending action");

}

// Step 3: Approve and resume

result = await agent.invoke(

  new Command({ resume: { decisions: [{ type: "approve" }] } }), config

);

result = agent.invoke(

    Command(resume={"decisions": [{"type": "reject", "message": "Run tests first"}]}),

    config=config,

)

const result = await agent.invoke(

  new Command({ resume: { decisions: [{ type: "reject", message: "Run tests first" }] } }),

  config,

);

result = agent.invoke(

    Command(resume={"decisions": [{

        "type": "edit",

        "edited_action": {

            "name": "execute_sql",

            "args": {"query": "DELETE FROM users WHERE last_login < '2020-01-01' LIMIT 100"},

        },

    }]}),

    config=config,

)

• Subagent names, tools, models, system prompts

• Which tools require approval

• Allowed decision types per tool

• TodoList content and structure

What Agents CANNOT Configure

• Tool names (task, write_todos)

• HITL protocol (approve/edit/reject structure)

• Skip checkpointer requirement for interrupts

• Make subagents stateful (they're ephemeral)

# WRONG

agent = create_deep_agent(interrupt_on={"write_file": True})

# CORRECT

agent = create_deep_agent(interrupt_on={"write_file": True}, checkpointer=MemorySaver())

// WRONG

const agent = await createDeepAgent({ interruptOn: { write_file: true } });

// CORRECT

const agent = await createDeepAgent({ interruptOn: { write_file: true }, checkpointer: new MemorySaver() });

# WRONG: Can't resume without thread_id

agent.invoke({"messages": [...]})

# CORRECT

config = {"configurable": {"thread_id": "session-1"}}

agent.invoke({...}, config=config)

# Resume with Command using same config

agent.invoke(Command(resume={"decisions": [{"type": "approve"}]}), config=config)

// WRONG: Can't resume without thread_id

await agent.invoke({ messages: [...] });

// CORRECT

const config = { configurable: { thread_id: "session-1" } };

await agent.invoke({ messages: [...] }, config);

// Resume with Command using same config

await agent.invoke(new Command({ resume: { decisions: [{ type: "approve" }] } }), config);

result = agent.invoke({...}, config=config)       # Step 1: triggers interrupt

if "__interrupt__" in result:                      # Step 2: check for interrupt

    result = agent.invoke(                         # Step 3: resume

        Command(resume={"decisions": [{"type": "approve"}]}),

        config=config,

    )