DeepCoWork #3: DeepAgents SDK Internals -- create_deep_agent, LocalShellBackend, ReAct Loop

TL;DR: agent_core.py is the sole SDK coupling point, configuring 7 tools + ReAct loop + HITL gate all in one file.

Table of contents

Architecture: Single Coupling Point

DeepCoWork isolates all Deep Agents SDK interaction into agent_core.py. The export contract has five items:

When the SDK version changes, only this file needs updating. The Deep Agents SDK API reference and LangGraph create_react_agent docs are the key references.

Dissecting create_deep_agent

The build_agent() function calls the SDK’s create_deep_agent:

from deepagents import create_deep_agent
from deepagents.backends import LocalShellBackend

def build_agent(
    workspace_dir: Path,
    checkpointer: Any,
    mode: str = "cowork",
    thread_id: str | None = None,
    with_hitl: bool = True,
    tools: list | None = None,
    system_prompt: str | None = None,
) -> Any:
    llm = build_llm()

    backend = LocalShellBackend(
        root_dir=str(workspace_dir),
        virtual_mode=False,
        timeout=60,
        max_output_bytes=50_000,
        inherit_env=True,
    )

    if system_prompt is None:
        system_prompt = build_system_prompt(mode, workspace_dir)

    interrupt_on: dict = (
        {"write_file": True, "edit_file": True, "execute": True}
        if with_hitl else {}
    )

    return create_deep_agent(
        model=llm,
        tools=tools or [],
        backend=backend,
        interrupt_on=interrupt_on,
        checkpointer=checkpointer,
        system_prompt=system_prompt,
        skills=_resolve_skills(workspace_dir) or None,
    )

Let’s examine each key parameter.

LocalShellBackend

LocalShellBackend gives the agent filesystem and shell access. Auto-provided tools:

Key configuration:

backend = LocalShellBackend(
    root_dir=str(workspace_dir),  # Sandbox boundary
    virtual_mode=False,            # Real filesystem
    timeout=60,                    # Shell command timeout (seconds)
    max_output_bytes=50_000,       # Output size limit
    inherit_env=True,              # Inherit environment variables
)

root_dir is the agent’s sandbox boundary. No file access is possible outside this directory.

interrupt_on: The HITL Gate

The interrupt_on dictionary decides which tool calls pause execution:

interrupt_on = {"write_file": True, "edit_file": True, "execute": True}

When these tools are invoked, LangGraph suspends graph execution and sends an approval request to the frontend. On approval, execution resumes with Command(resume={"decisions": [...]}):

def resume_agent_input(decisions: list[dict]) -> Any:
    return Command(resume={"decisions": decisions})

Sub-agents are created with with_hitl=False and run without HITL — the main agent already has approval.

The ReAct Loop

The core of DeepAgents SDK is a LangGraph-based ReAct (Reasoning + Acting) loop:

[User message]
     |
     v
  LLM Reasoning
     |
     +-- Tool call decision (Acting)
     |       |
     |       v
     |   [interrupt_on check]
     |       |
     |       +-- HITL needed -> pause -> wait for approval
     |       +-- HITL not needed -> execute immediately
     |       |
     |       v
     |   Tool execution result
     |       |
     +-------+
     |
     v
  LLM Reasoning (decide next action)
     |
     +-- Complete -> final response
     +-- Incomplete -> loop again

The stream_events() function converts this loop into SSE:

async for event in agent.astream(
    agent_input,
    stream_mode=["updates", "messages"],
    subgraphs=True,
    config=cfg,
):

stream_mode=["updates", "messages"] receives both tool call events and text tokens. subgraphs=True captures sub-agent events in ACP mode.

Checkpointer: State Persistence

AsyncSqliteSaver persists agent state to SQLite:

from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver

self.db_conn = await aiosqlite.connect(str(config.DB_PATH))
await self.db_conn.execute("PRAGMA journal_mode=WAL")
self.checkpointer = AsyncSqliteSaver(self.db_conn)

WAL (Write-Ahead Logging) mode ensures concurrent read/write performance. Conversation history and agent state survive app restarts.

Skill Resolution

_resolve_skills() scans global and workspace skill directories:

def _resolve_skills(workspace_dir: Path) -> list[str]:
    sources: list[str] = []
    global_skills = config.WORKSPACE_ROOT / "skills"
    if global_skills.is_dir():
        sources.append("skills/")
    ws_skills = workspace_dir / "skills"
    if ws_skills.is_dir():
        sources.append("skills/")
    return sources

Priority: global (~/.cowork/skills/) < workspace ({workspace}/skills/). Later-loaded skills take precedence.

Benchmark

FAQ

Can it work without the Deep Agents SDK?

No. create_deep_agent and LocalShellBackend are core dependencies. However, by replacing only agent_core.py, you could swap in a different agent framework.

What happens with virtual_mode=True?

File writes are handled in memory without touching disk. Useful for testing or preview, but DeepCoWork uses False since real filesystem changes are the goal.

Why max_output_bytes=50_000?

Large shell outputs consume LLM context. The 50KB limit prevents commands like npm install from overwhelming the agent with verbose output.

Series

1. DeepCoWork: I Built an AI Agent Desktop App
2. Tauri 2 + Python Sidecar
3. [This post] DeepAgents SDK Internals
4. System Prompt Design per Mode
5. SSE Streaming Pipeline
6. HITL Approval Flow
7. Multi-Agent ACP Mode
8. Agent Memory 4 Layers
9. Skills System
10. LLM Provider Integration
11. Security Checklist
12. GitHub Actions Cross-Platform Build