temporalio.contrib.deepagents._serde

module documentation
(source)

Serialization helpers, a result cache, and worker-runtime configuration.

The Deep Agents control loop runs inside the Temporal workflow, so the values that actually cross the workflow⇄activity boundary are a small set of LangChain types: chat messages, tool-call descriptors, and the RunnableConfig metadata attached to each model / tool call. LangChain messages are polymorphic Serializable models (an AIMessage must not be rehydrated as a ToolMessage) and RunnableConfig carries live callback / checkpointer references, so neither survives a naive round-trip. This module owns:

  • message (de)serialization via langchain_core.load.dumpd / load — the round-trip that preserves message subtype and tool-call structure;
  • the strip → ship → rebuild dance for RunnableConfig;
  • tool → JSON-schema advertisement (full name + description + argument schema, never {name, description} alone — without the argument schema the model picks the right tool but hallucinates its arguments);
  • the Pydantic data converter (exclude_unset=True) the plugin installs on the client and replayer, so message payloads stay small and round-trip cleanly;
  • a workflow-scoped result cache so model / tool results computed before a continue_as_new are reused rather than recomputed after it;
  • the sandbox passthrough module list covering LangChain's transitive eager-import tree.

LangChain imports are deferred into the functions that need them, so importing this module — and constructing the plugin — does not require LangChain to be installed on the machine assembling the worker.

Class DeepAgentsPayloadConverter Pydantic payload converter pinned to exclude_unset=True.
Class Settings Dispatch defaults shared by every TemporalModel on the worker.
Function build_data_converter Compose the plugin's converter with whatever the caller already set.
Function cache_key Stable key over (kind, call_id, args) for cache lookups.
Function cache_lookup Return (hit, value) for key in the active cache.
Function cache_put Record value under key when a cache is active for this run.
Function default_passthrough_modules The LangChain / deepagents transitive import tree passed through the sandbox.
Function dump_backend_result Encode a backend op's return value for the activity boundary.
Function dump_messages Serialize a sequence of LangChain messages to their dumpd form.
Function dump_object Serialize a single LangChain Serializable (message, tool call, …).
Function get_settings Return the active dispatch settings.
Function load_backend_result Rebuild a value produced by dump_backend_result.
Function load_messages Rehydrate messages serialized by dump_messages.
Function load_object Rehydrate a value produced by dump_object, preserving subtype.
Function rebuild_runnable_config Reconstruct a minimal RunnableConfig from strip_runnable_config.
Function resolve_passthrough_modules Merge caller-supplied passthrough modules with the plugin defaults.
Function result_cache_snapshot Return a serializable copy of the cache, or None when empty.
Function set_result_cache Seed the workflow-scoped result cache (e.g. carried across CAN).
Function set_settings Install the worker-wide model dispatch defaults (called by the plugin).
Function strip_runnable_config Reduce a live RunnableConfig to its JSON-safe subset for shipping.
Function tool_to_schema Advertise a tool to the model as a full OpenAI tool schema.
Variable data_converter The plugin's default data converter (LangChain messages are shipped as their dumpd JSON form, so the Pydantic converter only ever sees plain containers).
Function _is_jsonish Undocumented
Constant _BACKEND_DATACLASS_KEY Undocumented
Constant _DEFAULT_PASSTHROUGH Undocumented
Variable _result_cache Undocumented
Variable _settings Undocumented
def build_data_converter(user_converter: DataConverter | None) -> DataConverter: (source)

Compose the plugin's converter with whatever the caller already set.

  • None — install the plugin default.
  • the SDK default converter — swap in the LangChain-aware Pydantic converter via dataclasses.replace.
  • a custom converter — refuse rather than silently clobber it; the caller must fold DeepAgentsPayloadConverter into their own converter.
def cache_key(kind: str, call_id: str, args: Any) -> str: (source)

Stable key over (kind, call_id, args) for cache lookups.

def cache_lookup(key: str) -> tuple[bool, Any]: (source)

Return (hit, value) for key in the active cache.

def cache_put(key: str, value: Any): (source)

Record value under key when a cache is active for this run.

def default_passthrough_modules() -> tuple[str, ...]: (source)

The LangChain / deepagents transitive import tree passed through the sandbox.

def dump_backend_result(value: Any) -> Any: (source)

Encode a backend op's return value for the activity boundary.

Backend protocol results (WriteResult / ReadResult / GrepResult and their nested FileInfo / GrepMatch items, …) are plain dataclasses — not LangChain Serializable objects — and the filesystem middleware reads their ATTRIBUTES in-workflow, so a plain JSON round-trip (which decays them to dicts) breaks the seam at the first real backend op. Tag deepagents dataclasses with their import path so load_backend_result rebuilds the real type; anything else (str, dict, a custom backend's own types) passes through with today's plain-JSON behavior.

def dump_messages(messages: Any) -> list[Any]: (source)

Serialize a sequence of LangChain messages to their dumpd form.

def dump_object(obj: Any) -> Any: (source)

Serialize a single LangChain Serializable (message, tool call, …).

def get_settings() -> Settings: (source)

Return the active dispatch settings.

def load_backend_result(value: Any) -> Any: (source)

Rebuild a value produced by dump_backend_result.

Only deepagents.* dataclasses are reconstructed (the tag is written exclusively for them); anything else would mean a forged payload, so refuse rather than import arbitrary types. Reconstruction suppresses DeprecationWarning: this is transport, not user code — the backend already constructed the object once on the activity side, and required deprecated fields (e.g. WriteResult.files_update) would otherwise warn on every op. Field values equal to a declared default are omitted from the constructor call.

def load_messages(dumped: list[Any]) -> list[Any]: (source)

Rehydrate messages serialized by dump_messages.

def load_object(data: Any) -> Any: (source)

Rehydrate a value produced by dump_object, preserving subtype.

def rebuild_runnable_config(data: dict[str, Any]) -> RunnableConfig: (source)

Reconstruct a minimal RunnableConfig from strip_runnable_config.

def resolve_passthrough_modules(user: Any) -> tuple[str, ...]: (source)

Merge caller-supplied passthrough modules with the plugin defaults.

def result_cache_snapshot() -> dict[str, Any] | None: (source)

Return a serializable copy of the cache, or None when empty.

def set_result_cache(cache: dict[str, Any] | None): (source)

Seed the workflow-scoped result cache (e.g. carried across CAN).

def set_settings(*, model_activity_options: Any = None, streaming_topic: str | None = None): (source)

Install the worker-wide model dispatch defaults (called by the plugin).

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

Reduce a live RunnableConfig to its JSON-safe subset for shipping.

Keeps tags, run_name, run_id, recursion_limit, JSON-safe metadata and the JSON-safe (non-dunder) configurable keys. Drops callbacks, checkpointer / store / cache handles and every other live reference — those are reconstructed activity-side.

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

Advertise a tool to the model as a full OpenAI tool schema.

Carries name + description + argument JSON schema so the model can build valid arguments, not just select the tool by name.

data_converter = (source)

The plugin's default data converter (LangChain messages are shipped as their dumpd JSON form, so the Pydantic converter only ever sees plain containers).

def _is_jsonish(value: Any) -> bool: (source)

Undocumented

_BACKEND_DATACLASS_KEY: str = (source)

Undocumented

Value
'__deepagents_dataclass__'
_DEFAULT_PASSTHROUGH: tuple[str, ...] = (source)

Undocumented

Value
('langchain',
 'langchain_core',
 'langchain_anthropic',
 'langgraph',
 'deepagents',
 'langsmith',
 'numpy',
...
_result_cache: contextvars.ContextVar[dict[str, Any] | None] = (source)

Undocumented

_settings = (source)

Undocumented