Skip to main content

Deep Agents integration

View Markdown

Temporal's integration with Deep Agents makes an existing LangChain Deep Agent durable by adding one plugin. Build your agent with create_deep_agent(...) inside a Workflow, add DeepAgentsPlugin to your Client or Worker, and each LLM call and each I/O tool call becomes a Temporal Activity — while the agent's control loop runs, and deterministically replays, inside the Workflow.

The code you already wrote against deepagents doesn't change. Sub-agents, planning and todo state, the filesystem middleware, human-in-the-loop interrupts, and agent.ainvoke(...) all keep working. On top of that you get crash durability, automatic retries and timeouts on every model and tool call, resumable human-in-the-loop, and bounded Workflow history for long-running agents.

Code snippets in this guide are taken from the Deep Agents plugin samples. Refer to the samples for the complete code.

Prerequisites​

Install the plugin​

The plugin lives on the Temporal Python SDK's main branch and ships as the temporalio[deepagents] extra in the next SDK release. Until that release is on PyPI, install it from main:

uv add "temporalio[deepagents] @ git+https://github.com/temporalio/sdk-python.git"

or with pip:

pip install "temporalio[deepagents] @ git+https://github.com/temporalio/sdk-python.git"

This builds the SDK from source, including its Rust core, so expect a few minutes on first install. Once the next release is available, a plain uv add "temporalio[deepagents]" (or pip install "temporalio[deepagents]") is all you need.

Hello World​

A Deep Agent becomes durable without changing the agent code itself. The Workflow below builds a vanilla create_deep_agent(...) and drives it with await agent.ainvoke(...) — exactly the code you would write outside Temporal. Because model= is a bare "provider:name" string, the plugin auto-routes the model call through the deepagents.invoke_model Activity, so the LLM call gets Temporal-managed retries and timeouts while the agent's control loop replays deterministically in the Workflow.

deepagents_plugin/hello_world/workflow.py

# No `workflow.unsafe.imports_passed_through()` guard is needed: the plugin
# configures the workflow sandbox to pass the deepagents / LangChain import
# tree through, so workflow files import them like any other module.
from deepagents import create_deep_agent
from temporalio import workflow


@workflow.defn
class HelloWorldAgent:
    @workflow.run
    async def run(self, question: str) -> str:
        agent = create_deep_agent(
            model="anthropic:claude-sonnet-4-5",
            system_prompt="You are a helpful assistant. Answer concisely.",
        )
        result = await agent.ainvoke(
            {"messages": [{"role": "user", "content": question}]}
        )
        return result["messages"][-1].content

DeepAgentsPlugin is a client-level plugin: add it to Client.connect(...) and the SDK propagates it to any Worker built from that Client. Add it on exactly one side. The plugin registers the deepagents.* Activities and installs the LangChain-aware data converter, so the Worker needs no other wiring.

deepagents_plugin/hello_world/run_worker.py

import asyncio
import os

from temporalio.client import Client
from temporalio.contrib.deepagents import DeepAgentsPlugin
from temporalio.worker import Worker

from deepagents_plugin.hello_world.workflow import HelloWorldAgent


async def main() -> None:
    client = await Client.connect(
        os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"),
        plugins=[DeepAgentsPlugin()],
    )

    worker = Worker(
        client,
        task_queue="deepagents-hello-world",
        workflows=[HelloWorldAgent],
    )
    print("Worker started. Ctrl+C to exit.")
    await worker.run()


if __name__ == "__main__":
    asyncio.run(main())

Use model_activity_options on the plugin to control the timeout and retry policy for each model call. API keys live on the Worker via the model provider (LangChain's init_chat_model by default), never in Workflow inputs or history — the Workflow ships only the model name, and the Worker builds the real client. LLM-SDK-side retries are disabled so that Temporal owns retries and timeouts.

Start the Workflow like any other:

deepagents_plugin/hello_world/run_workflow.py

import asyncio
import os

from temporalio.client import Client

from deepagents_plugin.hello_world.workflow import HelloWorldAgent


async def main() -> None:
    client = await Client.connect(os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"))

    result = await client.execute_workflow(
        HelloWorldAgent.run,
        "What is Temporal in one sentence?",
        id="deepagents-hello-world",
        task_queue="deepagents-hello-world",
    )

    print(f"Result: {result}")


if __name__ == "__main__":
    asyncio.run(main())

Choose Workflow or Activity execution per tool​

A Deep Agent holds its tools in-Workflow. A tool that only reads or writes agent state is pure and belongs there; a tool that does real I/O must not run in Workflow code. The plugin gives you two explicit ways to move a tool's work into an Activity:

  • activity_as_tool(my_activity, ...) surfaces an existing @activity.defn function as a Deep Agents tool without re-declaring it.
  • tool_as_activity(tool, ...) wraps a LangChain tool whose body does I/O so its execution runs as a deepagents.invoke_tool Activity.

An unwrapped, non-builtin tool runs in-Workflow, and the plugin emits a best-effort warning at agent construction (the warning fires through the plugin's create_deep_agent seam, so code paths that bypass that seam construct without it). Deep Agents' pure built-ins (write_todos, the state-backed file tools) stay in-Workflow by design.

This sample wraps an existing activity and an I/O tool, and builds the agent with create_temporal_deep_agent, which wraps the model name in a durable TemporalModel and scopes activity_options to this agent's model calls:

deepagents_plugin/react_agent/workflow.py

@activity.defn
async def get_weather(city: str) -> str:
    """Return the current weather for a city."""
    # A real implementation would call a weather API here; this is a stand-in.
    return f"It is sunny and 22C in {city}."

deepagents_plugin/react_agent/workflow.py

@tool
def web_search(query: str) -> str:
    """Search the web for a query and return a short result."""
    # Real I/O (an HTTP call) would go here; wrapped with tool_as_activity so it
    # runs in an activity, not in workflow code.
    return f"Top result for {query!r}: Temporal makes code durable."


@workflow.defn
class ReactAgent:
    @workflow.run
    async def run(self, question: str) -> str:
        weather_tool = activity_as_tool(
            get_weather,
            start_to_close_timeout=timedelta(seconds=30),
        )
        search_tool = tool_as_activity(
            web_search,
            start_to_close_timeout=timedelta(seconds=30),
        )
        agent = create_temporal_deep_agent(
            model="anthropic:claude-sonnet-4-5",
            tools=[weather_tool, search_tool],
            system_prompt=(
                "You are a research assistant. Use the get_weather and "
                "web_search tools when they help answer the question."
            ),
            # Scopes the model-call activity options to this agent — the
            # recommended way to set model timeouts/retries per agent.
            activity_options={"start_to_close_timeout": timedelta(minutes=2)},
        )
        result = await agent.ainvoke(
            {"messages": [{"role": "user", "content": question}]}
        )
        return result["messages"][-1].content

Durable backends​

A Deep Agent's built-in file tools (write_file, read_file, ls, and so on) delegate to a backend. The default StateBackend keeps files in agent state — pure Workflow state, replay-safe, and needs no wrapping. A FilesystemBackend, LocalShellBackend, or StoreBackend touches real resources, which must not happen from Workflow code.

Wrap such a backend with TemporalBackend(inner, activity_options=...) so each file or shell operation the agent's tools invoke becomes a deepagents.backend_op Activity instead of running in the Workflow. The agent code is unchanged; only the backend is wrapped.

deepagents_plugin/filesystem_backend/workflow.py

from datetime import timedelta

from deepagents import create_deep_agent
from deepagents.backends import FilesystemBackend
from temporalio import workflow
from temporalio.contrib.deepagents import TemporalBackend


@workflow.defn
class FilesystemAgent:
    @workflow.run
    async def run(self, root_dir: str, instruction: str) -> str:
        # Wrap the real-I/O backend so every file op runs in an activity.
        backend = TemporalBackend(
            # virtual_mode roots every path the agent uses under root_dir, so the
            # agent's file tools stay sandboxed to this working directory.
            FilesystemBackend(root_dir=root_dir, virtual_mode=True),
            activity_options={"start_to_close_timeout": timedelta(seconds=30)},
        )
        agent = create_deep_agent(
            model="anthropic:claude-sonnet-4-5",
            # TemporalBackend delegates the backend protocol to the wrapped
            # backend at runtime, which the static type can't see through.
            backend=backend,  # type: ignore[arg-type]
            system_prompt=(
                "You are a file-savvy assistant. Use the write_file and "
                "read_file tools to complete the task."
            ),
        )
        result = await agent.ainvoke(
            {"messages": [{"role": "user", "content": instruction}]}
        )
        return result["messages"][-1].content

Sub-agents​

A coordinator built with create_deep_agent(..., subagents=[...]) delegates to its sub-agents via the built-in task tool. Deep Agents builds each sub-agent as a separate graph, but they inherit the parent's model object by default. Because the plugin makes that model object durable, every sub-agent's model call is automatically durable too — you wire the plugin once and the whole agent tree is covered, with no per-sub-agent wiring.

deepagents_plugin/subagents/workflow.py

from deepagents import create_deep_agent
from temporalio import workflow


@workflow.defn
class SubagentsWorkflow:
    @workflow.run
    async def run(self, question: str) -> str:
        agent = create_deep_agent(
            model="anthropic:claude-sonnet-4-5",
            system_prompt=(
                "You are a research coordinator. Delegate deep investigation to "
                "the researcher sub-agent via the task tool, then synthesize a "
                "final answer."
            ),
            subagents=[
                {
                    "name": "researcher",
                    "description": "Researches a topic in depth and reports findings.",
                    "system_prompt": "You research topics thoroughly and report back.",
                }
            ],
        )
        result = await agent.ainvoke(
            {"messages": [{"role": "user", "content": question}]}
        )
        return result["messages"][-1].content

Human-in-the-loop​

create_deep_agent(..., interrupt_on=...) makes the agent pause before a guarded tool runs. With an in-Workflow InMemorySaver checkpointer, LangGraph does not raise out of ainvoke — it returns the current state with an __interrupt__ entry describing the pending approval. Because the agent loop runs in the Workflow, that pause surfaces directly in Workflow code.

The plugin adds no shim here; the native LangGraph resume protocol is used as-is. The recommended Temporal mapping is to expose the pending approval via a Query and resume via an Update that feeds the human's decision back with Command(resume={"decisions": [...]}). The InMemorySaver is replay-safe because its state lives in the Workflow's own memory, rehydrated by deterministic replay, and the stable Workflow ID is used as the thread_id.

deepagents_plugin/human_in_the_loop/workflow.py

from datetime import timedelta

from deepagents import create_deep_agent
from langchain_core.runnables import RunnableConfig
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
from temporalio import workflow
from temporalio.contrib.deepagents import tool_as_activity


@workflow.defn
class HumanInTheLoopAgent:
    def __init__(self) -> None:
        self._pending: str | None = None
        self._decision: str | None = None
        self._resumed = False

    @workflow.run
    async def run(self, city: str) -> str:
        def book_trip(city: str) -> str:
            """Book a trip to a city (requires human approval)."""
            return f"Booked a trip to {city}."

        trip_tool = tool_as_activity(
            book_trip, start_to_close_timeout=timedelta(seconds=30)
        )
        agent = create_deep_agent(
            model="anthropic:claude-sonnet-4-5",
            tools=[trip_tool],
            interrupt_on={"book_trip": True},
            checkpointer=InMemorySaver(),
        )
        config = RunnableConfig(configurable={"thread_id": workflow.info().workflow_id})

        result = await agent.ainvoke(
            {"messages": [{"role": "user", "content": f"Book a trip to {city}."}]},
            config=config,
        )
        # LangGraph returns (not raises) the pending approval under __interrupt__.
        pending = result.get("__interrupt__")
        if pending:
            self._pending = str(getattr(pending[0], "value", pending[0]))
            # Block until a client approves/rejects via the `resume` update.
            await workflow.wait_condition(lambda: self._resumed)
            # No longer paused: the query goes back to reporting None.
            self._pending = None
            result = await agent.ainvoke(
                Command(resume={"decisions": [{"type": self._decision}]}),
                config=config,
            )
        return result["messages"][-1].content

    @workflow.query
    def pending_approval(self) -> str | None:
        """Return the pending approval prompt, or ``None`` if not paused."""
        return self._pending

    @workflow.update
    async def resume(self, decision: str) -> None:
        """Resume the paused agent with ``"approve"`` or ``"reject"``."""
        self._decision = decision
        self._resumed = True

    @resume.validator
    def validate_resume(self, decision: str) -> None:
        # Runs before the update is accepted, keeping invalid decisions out of
        # workflow history entirely. Only the decisions this workflow feeds to
        # `Command(resume=...)` are allowed.
        if decision not in ("approve", "reject"):
            raise ValueError('decision must be "approve" or "reject"')

Continue-as-new​

Long conversations bloat Workflow history until it hits Temporal's limits. run_deep_agent(agent, input, state_snapshot=...) snapshots state and continues into a fresh run when a turn ends with pending todos and the server recommends continuing (workflow.info().is_continue_as_new_suggested(), the default and recommended mode). Pass continue_as_new_after=N to trigger on a fixed history-event count instead.

  • Carries forward: the accumulated messages and the model/tool result cache, so an LLM or tool call completed before the continue-as-new is not re-run afterward.
  • Does not carry forward: anything else — the snapshot is exactly the messages plus the result cache. Pending todos only gate whether to continue; the todo list itself (and any in-memory checkpointer state) starts fresh in the continued run and is re-derived by the agent from the carried conversation.

Your @workflow.run method must accept the carried state — its signature is run(self, input, state_snapshot=None). On a continue-as-new, run_deep_agent re-invokes the method with args=[input, snapshot], so input must be passed straight through, not re-wrapped, or the carried conversation is corrupted.

deepagents_plugin/continue_as_new/workflow.py

from typing import Any

from deepagents import create_deep_agent
from temporalio import workflow
from temporalio.contrib.deepagents import run_deep_agent


@workflow.defn
class LongResearchAgent:
    @workflow.run
    async def run(
        self, input: dict[str, Any], state_snapshot: dict | None = None
    ) -> str:
        agent = create_deep_agent(
            model="anthropic:claude-sonnet-4-5",
            system_prompt=(
                "You are a research agent. Break large tasks into todos and work "
                "through them until the research is complete."
            ),
        )
        result = await run_deep_agent(
            agent,
            # ``input`` is the messages mapping. Pass it through unchanged: on a
            # continue-as-new, run_deep_agent re-invokes this method with the
            # carried input as its first arg, so re-wrapping it here would nest a
            # dict where a message is expected and corrupt the conversation.
            input,
            # No threshold: continue-as-new fires when the agent still has
            # pending todos and the server suggests continuing — the recommended
            # mode. Pass continue_as_new_after=N to use a fixed event count.
            state_snapshot=state_snapshot,
        )
        return result["messages"][-1].content

Streaming​

Constructing the plugin with DeepAgentsPlugin(streaming_topic="...") flips model dispatch from deepagents.invoke_model to deepagents.invoke_model_streaming: the streaming Activity coalesces chunk batches and publishes them to a WorkflowStream topic for external subscribers, while the aggregated final message is still returned to the Workflow. The durable result is identical to the non-streaming path.

Streaming is async-only, so the Workflow drives an explicit TemporalModel.astream(...) and hosts a WorkflowStream so external subscribers can attach by Workflow ID.

deepagents_plugin/streaming/workflow.py

from langchain_core.messages import HumanMessage
from temporalio import workflow
from temporalio.contrib.deepagents import TemporalModel
from temporalio.contrib.workflow_streams import WorkflowStream

STREAMING_TOPIC = "model-chunks"


@workflow.defn
class StreamingWorkflow:
    def __init__(self) -> None:
        # Host the stream so the publish-Signal handler is registered before the
        # streaming activity (the external publisher) starts publishing.
        self.stream = WorkflowStream()

    @workflow.run
    async def run(self, prompt: str) -> str:
        model = TemporalModel(model="anthropic:claude-sonnet-4-5")
        parts: list[str] = []
        async for chunk in model.astream([HumanMessage(content=prompt)]):
            parts.append(str(chunk.content))
        return "".join(parts)

Compose with an observability plugin​

This plugin carries no tracing context of its own. For observability, compose it with an observability plugin such as temporalio.contrib.langsmith or temporalio.contrib.opentelemetry. Registration order of the two plugins does not matter; the observability plugin captures the LLM calls that DeepAgentsPlugin runs as Activities. The Workflow itself is an ordinary Deep Agent — the tracing comes entirely from composing the plugins on the Client.

deepagents_plugin/langsmith_tracing/workflow.py

from deepagents import create_deep_agent
from temporalio import workflow


@workflow.defn
class TracedAgent:
    @workflow.run
    async def run(self, question: str) -> str:
        agent = create_deep_agent(
            model="anthropic:claude-sonnet-4-5",
            system_prompt="You are a helpful assistant.",
        )
        result = await agent.ainvoke(
            {"messages": [{"role": "user", "content": question}]}
        )
        return result["messages"][-1].content

For agents built directly as LangGraph graphs rather than a compiled Deep Agent, see the LangGraph integration.

Runtime behavior and limitations​

  • Auto-routing is Workflow-scoped. While a Worker built with this plugin is running, the plugin patches Deep Agents' model-resolution seam so a bare model="provider:name" string is auto-routed through an Activity, regardless of how you imported create_deep_agent. This seam is shared by the agent and every sub-agent, and it only rewrites the model when resolved inside a Workflow, so importing deepagents on a plain client or Activity Worker is unaffected, and the patched seam is restored when the Worker stops. Pass TemporalModel("provider:name") yourself if you would rather be explicit.
  • Built-ins stay in-Workflow. Deep Agents' pure built-in tools (write_todos, state-backed file tools) run in-Workflow by design and do not need wrapping. Only tools and backends that do real I/O should be moved to Activities.
  • Durable checkpointers are not replay-safe. The default in-Workflow InMemorySaver is rehydrated for free by deterministic replay. A durable checkpointer that does its own I/O is not replay-safe from inside a Workflow, and the plugin warns if you pass one — prefer the snapshot plus continue-as-new path above.

Samples​

The Deep Agents plugin samples demonstrate all supported patterns, including tool choice, durable backends, sub-agents, human-in-the-loop, continue-as-new, streaming, and observability composition.