# Deep Agents integration > For the complete documentation index, see [llms.txt](https://docs.temporal.io/llms.txt). > Any documentation page is available as raw Markdown by appending `.md` to its URL. Temporal's integration with [Deep Agents](https://github.com/langchain-ai/deepagents) 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. > **Pre-release** Code snippets in this guide are taken from the [Deep Agents plugin samples](https://github.com/temporalio/samples-python/tree/main/deepagents_plugin). Refer to the samples for the complete code. ## Prerequisites - Python 3.11 or newer. - This guide assumes you are already familiar with Deep Agents. If you aren't, refer to the [Deep Agents documentation](https://github.com/langchain-ai/deepagents) for more details. - If you are new to Temporal, we recommend reading [Understanding Temporal](/evaluate/understanding-temporal) or taking the [Temporal 101](https://learn.temporal.io/courses/temporal_101/) course. - Ensure you have set up your local development environment by following the [Set up your local development environment](/develop/python/set-up-your-local-python) guide. When you're done, leave the Temporal development server running if you want to test your code locally. ## 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`: ```bash uv add "temporalio[deepagents] @ git+https://github.com/temporalio/sdk-python.git" ``` or with pip: ```bash 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](https://github.com/temporalio/samples-python/blob/main/deepagents_plugin/hello_world/workflow.py) ```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](https://github.com/temporalio/samples-python/blob/main/deepagents_plugin/hello_world/run_worker.py) ```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](https://github.com/temporalio/samples-python/blob/main/deepagents_plugin/hello_world/run_workflow.py) ```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](https://github.com/temporalio/samples-python/blob/main/deepagents_plugin/react_agent/workflow.py) ```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](https://github.com/temporalio/samples-python/blob/main/deepagents_plugin/react_agent/workflow.py) ```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](https://github.com/temporalio/samples-python/blob/main/deepagents_plugin/filesystem_backend/workflow.py) ```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](https://github.com/temporalio/samples-python/blob/main/deepagents_plugin/subagents/workflow.py) ```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](/develop/python/workflows/message-passing#queries) and resume via an [Update](/develop/python/workflows/message-passing#updates) 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](https://github.com/temporalio/samples-python/blob/main/deepagents_plugin/human_in_the_loop/workflow.py) ```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](https://github.com/temporalio/samples-python/blob/main/deepagents_plugin/continue_as_new/workflow.py) ```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`](https://github.com/temporalio/sdk-python/tree/main/temporalio/contrib/workflow_streams) 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](https://github.com/temporalio/samples-python/blob/main/deepagents_plugin/streaming/workflow.py) ```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`](https://github.com/temporalio/sdk-python/tree/main/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](https://github.com/temporalio/samples-python/blob/main/deepagents_plugin/langsmith_tracing/workflow.py) ```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](/develop/python/integrations/langgraph). ## 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](https://github.com/temporalio/samples-python/tree/main/deepagents_plugin) demonstrate all supported patterns, including tool choice, durable backends, sub-agents, human-in-the-loop, continue-as-new, streaming, and observability composition.