temporalio.contrib.deepagents._tools

module documentation
(source)

The tool + backend seams: the explicit per-unit Workflow-vs-Activity choice.

Deep Agents holds its tools and filesystem/shell backends in-workflow. A tool or backend op that only reads and writes DeepAgentState is pure and belongs in the workflow (deterministic, replay-safe). One that does real I/O — a web search, a shell command, a disk write — must not run there. This module gives the user three explicit ways to move that work to an activity:

  • activity_as_tool — expose an existing @activity.defn as a Deep Agents tool (Temporal adopters already have activities; don't make them re-declare);
  • tool_as_activity — wrap a LangChain BaseTool / callable so its execution runs as an activity;
  • TemporalBackend — wrap a real-I/O backend so each file/exec op runs as an activity.

The choice is always explicit: an unwrapped non-builtin tool runs in-workflow, and the plugin warns at construction so that is a conscious decision, never a silent one.

Registries here live in a temporalio-namespaced (sandbox-passthrough) module, so the object the worker's activity sees is the same one the module-level tool_as_activity(...) / TemporalBackend(...) call populated. They hold worker-wide wiring, not per-workflow state.

Function get_registered_tool Look up a tool registered by register_tool.
Function install_backend_async_patch Make BackendProtocol's async defaults workflow-safe.
Function register_backend Record a backend so DeepAgentActivities.backend_op can reach it.
Function register_tool Record tool so DeepAgentActivities.invoke_tool can run it.
Function registered_backends Return the live backend registry (read by the plugin at worker build).
Function set_tool_defaults Install the plugin's default tool_activity_options (called by the plugin).
Function uninstall_backend_async_patch Restore BackendProtocol's upstream async defaults.
Function warn_unwrapped_tools Warn once per unwrapped, non-builtin tool passed to create_deep_agent.
Function _is_coroutine Undocumented
Function _resolve_tool_options Merge plugin tool defaults (possibly per-tool) with a per-call override.
Function _unregister_backend Drop ref from the registry if it still maps to inner.
Constant _ASYNC_TO_SYNC_OPS Undocumented
Constant _BACKEND_OPS Undocumented
Constant _BACKEND_REGISTRY Undocumented
Constant _BACKEND_REGISTRY_LOCK Undocumented
Constant _BUILTIN_TOOL_NAMES Undocumented
Constant _ROUTED_TOOL_NAMES Undocumented
Constant _TOOL_REGISTRY Undocumented
Variable _original_backend_async_defaults Undocumented
Variable _tool_defaults Undocumented
def get_registered_tool(name: str) -> BaseTool | None: (source)

Look up a tool registered by register_tool.

def install_backend_async_patch(): (source)

Make BackendProtocol's async defaults workflow-safe.

Inside a workflow, run the sync twin inline: for state-only backends that is deterministic and semantically identical to the upstream default, which merely moves the same sync call onto a worker thread. Outside a workflow (activities, clients) the upstream default — thread hop plus timeout guard — is used unchanged. Subclasses that override an async method natively are unaffected; only the protocol defaults are replaced. Idempotent.

def register_backend(ref: str, backend: Any): (source)

Record a backend so DeepAgentActivities.backend_op can reach it.

def register_tool(tool: BaseTool): (source)

Record tool so DeepAgentActivities.invoke_tool can run it.

def registered_backends() -> dict[str, Any]: (source)

Return the live backend registry (read by the plugin at worker build).

def set_tool_defaults(options: Any): (source)

Install the plugin's default tool_activity_options (called by the plugin).

Accepts a single ActivityConfig or a Mapping[tool_name, ActivityConfig].

def uninstall_backend_async_patch(): (source)

Restore BackendProtocol's upstream async defaults.

def warn_unwrapped_tools(tools: Any): (source)

Warn once per unwrapped, non-builtin tool passed to create_deep_agent.

Running a tool in-workflow is only safe if it is pure/deterministic. The Workflow-vs-Activity choice must be conscious, so any user tool that was not routed through tool_as_activity / activity_as_tool gets a construction-time warning rather than silently executing in the workflow.

def _is_coroutine(fn: Any) -> bool: (source)

Undocumented

def _resolve_tool_options(tool_name: str, instance_options: Mapping[str, Any] | None) -> dict[str, Any]: (source)

Merge plugin tool defaults (possibly per-tool) with a per-call override.

def _unregister_backend(ref: str, inner: Any): (source)

Drop ref from the registry if it still maps to inner.

GC hook for TemporalBackend (via weakref.finalize): a wrapper is typically constructed per workflow run, so without cleanup a long-lived worker accumulates one registry entry per run. The identity guard is load-bearing: refs are deterministic per run, so after a cache eviction a replay re-registers the same ref with a fresh inner backend — the evicted wrapper's finalizer must not remove that live registration.

_ASYNC_TO_SYNC_OPS: dict[str, str] = (source)

Undocumented

Value
{'als': 'ls',
 'als_info': 'ls_info',
 'aread': 'read',
 'awrite': 'write',
 'aedit': 'edit',
 'aglob': 'glob',
 'aglob_info': 'glob_info',
...
_BACKEND_OPS: tuple[str, ...] = (source)

Undocumented

Value
('ls',
 'ls_info',
 'read',
 'write',
 'edit',
 'glob',
 'glob_info',
...
_BACKEND_REGISTRY: dict[str, Any] = (source)

Undocumented

Value
{}
_BACKEND_REGISTRY_LOCK = (source)

Undocumented

Value
threading.Lock()
_BUILTIN_TOOL_NAMES = (source)

Undocumented

Value
frozenset(set(['write_todos',
               'ls',
               'read_file',
               'write_file',
               'edit_file',
               'glob',
               'grep',
...
_ROUTED_TOOL_NAMES: set[str] = (source)

Undocumented

Value
set()
_TOOL_REGISTRY: dict[str, BaseTool] = (source)

Undocumented

Value
{}
_original_backend_async_defaults: dict[str, Any] = (source)

Undocumented

_tool_defaults: dict[str, Any] = (source)

Undocumented