Deep Analysts

Middleware, pluggable file backends, and streaming helpers for building LangChain v1 / LangGraph agents that have memory, skills, a filesystem, a sandbox, and subagents.

deepanalysts is a library only — no CLI, no server, no domain logic. It depends on langchain, langgraph, httpx, pyyaml, tenacity, and wcmatch; model providers and storage services are the caller's choice.

Install

uv add deepanalysts            # or: pip install deepanalysts
uv add "deepanalysts[e2b]"     # + E2B microVM sandbox backend
uv add "deepanalysts[postgres]" # + LangGraph Postgres checkpointer deps

Requires Python 3.11+, langchain >= 1.3.13, and langgraph >= 1.2.9.

What's in the box

Middleware (deepanalysts.middleware)

• ToolErrorHandlingMiddleware — turns tool exceptions into ToolMessages and trips a circuit breaker after N consecutive failures of the same tool.
• SummarizationMiddleware — compacts what the model sees while leaving the checkpointed message log intact, so history, replay, and evals keep the source messages.
• MemoryMiddleware — always-on context from AGENTS.md-style sources.
• SkillsMiddleware — on-demand workflows via progressive disclosure of SKILL.md frontmatter.
• FilesystemMiddleware — ls / read_file / write_file / edit_file / glob / grep, plus execute when the backend is a sandbox. Oversized tool results are spilled to a file and replaced with a preview.
• SubAgentMiddleware — a task tool that delegates to named subagents.
• PatchToolCallsMiddleware — repairs dangling tool calls before the next model turn.

Backends (deepanalysts.backends) — all implement BackendProtocol (and SandboxBackendProtocol where they can execute):

StateBackend (agent state, ephemeral) · StoreBackend (LangGraph BaseStore, persistent) · LocalFilesystemBackend (real disk, jailed) · RestrictedSubprocessBackend (hardened local subprocess) · E2BSandboxBackend (Firecracker microVM) · SupabaseStorageBackend (async object storage) · CompositeBackend (route by path prefix).

Also: BasementClient + BasementMemoryLoader / BasementSkillsLoader for API-sourced skills and memories, DeepAnalystsState for delta-channel message checkpointing, and SubagentTransformer for typed subagent handles on LangGraph v3 streams.

Minimal usage

from langchain.agents import create_agent
from langchain.chat_models import init_chat_model

from deepanalysts import DeepAnalystsState
from deepanalysts.backends import (
    CompositeBackend,
    RestrictedSubprocessBackend,
    StoreBackend,
)
from deepanalysts.middleware import (
    FilesystemMiddleware,
    MemoryMiddleware,
    PatchToolCallsMiddleware,
    SkillsMiddleware,
    ToolErrorHandlingMiddleware,
    create_summarization_middleware,
)

model = init_chat_model("<provider>:<model-id>")  # any LangChain chat model

# Sandbox for scratch work; persistent store for skills and memories.
def backend(runtime):
    return CompositeBackend(
        default=RestrictedSubprocessBackend(timeout=30),
        routes={
            "/skills/": StoreBackend(runtime),
            "/memories/": StoreBackend(runtime),
        },
    )

agent = create_agent(
    model,
    tools=[],
    state_schema=DeepAnalystsState,  # optional: delta-channel message checkpointing
    middleware=[
        ToolErrorHandlingMiddleware(),                      # first: catches every tool error
        create_summarization_middleware(model, backend),    # model-aware compaction defaults
        MemoryMiddleware(backend=backend, sources=["/memories/AGENTS.md"]),
        SkillsMiddleware(backend=backend, sources=["/skills/user/", "/skills/project/"]),
        FilesystemMiddleware(backend=backend),              # file tools + `execute`
        PatchToolCallsMiddleware(),                         # last: repair dangling tool calls
    ],
)

result = agent.invoke(
    {"messages": [{"role": "user", "content": "Write a note to /notes.md, then read it back."}]},
    config={"configurable": {"user_id": "user-123"}},
)
print(result["messages"][-1].content)

Notes on the example:

• Every middleware's backend= accepts either a backend instance or a factory (ToolRuntime) -> BackendProtocol. Use a factory for anything runtime-scoped, as StoreBackend is.
• Order matters: error handling first, summarization before the prompt-injecting middleware, filesystem before subagents, patching last.
• configurable.user_id is what StoreBackend uses to namespace files per tenant.
• create_summarization_middleware(model, backend) derives trigger/retention from the model profile; construct SummarizationMiddleware(...) directly if you want explicit trigger=("tokens", 100_000) / keep=("messages", 20).

Subagents

from deepanalysts.middleware import (
    SubAgent,
    SubAgentMiddleware,
    private_state_field_names,
)

technical_analyst: SubAgent = {
    "name": "technical_analyst",
    "description": "Analyzes charts and technical indicators.",
    "system_prompt": "You are a technical analyst...",
    "tools": [get_indicators],
}

sub = SubAgentMiddleware(default_model=model, default_tools=[], subagents=[technical_analyst])
middleware = [ToolErrorHandlingMiddleware(), FilesystemMiddleware(backend=backend), sub, PatchToolCallsMiddleware()]

# Keep middleware-private state out of spawned subagents (assign AFTER the stack exists —
# the setter rebuilds the `task` tool).
sub.private_state_keys = private_state_field_names(
    *(m.state_schema for m in middleware if getattr(m, "state_schema", None) is not None)
)

agent = create_agent(model, tools=[], middleware=middleware)

The middleware exposes one task(description, subagent_type) tool. It injects a session-context header (symbol / exchange / interval from config.configurable, plus the current UTC time) into the subagent's prompt, retries transient failures, and returns the last non-empty assistant message to the caller.

API-backed skills and memories

from deepanalysts.backends import BasementMemoryLoader, BasementSkillsLoader
from deepanalysts.middleware import MemoryMiddleware, SkillsMiddleware

memory = MemoryMiddleware(loader=BasementMemoryLoader(token_provider=get_jwt))
skills = SkillsMiddleware(
    loader=BasementSkillsLoader(token_provider=get_jwt, built_in_dirs=["./skills"]),
    agent_name="technical_analyst",  # filters by the skill's target_agents
)

Loader mode takes precedence over backend mode when both are configured. Point the client at another host with BasementClient(base_url=...) or the BASEMENT_API env var.

Sandboxed execution

FilesystemMiddleware adds an execute tool only when the backend can run commands, and hides it otherwise.

from deepanalysts.backends import E2BSandboxBackend, RestrictedSubprocessBackend

local = RestrictedSubprocessBackend(timeout=30)     # same host: process-group kill, rlimits, path jail
micro = E2BSandboxBackend(template="my-template")   # real isolation; egress denied by default
try:
    print(micro.execute("python3 -c 'print(2**10)'").output)
finally:
    micro.close()

RestrictedSubprocessBackend is hardened but not container isolation — it runs as the same OS user with open network egress. Use a microVM backend for untrusted code.

Development

uv sync --all-extras
uv run pytest
uv run ruff check . && uv run ruff format .
uv build

Async tests use anyio (@pytest.mark.anyio) with the asyncio backend. Tests are offline by default; the E2B integration tests skip unless E2B_API_KEY is set.

Releasing

Bump version in pyproject.toml, merge to main, and publish a GitHub release — CI runs the tests, builds, and pushes to PyPI. A workflow_dispatch run with test_pypi=true publishes to TestPyPI instead.

License

MIT