temporalio.contrib.deepagents.workflow

module documentation
(source)

Workflow-side surface: the failure type, the dispatch helpers, and the runner.

Everything here runs inside the workflow. The dispatch helpers (call_model / call_tool / call_backend_op) are the single choke point through which the in-workflow model / tool / backend stubs reach their activities; they also consult the continue-as-new result cache so work done before a continue_as_new is reused rather than repeated after it.

run_deep_agent is the optional driver that adds continue-as-new state-carry around a native agent.ainvoke(...) — plain agent.ainvoke(...) still works without it.

Async Function call_backend_op Dispatch one backend op, reusing a cached result across continue-as-new.
Async Function call_model Dispatch one model call, reusing a cached result across continue-as-new.
Async Function call_tool Dispatch one tool call, reusing a cached result across continue-as-new.
Function warn_durable_checkpointer Warn when a user hands create_deep_agent a durable checkpointer.
Function _as_message_list Undocumented
Function _extract_messages Undocumented
Function _has_pending_work True when the agent left unfinished todos worth carrying past a CAN.
Function _merge_snapshot Prepend a snapshot's carried messages onto the next turn's input.
Constant _CACHE_KEY Undocumented
Constant _IN_WORKFLOW_SAVERS Undocumented
async def call_backend_op(activity_input: _activity.BackendOpInput, *, summary: str, **opts: Any) -> _activity.BackendOpOutput: (source)

Dispatch one backend op, reusing a cached result across continue-as-new.

async def call_model(activity_name: str, activity_input: _activity.ModelActivityInput, *, summary: str, **opts: Any) -> _activity.ModelActivityOutput: (source)

Dispatch one model call, reusing a cached result across continue-as-new.

async def call_tool(activity_input: _activity.ToolActivityInput, *, summary: str, **opts: Any) -> _activity.ToolActivityOutput: (source)

Dispatch one tool call, reusing a cached result across continue-as-new.

def warn_durable_checkpointer(checkpointer: Any): (source)

Warn when a user hands create_deep_agent a durable checkpointer.

The Deep Agents loop runs inside the workflow, so a checkpointer that does its own database / disk I/O would run that I/O from workflow code — not replay-safe. We respect the user's choice (a warning, not a hard failure), and point them at the durability path that is safe: the default in-workflow InMemorySaver rehydrated by replay, plus run_deep_agent with continue_as_new_after for long conversations.

def _as_message_list(input: Any) -> list[Any]: (source)

Undocumented

def _extract_messages(result: Any) -> list[Any]: (source)

Undocumented

def _has_pending_work(result: Any) -> bool: (source)

True when the agent left unfinished todos worth carrying past a CAN.

A finished single-shot run has no pending todos, so this returns False and the driver returns the result instead of looping on continue-as-new forever.

def _merge_snapshot(input: Any, snapshot: Mapping[str, Any]) -> Any: (source)

Prepend a snapshot's carried messages onto the next turn's input.

_CACHE_KEY: str = (source)

Undocumented

Value
'__temporal_cache__'
_IN_WORKFLOW_SAVERS = (source)

Undocumented

Value
frozenset(set(['InMemorySaver', 'MemorySaver']))