module documentation

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. Every dispatch runs its own Activity; only executions recorded before the deepagents.retire-result-cache patch consult the legacy continue-as-new result cache during replay.

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 as its own Activity.
Async Function call_model Dispatch one model call as its own Activity.
Async Function call_tool Dispatch one tool call as its own Activity.
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 _legacy_lookup Legacy-cache lookup: (key, hit, value); key is None on new executions.
Function _legacy_result_cache Whether this execution uses the legacy continue-as-new result cache.
Function _merge_snapshot Prepend a snapshot's carried messages onto the next turn's input.
Constant _CACHE_KEY Undocumented
Constant _IN_WORKFLOW_SAVERS Undocumented
Constant _INPUT_CARRIED_KEY Undocumented
async def call_backend_op(activity_input: _activity.BackendOpInput, *, summary: str, **opts: Any) -> _activity.BackendOpOutput: (source)

Dispatch one backend op as its own Activity.

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

Dispatch one model call as its own Activity.

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

Dispatch one tool call as its own Activity.

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 _legacy_lookup(kind: str, name: str, payload: Any) -> tuple[str | None, bool, Any]: (source)

Legacy-cache lookup: (key, hit, value); key is None on new executions.

def _legacy_result_cache() -> bool: (source)

Whether this execution uses the legacy continue-as-new result cache.

New executions do not cache at all: repeated identical calls are legitimate work (a re-issued tool call, a deliberate model resample, a re-read after a write), and under the resume-from-transcript continue-as-new semantics a continued run never re-executes prior dispatches — so a carried cache entry could only ever serve a stale result to a genuinely new call. Replay of a single run needs no cache: history supplies recorded activity results.

Patch-gated because histories recorded under the legacy cache contain dedup decisions (a repeated call answered with no activity scheduled); replaying them without the cache would emit commands history does not have.

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

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

The driver's own continue-as-new re-invocation avoids duplicating the original prompt: a Mapping input travels without its "messages" key, and a bare (non-Mapping) prompt travels as-is with a snapshot marker telling this merge not to re-append it (the type must survive for the user's @workflow.run signature). An externally supplied state_snapshot plus a fresh input still composes: carried history first, new input after. Agents are expected to return the accumulated transcript in result["messages"] (as deepagents/LangGraph reducers do) — the carry only strips input messages when the transcript is non-empty.

_CACHE_KEY: str = (source)

Undocumented

Value
'__temporal_cache__'
_IN_WORKFLOW_SAVERS = (source)

Undocumented

Value
frozenset(set(['InMemorySaver', 'MemorySaver']))
_INPUT_CARRIED_KEY: str = (source)

Undocumented

Value
'__temporal_input_in_transcript__'