[AI-250] Add Deep Agents plugin#1644
Conversation
temporalio.contrib.deepagents makes LangChain Deep Agents durable: unmodified create_deep_agent(...).ainvoke(...) code runs inside a workflow with plugins=[DeepAgentsPlugin()], routing every model turn, I/O tool call, and backend file/shell op through Temporal activities while the agent loop replays deterministically. Includes explicit per-tool workflow-vs-activity choice (activity_as_tool / tool_as_activity), TemporalBackend interception of the full backend protocol (sync + async + execute) with typed protocol-dataclass round-trip, continue-as-new state carry via run_deep_agent, streaming through workflow_streams, and native LangGraph interrupt/resume for human-in-the-loop. Ships as the temporalio[deepagents] extra (Python >= 3.11, matching deepagents' own floor).
deepagents' BackendProtocol implements every async method's DEFAULT as asyncio.to_thread(sync_twin, ...), and neither StateBackend nor FilesystemBackend overrides any of them. The deterministic workflow event loop has no thread executor, so an agent's first built-in tool call against an unwrapped in-workflow backend — e.g. a model spontaneously invoking grep on the default state backend — failed the workflow task with NotImplementedError. Scripted-model tests never invoke built-ins on the default backend, so only a live-model run surfaced it. The plugin now patches the protocol's async defaults at worker start (same seam pattern as the resolve_model patch): inside a workflow the sync twin runs inline, which for state-only backends is deterministic and semantically identical to the upstream default; outside a workflow the upstream thread-hop default is untouched, as are subclasses that override an async method natively.
tool_as_activity returned the ToolMessage assembled inside the invoke_tool activity, whose tool_call_id is workflow-generated — the activity cannot know the id the model minted for the call. A real provider (Anthropic) rejects the next model turn with 400 "unexpected tool_use_id found in tool_result blocks" because the tool_result does not pair with any tool_use in the previous message. Offline fakes never validate the pairing, so only a live-model run surfaced it. The wrapper now returns the tool result CONTENT and lets the tool node stamp the model's own tool_call_id — the same path unwrapped tools take. activity_as_tool already returned raw content and is unaffected. The regression test records the fake model's second-turn request and asserts the tool_result rides under the scripted id.
Two CI-only failure classes the local scoped runs missed: - basedpyright fails on warnings repo-wide. Replace deprecated typing aliases (Mapping/Sequence/AsyncIterator/Iterator/List/Optional/Union) with collections.abc / PEP 604 forms, type the test fixtures the way the rest of the suite does, drop unused imports/params, and keep the interpreter-floor warning behind a module constant so newer runtimes do not narrow it into unreachable code. - Python 3.10 jobs cannot install deepagents (its floor is 3.11), so pyright cannot resolve those imports there. Runtime imports of deepagents/langchain in module code go through importlib (attribute access on ModuleType is dynamic; monkeypatch writes use setattr), and the deepagents test modules carry a file-level pyright directive alongside their existing importorskip guards. langchain-core stays statically imported — it resolves everywhere via the langgraph extra.
The API-docs build rejects an inline literal whose end-string is followed by a letter (``Serializable``s), and cannot resolve a :class: link to the package re-export, so both become plain prose / literals. On 3.10, where the real deepagents package cannot install, basedpyright resolves `from deepagents import ...` in tests/contrib/deepagents/ implicitly relative to the same-named test directory — extend the existing file-level directive; Python 3 has no implicit relative imports at runtime and collection is already importorskip-gated.
The previous round's file-level directive used reportImplicitRelativeImport, which is basedpyright-only vocabulary — plain pyright hard-errors on unknown rules in pyright comments, taking every lint leg down. Rather than juggle two checkers' rule sets, drop the static `from deepagents import ...` lines from the tests entirely: symbols now bind off the module object pytest.importorskip already returns, which is dynamically typed for every checker, resolves nothing against the same-named test directory on 3.10, and lets the directives (and the now-moot protocol cast) be deleted outright.
There was a problem hiding this comment.
Pull request overview
Adds a new temporalio.contrib.deepagents integration that makes LangChain Deep Agents durable inside Temporal Workflows by routing model/tool/backend I/O through Activities while keeping the agent loop deterministic and replay-safe in-workflow.
Changes:
- Introduces
DeepAgentsPluginplus workflow-side helpers (run_deep_agent, dispatch choke points) and worker-side activities for model/tool/backend execution. - Adds serialization/caching utilities for LangChain objects and deepagents backend dataclasses, plus sandbox passthrough configuration.
- Adds a comprehensive pytest suite under
tests/contrib/deepagents/and wires new packaging extras/deps (temporalio[deepagents], Python ≥ 3.11).
Reviewed changes
Copilot reviewed 24 out of 27 changed files in this pull request and generated 4 comments.
Show a summary per file
| File | Description |
|---|---|
| uv.lock | Adds locked dependencies for the new deepagents extra and related LangChain packages. |
| pyproject.toml | Adds the deepagents extra + dev deps gated to Python ≥ 3.11. |
| tests/contrib/deepagents/init.py | New test package marker for the deepagents contrib tests. |
| tests/contrib/deepagents/helpers.py | Adds shared helper to count scheduled activities from workflow history. |
| tests/contrib/deepagents/test_backends.py | Tests TemporalBackend routing (including deepagents FilesystemBackend async protocol + replay). |
| tests/contrib/deepagents/test_checkpointer.py | Tests durable-checkpointer warning semantics and replay rehydration behavior. |
| tests/contrib/deepagents/test_continue_as_new.py | E2E tests for run_deep_agent(..., continue_as_new_after=...) snapshot/carry and cache reuse. |
| tests/contrib/deepagents/test_failures.py | Tests HTTP-error→retry translation and stable workflow failure typing. |
| tests/contrib/deepagents/test_hitl.py | Tests HITL interrupt surfaced via Query + resume via Update/Command protocol. |
| tests/contrib/deepagents/test_model_activity.py | Tests TemporalModel calls becoming a single model activity (and per-model overrides). |
| tests/contrib/deepagents/test_native_e2e.py | End-to-end “plugin only” durability test for unmodified create_deep_agent(...).ainvoke(...). |
| tests/contrib/deepagents/test_readme.py | Ensures README fenced python snippets compile. |
| tests/contrib/deepagents/test_replay.py | Records history and replays it through Replayer with the plugin configured. |
| tests/contrib/deepagents/test_side_effects.py | Forces replay (max_cached_workflows=0) and asserts bounded/expected activity scheduling. |
| tests/contrib/deepagents/test_streaming.py | Tests streaming dispatch via invoke_model_streaming and batch interval wiring. |
| tests/contrib/deepagents/test_subagents.py | Tests sub-agent model calls still route via activities. |
| tests/contrib/deepagents/test_tools.py | Tests activity_as_tool and tool_as_activity routing + tool_call_id pairing behavior. |
| temporalio/contrib/deepagents/init.py | Lazy public surface for the plugin and helpers without requiring LangChain at import time. |
| temporalio/contrib/deepagents/README.md | Documents installation and usage patterns (hello world, CAN snapshotting, streaming, composition). |
| temporalio/contrib/deepagents/_activity.py | Implements the four activities and boundary dataclasses, plus error translation and heartbeating. |
| temporalio/contrib/deepagents/_model.py | Implements TemporalModel and patches deepagents model resolution inside workflows. |
| temporalio/contrib/deepagents/_plugin.py | Wires activities, data converter, sandbox passthrough, run-context patching, and failure typing. |
| temporalio/contrib/deepagents/_serde.py | Provides LangChain object (de)serialization, runnable config stripping, result cache, passthrough list. |
| temporalio/contrib/deepagents/_tools.py | Implements tool/backend seams (activity_as_tool, tool_as_activity, TemporalBackend) and registries. |
| temporalio/contrib/deepagents/testing.py | Provides offline testing helpers (scripted fake model and mock model provider). |
| temporalio/contrib/deepagents/workflow.py | Provides workflow-side dispatch choke points, durable-checkpointer warning, and CAN runner. |
| temporalio/contrib/deepagents/py.typed | Marks the package as typed for downstream type checkers. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| try: | ||
| return await fn(*args, **kwargs) | ||
| finally: | ||
| if beat_task is not None: | ||
| beat_task.cancel() | ||
|
|
There was a problem hiding this comment.
Good catch — fixed in a54d445. The finally block now awaits the cancelled task via await asyncio.wait([beat_task]), which lets the cancellation land without re-raising the task's CancelledError, so no pending task outlives the activity. (Small note: the auto_heartbeater helper is in the samples-python repo rather than this one, but it uses this same cancel-then-wait pattern and we now match it.)
| patched = False | ||
| try: | ||
| from temporalio.contrib.deepagents import _model, _tools | ||
|
|
||
| _model.install_model_patch() | ||
| _tools.install_backend_async_patch() | ||
| patched = True | ||
| except Exception: # LangChain / deepagents not importable on this worker | ||
| warnings.warn( | ||
| "DeepAgentsPlugin could not patch create_deep_agent; use explicit " | ||
| "TemporalModel(...) instances to route model calls through " | ||
| "activities.", | ||
| stacklevel=2, | ||
| ) |
There was a problem hiding this comment.
Agreed on both points — fixed in a54d445. The handler now catches only ImportError (ModuleNotFoundError is a subclass), matching the documented intent — a worker without the optional langchain/deepagents deps must still start — while genuine patch-installation bugs (e.g. an upstream rename of deepagents.graph.resolve_model raising AttributeError) now surface at worker startup instead of silently degrading. The warning also includes the underlying exception text.
| langsmith_patched = _install_langsmith_temporal_override() | ||
| try: | ||
| yield | ||
| finally: | ||
| if patched: | ||
| # Import is cached: patched=True implies the import above succeeded. | ||
| from temporalio.contrib.deepagents import _model, _tools | ||
|
|
||
| _model.uninstall_model_patch() | ||
| _tools.uninstall_backend_async_patch() | ||
| if langsmith_patched: | ||
| _uninstall_langsmith_temporal_override() |
There was a problem hiding this comment.
You're right — we verified set_runtime_overrides() replaces a single process-wide slot wholesale, and contrib.langsmith installs its (behaviorally identical) override exactly once behind a module flag, so our reset would have stripped it permanently with no reinstall. Removed the uninstall entirely in a54d445: the override is now installed for the life of the process, which is safe because it defers to LangSmith's default thread hop whenever workflow.in_workflow() is false, so it is inert outside workflows. Rationale documented on _install_langsmith_temporal_override.
| self._opts: dict[str, Any] = dict(activity_options or {}) | ||
| self._opts.setdefault("start_to_close_timeout", timedelta(minutes=1)) | ||
| register_backend(self._ref, inner) |
There was a problem hiding this comment.
Good catch — incorporated in a54d445 via weakref.finalize, with one refinement: a plain pop-by-ref would be unsafe here, because refs are deliberately deterministic per run, so after a cache eviction a replaying workflow re-registers the same ref with a fresh inner backend, and the evicted wrapper's late-firing finalizer would have deleted that live registration. The finalizer therefore captures (ref, inner) — not self — and removes the entry only if the registry still maps the ref to that exact inner object (lock-guarded against the GC-time race). Added tests covering both GC cleanup and the re-registration-survives-stale-finalizer case.
- Await the cancelled heartbeat task so no pending task outlives an activity at event-loop shutdown. - Catch only ImportError when installing the create_deep_agent patches and include the original error in the warning; genuine installation bugs now surface at worker startup. - Keep the LangSmith aio_to_thread override installed for the process lifetime: the override slot is global and resetting it would strip a composed contrib.langsmith plugin's identical override. - Unregister a TemporalBackend's registry entry when the wrapper is garbage-collected, identity-guarded so a replay's re-registration of the same deterministic ref survives the evicted wrapper's cleanup.
| ## Install | ||
|
|
||
| ```bash | ||
| pip install "temporalio[deepagents]" |
| # API keys live on the worker via the model provider, never in workflow | ||
| # inputs or history. The default provider is LangChain's init_chat_model. | ||
| plugin = DeepAgentsPlugin( | ||
| model_activity_options={"start_to_close_timeout": timedelta(minutes=5)}, |
There was a problem hiding this comment.
I think we just call this activity_options in most plugins. Not sure if "model" correctly disambiguates because all of your activities might be calling models too?
There was a problem hiding this comment.
It could be more similar to other plugins if we wrote create_temporal_deep_agent() to wrap create_deep_agent() and pass the activity options there. That would also remove any ambiguity.
| - **Existing activities as tools.** Already have `@activity.defn` functions? | ||
| `activity_as_tool(my_activity, start_to_close_timeout=...)` exposes one to the | ||
| agent without re-declaring it. | ||
| - **Explicit Workflow-vs-Activity choice per tool.** `tool_as_activity(tool, ...)` |
There was a problem hiding this comment.
In the strands integration, we instructed users to manually wrap their strands tools in a Temporal activity. I'm not sure we can do the same thing here, so we may want to standardize strands to also take this approach.
| ## Continue-as-new: what carries and what does not | ||
|
|
||
| Long conversations bloat workflow history. `run_deep_agent(agent, input, | ||
| continue_as_new_after=N)` snapshots state and continues into a fresh run once |
There was a problem hiding this comment.
Can we default to is_continue_as_new_suggested() if continue_as_new_after is None and recommend that in our docs?
| return await Client.connect( | ||
| "localhost:7233", | ||
| plugins=[ | ||
| # ObservabilityPlugin(...), # register observability first |
There was a problem hiding this comment.
I think we've proven before that order doesn't actually matter here like we once thought
| with workflow.unsafe.imports_passed_through(): | ||
| from deepagents import create_deep_agent |
There was a problem hiding this comment.
What would it take to avoid needing to do this?
| - **Streaming.** Set `streaming_topic="..."` to stream chat-completion chunk | ||
| batches out of the model Activity to external subscribers; the aggregated final | ||
| message is returned to the workflow so the durable result is identical to the | ||
| non-streaming path. |
There was a problem hiding this comment.
This bullet point (and a few others) might deserve their own section with a code snippet, which could also be a good practice for agentic SEO. I think most other plugin READMEs have more comprehensive sections in general.
What
temporalio.contrib.deepagents— durable execution for LangChain Deep Agents. Unmodified agent code (create_deep_agent(...).ainvoke(...)) runs inside a workflow withplugins=[DeepAgentsPlugin()]as the only change; the plugin hooks the SDK's native runtime so the agent loop replays deterministically in the workflow while everything effectful becomes a Temporal Activity.deepagents.invoke_model(orinvoke_model_streamingwhen astreaming_topicis configured, publishing chunk batches viacontrib.workflow_streamswhile returning the identical durable result). Workflows ship only a model name; the worker'smodel_providerreconstructs the real model — no credentials cross the boundary.activity_as_toolsurfaces an existing@activity.defnto the agent;tool_as_activityroutes an I/O LangChain tool throughdeepagents.invoke_tool. Pure state-mutating built-ins stay in-workflow.TemporalBackendwraps a real backend (e.g.FilesystemBackend) so the agent's built-in file/shell tools execute asdeepagents.backend_opactivities. It intercepts the full backend protocol — sync and async twins (read/aread, …) plusexecute/aexecute— and round-trips the protocol's result dataclasses (WriteResult/ReadResult/…) across the activity boundary as real typed objects.run_deep_agent(..., continue_as_new_after=N)carries the conversation and the model/tool result cache across continue-as-new; completed calls are not re-run.Command(resume=...)protocol maps onto Temporal Queries/Updates; no custom shim.exclude_unsetto keep payloads small) and an explicit transitivepassthrough_moduleslist.Tests
25 tests under
tests/contrib/deepagents/, all against a real dev server via the sharedenv/clientfixtures, offline (scriptedtesting.mock_model_provider, no API keys):create_deep_agentcode + plugin only; tool-loop routing asserts exact activity counts.FilesystemBackendops undermax_cached_workflows=0(replay by construction) — the built-inwrite_file/read_filetools cross the activity boundary and the write really lands on disk activity-side.ApplicationErrortype (429/400/503 +Retry-Afterhandling), continue-as-new conversation carry, sub-agent durability inheritance, streaming, checkpointer warning semantics, README snippets compile.Gates run locally:
pyright,basedpyright,mypy --namespace-packages --check-untyped-defs,pydocstyle,ruff check --select I,ruff format --check— all clean on the new code.Packaging
New
deepagentsextra (temporalio[deepagents]), with every dependency gated onpython_version >= '3.11'(deepagents' own floor; the SDK's 3.10 floor is unaffected). Test deps added to the dev group with the same markers; the test modules skip cleanly on 3.10.Deliberately deferred
ContextHubBackend/LangSmithSandbox(hosted services), stateful MCP sessions, and sub-agents as child workflows (v1 runs sub-agents in-workflow; they inherit the durable model, so their calls are already activities).Related
Samples: temporalio/samples-python#328 (draft; its
deepagentsdependency group picks up this extra once released).