temporalio.contrib.deepagents.DeepAgentsPlugin

class documentation

class DeepAgentsPlugin(SimplePlugin): (source)

Constructor: DeepAgentsPlugin(model_provider, model_activity_options, tool_activity_options, streaming_topic, ...)

View In Hierarchy

Temporal plugin that makes LangChain Deep Agents durable.

Parameters
model_providerBuilds a real chat model from a name string, worker-side. This is where API keys live; only the model name ever crosses the workflow boundary. Defaults to LangChain's init_chat_model with LLM-SDK retries disabled (Temporal owns retries).
model_activity_optionsActivityConfig (or Mapping[model_name, ActivityConfig]) for the model activities — timeouts, retry policy.
tool_activity_optionsActivityConfig (or Mapping[tool_name, ActivityConfig]) default for tools wrapped with tool_as_activity.
streaming_topicWhen set, model calls stream through the streaming activity and publish chunk batches to this workflow-streams topic.
streaming_batch_intervalHow long the streaming activity coalesces chunks before publishing a batch.
passthrough_modulesExtra sandbox-passthrough modules, merged with the plugin's LangChain/deepagents defaults.
data_converterOverride the default LangChain-aware converter. None installs the default; the SDK default is upgraded in place; any other converter raises (fold DeepAgentsPayloadConverter into your own).
Method __init__ Configure the plugin; see the class docstring for parameters.
Method configure_worker Deduplicate activity registrations after the base configuration.
Method _make_workflow_runner Undocumented
Async Method _run_context Patch Deep Agents' model resolution seam for the worker's lifetime.
Instance Variable _activities Undocumented
Instance Variable _data_converter Undocumented
Instance Variable _passthrough_modules Undocumented
Instance Variable _tool_activity_options Undocumented

Inherited from SimplePlugin:

Method configure_client See base class.
Method configure_replayer See base class.
Async Method connect_service_client See base class.
Method name See base class.
Async Method run_replayer See base class.
Async Method run_worker See base class.
Instance Variable activities Undocumented
Instance Variable data_converter Undocumented
Instance Variable interceptors Undocumented
Instance Variable nexus_service_handlers Undocumented
Instance Variable run_context Undocumented
Instance Variable workflow_failure_exception_types Undocumented
Instance Variable workflow_runner Undocumented
Instance Variable workflows Undocumented
Instance Variable _name Undocumented
def __init__(self, *, model_provider: Callable[[str], Any] | None = None, model_activity_options: Any = None, tool_activity_options: Any = None, streaming_topic: str | None = None, streaming_batch_interval: timedelta = timedelta(milliseconds=100), passthrough_modules: Sequence[str] | None = None, data_converter: Any = None): (source)
overrides temporalio.plugin.SimplePlugin.__init__

Configure the plugin; see the class docstring for parameters.

def configure_worker(self, config: Any) -> Any: (source)
overrides temporalio.plugin.SimplePlugin.configure_worker

Deduplicate activity registrations after the base configuration.

def _make_workflow_runner(self) -> Callable[[(WorkflowRunner | None)], WorkflowRunner]: (source)

Undocumented

@asynccontextmanager
async def _run_context(self) -> AsyncIterator[None]: (source)

Patch Deep Agents' model resolution seam for the worker's lifetime.

The patch (on deepagents.graph.resolve_model, the seam create_deep_agent uses for the agent and every sub-agent) only rewrites model= strings when running inside a workflow, so it is inert on plain clients / activity workers. Determinism itself rides on the same mechanism contrib.langgraph uses — the loop is in-workflow and every nondeterministic call is an activity — so no speculative time/uuid shims are installed here.

One shim is required: LangChain's create_agent factory wraps every model node with LangSmith's @traceable, whose async path hops onto a thread via asyncio.run_in_executor — which the deterministic workflow event loop does not implement (it raises NotImplementedError). Tracing is an observability concern that must not run in-workflow, so we install LangSmith's own Temporal escape hatch (set_runtime_overrides) to run that setup inline when in_workflow(), and defer to the default thread hop everywhere else (activities / clients, where tracing is fine).

_activities = (source)

Undocumented

_data_converter = (source)

Undocumented

_passthrough_modules = (source)

Undocumented

_tool_activity_options = (source)

Undocumented