temporalio.contrib.deepagents

package documentation

(source)

Temporal plugin for LangChain Deep Agents.

Make an existing Deep Agent durable by adding one plugin: build your agent with create_deep_agent(...) inside a @workflow.defn and add plugins=[DeepAgentsPlugin(...)] to your Client or Worker. 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.

Warning

This package is experimental and may change in future versions.

The public names are imported lazily so import temporalio.contrib.deepagents succeeds before LangChain is installed; touching a name that needs LangChain imports it on first access.

Module	`testing`	Test helpers for users adopting `DeepAgentsPlugin`.
Module	`workflow`	Workflow-side surface: the failure type, the dispatch helpers, and the runner.
Module	`_activity`	The activities that carry every nondeterministic Deep Agents operation.
Module	`_model`	The model seam: a chat model whose every call becomes a Temporal activity.
Module	`_plugin`	The plugin object users add to `plugins=[...]`.
Module	`_serde`	Serialization helpers, a result cache, and worker-runtime configuration.
Module	`_tools`	The tool + backend seams: the explicit per-unit Workflow-vs-Activity choice.

From __init__.py:

Class	`DeepAgentsPlugin`	Temporal plugin that makes LangChain Deep Agents durable.
Class	`TemporalBackend`	Route a real-I/O backend's operations through Temporal activities.
Class	`TemporalModel`	A `BaseChatModel` that runs each generation as a Temporal activity.
Exception	`DeepAgentsWorkflowError`	Raised for non-retryable Deep Agents failures surfaced in the workflow.
Function	`__getattr__`	Undocumented
Function	`activity_as_tool`	Expose an existing Temporal activity as a Deep Agents tool.
Function	`create_temporal_deep_agent`	Build a Deep Agent whose model calls run as durable activities.
Async Function	`run_deep_agent`	Drive `agent.ainvoke(input)` with continue-as-new state carry.
Function	`tool_as_activity`	Wrap a LangChain tool / callable so its execution runs as an activity.

def __getattr__(name: str) -> object: (source) ¶

Undocumented

def activity_as_tool(activity: Callable, *, start_to_close_timeout: timedelta, name: str | None = None, description: str | None = None, retry_policy: Any = None, summary: str | None = None) -> BaseTool: (source) ¶

Expose an existing Temporal activity as a Deep Agents tool.

The returned tool advertises the activity's argument schema to the model and, when called in-workflow, dispatches to the activity via workflow.execute_activity — Temporal owns its retries and timeout.

Parameters
activity:`Callable`	A function decorated with `@activity.defn`.
start_to_close_timeout:`timedelta`	Required per-call timeout for the activity.
name:`str \| None`	Override the tool name advertised to the model (defaults to the activity definition name).
description:`str \| None`	Override the tool description advertised to the model (defaults to the activity docstring).
retry_policy:`Any`	Optional Temporal retry policy for the activity.
summary:`str \| None`	Optional `summary=` recorded on each activity invocation.

Returns
`BaseTool`	Undocumented

def create_temporal_deep_agent(*args: Any, activity_options: Mapping[str, Any] | None = None, **kwargs: Any) -> Any: (source) ¶

Build a Deep Agent whose model calls run as durable activities.

A thin wrapper over deepagents.create_deep_agent that makes the Temporal wiring explicit: a model= name string is wrapped in TemporalModel carrying this agent's activity_options (execute_activity overrides — timeouts, retry policy — for its model calls). Every other argument — tools, backend, sub-agents, interrupt_on — is forwarded unchanged.

Unmodified create_deep_agent(...) also works inside a workflow (the plugin substitutes the durable model automatically, using the plugin's model_activity_options); use this wrapper to scope activity options to one agent instead of configuring them plugin-wide.

async def run_deep_agent(agent: Any, input: Any, *, continue_as_new_after: int | None = None, state_snapshot: Mapping[str, Any] | None = None) -> Any: (source) ¶

Drive agent.ainvoke(input) with continue-as-new state carry.

Once the completed turn leaves pending todos AND history has grown past the limit, the turn's state (messages + the model/tool result cache) is snapshotted and carried into a fresh run via workflow.continue_as_new, so long conversations do not accumulate unbounded history.

By default (continue_as_new_after=None) the limit is the server's own recommendation — workflow.info().is_continue_as_new_suggested() — which accounts for both history length and size; this is the recommended mode. Pass an explicit continue_as_new_after=N to trigger on a fixed history event count instead. To run an agent with NO continue-as-new behavior, call agent.ainvoke(...) directly rather than using this driver.

The enclosing @workflow.run method must accept the continued call — i.e. its signature is (input, state_snapshot=None) — because that is how the carried state is threaded into the next run.

def tool_as_activity(tool: BaseTool | Callable, *, start_to_close_timeout: timedelta, activity_options: Mapping[str, Any] | None = None) -> BaseTool: (source) ¶

Wrap a LangChain tool / callable so its execution runs as an activity.

The underlying tool is registered on the worker; the returned tool keeps the same name and argument schema (so the model's calls are unchanged) but, when invoked in-workflow, dispatches deepagents.invoke_tool instead of running the tool body inline.