temporalio.contrib.deepagents._plugin

module documentation
(source)

The plugin object users add to plugins=[...].

DeepAgentsPlugin wires everything together so that existing deepagents code — create_deep_agent(...).ainvoke(...) — runs durably inside a @workflow.defn with no other changes:

  • registers the four activities that carry the nondeterministic work;
  • installs the LangChain-aware data converter (composing, never clobbering, a user converter);
  • passes the LangChain / LangGraph / deepagents import tree through the workflow sandbox;
  • wraps the worker run in a run_context that patches Deep Agents' model resolution seam (deepagents.graph.resolve_model) to auto-route bare model= strings through activities — regardless of how the user imported create_deep_agent;
  • registers DeepAgentsWorkflowError as a workflow-failure type;
  • pushes the model / tool dispatch defaults down to the seams that read them.

The plugin auto-propagates from a Client to any Worker built from it, so add it on exactly one side. configure_worker additionally de-duplicates activities by name, so a user who mistakenly adds it on both sides gets a clean no-op instead of a "More than one activity named ..." crash.

Function _dedupe_activities Drop duplicate activity registrations by defn name, keeping the first.
Function _install_langsmith_temporal_override Route LangSmith's thread hop inline while in a workflow.
Async Function _temporal_aio_to_thread Run LangSmith's aio_to_thread seam safely inside a workflow.
Constant _MIN_PYTHON Undocumented
def _dedupe_activities(activities: Sequence[Any]) -> list[Any]: (source)

Drop duplicate activity registrations by defn name, keeping the first.

Guards the "plugin added on both Client and Worker" trap, where the plugin's activities would otherwise be appended twice and the worker would reject the duplicate names.

def _install_langsmith_temporal_override(): (source)

Route LangSmith's thread hop inline while in a workflow.

set_runtime_overrides mutates a module global in the sandbox-passthrough langsmith package, so the in-workflow tracer sees it too. No-op (and harmless) when LangSmith is not installed or too old to expose set_runtime_overrides.

The override is deliberately installed for the life of the process and never uninstalled: LangSmith exposes a single process-wide override slot (each set_runtime_overrides call replaces it wholesale), and temporalio.contrib.langsmith installs a behaviorally identical override exactly once, without reinstalling. Resetting the slot on this worker's shutdown would therefore permanently strip a composed LangSmith plugin's workflow-safety override. Leaving it installed is safe: the override defers to LangSmith's default thread hop whenever workflow.in_workflow() is false, so it is inert outside workflows.

async def _temporal_aio_to_thread(default_aio_to_thread: Callable[..., Any], ctx: Any, func: Callable[..., Any], /, *args: Any, **kwargs: Any) -> Any: (source)

Run LangSmith's aio_to_thread seam safely inside a workflow.

Outside a workflow (activities, clients) we defer to LangSmith's default thread hop. Inside a workflow the deterministic event loop cannot spawn a thread, so we run the setup inline in the requested context. func here is LangSmith's own run-tree bookkeeping — cheap and side-effect-free with respect to workflow determinism.

_MIN_PYTHON: tuple[int, ...] = (source)

Undocumented

Value
(3, 11)