Add Deep Agents plugin samples (#328)

* Add Deep Agents plugin samples

* Address review feedback on HITL state, conftest guard, and streaming test

- human_in_the_loop: clear the pending-approval prompt on resume so the
  query honors its documented contract, and add an update validator that
  rejects decisions other than approve/reject before they enter history
- tests conftest: replace find_spec with a guarded import so collection
  is skipped when the plugin package exists but its runtime deps do not
- streaming test: replace the fixed sleep-then-cancel drain with a
  condition-based subscriber awaited via wait_for, matching the other
  streaming tests

* Address review feedback: CODEOWNERS, docs, stream draining, stronger tests

- CODEOWNERS: add /deepagents_plugin/ and /tests/deepagents_plugin/ for the
  AI SDK team, matching the sibling AI suites
- Suite README: state the Python >= 3.11 floor in Prerequisites (on 3.10 the
  dependency group silently resolves to nothing)
- streaming/run_workflow.py: drain the subscriber until the full durable
  result has been printed (bounded by a timeout) instead of cancelling it
  immediately and dropping tail chunks
- subagents_test: script the coordinator -> task tool -> researcher ->
  synthesis path so the delegation headline is actually exercised, and assert
  three invoke_model activities in history
- hello_world_test: assert the model call was scheduled as a
  deepagents.invoke_model activity (shared count_scheduled_activities helper)
- pyproject: cap langchain-anthropic at <2 like its group siblings

* Align samples with the plugin's recommended patterns

- Drop the workflow.unsafe.imports_passed_through() guards from all eight
  workflows: the plugin passes the deepagents/LangChain import tree through
  the sandbox itself, and its README highlights bare imports as the intended
  developer experience. hello_world carries a comment explaining why no
  guard is needed. Verified by the full test suite (real sandboxed worker)
  plus an ad-hoc sandbox run of the untested langsmith_tracing workflow.
- continue_as_new: use run_deep_agent's default server-suggested mode (the
  documented recommended mode) instead of a hardcoded event threshold; the
  probe test retains continue_as_new_after=1 as explicit-override coverage.
- react_agent: build the agent with create_temporal_deep_agent and per-agent
  activity_options — the recommended way to scope model-call timeouts —
  replacing the bare TemporalModel construction.
- Extend the history seam assertions to every testable scenario: react_agent
  (get_weather + invoke_tool), filesystem_backend (backend_op >= 2),
  streaming (invoke_model_streaming, no invoke_model), human_in_the_loop
  (invoke_tool after resume).
- HITL README: note that a production loop would re-check __interrupt__
  after each resume.

* Account for the merged-but-unreleased plugin (temporalio/sdk-python#1644)

The plugin is on sdk-python main and ships as the temporalio[deepagents]
extra in the next release (>= 1.31.0); it is not on PyPI yet. Point the
interim install at sdk-python main (the per-plugin overlay path retired
at merge), name the real post-release group entry in the pyproject note,
and drop the plugin-ordering claim from langsmith_tracing — registration
order does not matter. Suite verified against merged main: 8 passed.

* Complete the main merge: adopt setuptools packaging, drop hatch remnants

The previous merge commit was pushed with unresolved conflict markers
in pyproject.toml. Resolve to main's setuptools auto-discovery (which
finds deepagents_plugin without registration) and regenerate the lock.

* Drop the dependency-group comment; the suite README covers the install story

* Apply self-review findings: version floors, guard visibility, test rigor

The review's headline: temporalio 1.31.0 already shipped on PyPI
(2026-07-29) WITHOUT the deepagents extra, so every ">= 1.31.0" claim
in the install story was wrong and the documented group swap would
have resolved to an extra-less release. Floors now say "> 1.31.0 /
the first release that ships the extra", and the group's temporalio
pin rises to >= 1.31.0 to match the repo base pin.

Also: the conftest collection guard now announces itself via
pytest_report_header instead of silently collecting nothing, and its
docstring drops the retired temporalio-contrib-deepagents dist story;
the continue-as-new probe pins the first run's close event to
CONTINUED_AS_NEW (a loop-in-one-run regression previously passed);
the HITL suite covers the validator's invalid-decision rejection and
the reject path (guarded tool never runs as an activity); scenario
READMEs name the Python floor and defer to the suite setup instead of
repeating it; the streaming README describes what the workflow
actually drives (TemporalModel.astream).

Author: DABH

.github/CODEOWNERS

@@ -14,6 +14,7 @@
 
 # The AI SDK team owns the AI integration samples and their tests. We add
 # @temporalio/sdk too, so the SDK team can continue to manage repo-wide concerns.
+/deepagents_plugin/       @temporalio/sdk @temporalio/ai-sdk
 /google_adk_agents/       @temporalio/sdk @temporalio/ai-sdk
 /google_genai/            @temporalio/sdk @temporalio/ai-sdk
 /langfuse_tracing/        @temporalio/sdk @temporalio/ai-sdk
@@ -22,6 +23,7 @@
 /litellm_activity/        @temporalio/sdk @temporalio/ai-sdk
 /openai_agents/           @temporalio/sdk @temporalio/ai-sdk
 /strands_plugin/          @temporalio/sdk @temporalio/ai-sdk
+/tests/deepagents_plugin/ @temporalio/sdk @temporalio/ai-sdk
 /tests/google_adk_agents/ @temporalio/sdk @temporalio/ai-sdk
 /tests/google_genai/      @temporalio/sdk @temporalio/ai-sdk
 /tests/langfuse_tracing/  @temporalio/sdk @temporalio/ai-sdk

README.md

@@ -67,6 +67,7 @@ Some examples require extra dependencies. See each sample's directory for specif
 * [custom_converter](custom_converter) - Use a custom payload converter to handle custom types.
 * [custom_decorator](custom_decorator) - Custom decorator to auto-heartbeat a long-running activity.
 * [custom_metric](custom_metric) - Custom metric to record the workflow type in the activity schedule to start latency.
+* [deepagents_plugin](deepagents_plugin) - Make LangChain Deep Agents durable: each LLM/tool/backend call becomes a Temporal Activity while the agent loop replays in the Workflow.
 * [dsl](dsl) - DSL workflow that executes steps defined in a YAML file.
 * [eager_wf_start](eager_wf_start) - Run a workflow using Eager Workflow Start
 * [encryption](encryption) - Apply end-to-end encryption for all input/output.

deepagents_plugin/README.md

@@ -0,0 +1,134 @@
+# Deep Agents Samples
+
+These samples demonstrate the [Temporal Deep Agents plugin](https://github.com/temporalio/sdk-python/tree/main/temporalio/contrib/deepagents),
+which makes [LangChain Deep Agents](https://github.com/langchain-ai/deepagents)
+durable. Build your agent with `create_deep_agent(...)` inside a
+`@workflow.defn` and add `DeepAgentsPlugin()` to your client — each LLM call and
+each I/O tool/backend operation becomes a Temporal Activity, while the agent's
+control loop runs (and deterministically replays) inside the Workflow.
+
+> **Experimental.** The `temporalio.contrib.deepagents` plugin is experimental
+> and its API may change.
+
+`DeepAgentsPlugin` is a **client-level** plugin: add it to `Client.connect(...)`
+and the SDK propagates it to any Worker built from that client. Add it on exactly
+one side.
+
+## Samples
+
+| Sample | Description |
+|--------|-------------|
+| [hello_world](hello_world) | Minimal single-shot Deep Agent; a bare `model=` string auto-routed through the model activity. Start here. |
+| [react_agent](react_agent) | Tool-calling loop showing the explicit per-tool choice: `activity_as_tool` for an existing activity, `tool_as_activity` for an I/O tool, plus per-agent `activity_options` via `create_temporal_deep_agent`. |
+| [human_in_the_loop](human_in_the_loop) | Pause on `interrupt_on` and resume via the native LangGraph protocol, mapped to a Temporal Query + Update. |
+| [continue_as_new](continue_as_new) | Long-running agent that carries messages and the model/tool result cache across continue-as-new via `run_deep_agent`. |
+| [filesystem_backend](filesystem_backend) | Durable real filesystem I/O by wrapping a `FilesystemBackend` in `TemporalBackend`. |
+| [subagents](subagents) | Durability propagates across the agent tree — sub-agent model calls become activities with no per-sub-agent wiring. |
+| [streaming](streaming) | Stream model chunks to external subscribers via `streaming_topic` + `WorkflowStream`, keeping the durable result identical. |
+| [langsmith_tracing](langsmith_tracing) | Compose `DeepAgentsPlugin` with `LangSmithPlugin` for durable execution + LLM tracing. |
+
+## Prerequisites
+
+> **Python ≥ 3.11 required.** `deepagents` (and therefore the plugin) does not
+> support older interpreters. On Python 3.10 the `deepagents` dependency group
+> resolves to nothing, so `uv sync` silently installs none of the dependencies
+> below.
+
+1. Install dependencies:
+
+   ```bash
+   uv sync --group deepagents
+   ```
+
+   > The Deep Agents plugin ships as the `temporalio[deepagents]` extra. It
+   > is merged to `sdk-python` `main` but the current PyPI release (1.31.0)
+   > predates the merge and does not carry the extra, so the `deepagents`
+   > group above does not include it yet. Until a release ships the extra
+   > (> 1.31.0), install it from main:
+   >
+   > ```bash
+   > uv pip install "temporalio[deepagents] @ git+https://github.com/temporalio/sdk-python.git"
+   > ```
+   >
+   > This builds the SDK from source (including its Rust core), so expect a
+   > few minutes on first install. Once a release with the extra is on PyPI
+   > this step goes away: `temporalio[deepagents]` joins the `deepagents`
+   > group and a plain `uv sync --group deepagents` is all you need.
+
+2. Configure a model provider. The samples use
+   `anthropic:claude-sonnet-4-5`, which needs an Anthropic API key:
+
+   ```bash
+   export ANTHROPIC_API_KEY=...
+   ```
+
+   To use a different provider, change the `model=` string in the sample's
+   `workflow.py` and set that provider's credentials (the plugin resolves the
+   model worker-side via LangChain's `init_chat_model`).
+
+3. Start a [Temporal dev server](https://docs.temporal.io/cli#start-dev-server):
+
+   ```bash
+   temporal server start-dev
+   ```
+
+## Running a Sample
+
+> **Use `uv run --no-sync`.** Because the plugin is installed out-of-band
+> from sdk-python main (see Prerequisites) and is not yet in any dependency
+> group, a bare `uv run` or `uv sync` re-syncs the environment to the lockfile
+> first and uninstalls it. `--no-sync` runs against the environment as-is.
+> (Once a released `temporalio[deepagents]` joins the `deepagents` group, the
+> flag becomes unnecessary.)
+
+Most samples have two scripts. Start the Worker first, then the Workflow starter
+in a separate terminal:
+
+```bash
+# Terminal 1: start the Worker
+uv run --no-sync deepagents_plugin/<sample>/run_worker.py
+
+# Terminal 2: start the Workflow
+uv run --no-sync deepagents_plugin/<sample>/run_workflow.py
+```
+
+For example, to run the hello world sample:
+
+```bash
+# Terminal 1
+uv run --no-sync deepagents_plugin/hello_world/run_worker.py
+
+# Terminal 2
+uv run --no-sync deepagents_plugin/hello_world/run_workflow.py
+```
+
+The `langsmith_tracing` sample instead bundles the worker and starter into a
+single driver:
+
+```bash
+uv run --no-sync deepagents_plugin/langsmith_tracing/main.py
+```
+
+## Key Features Demonstrated
+
+- **Durable model invocation** — every LLM call runs in an `invoke_model`
+  activity with configurable timeouts and retries; a bare `model=` string is
+  auto-routed, or use `create_temporal_deep_agent(..., activity_options=...)`
+  to scope model-call options per agent (recommended).
+- **Explicit Workflow-vs-Activity tool choice** — `activity_as_tool`,
+  `tool_as_activity`, and `TemporalBackend` move I/O out of workflow code.
+- **Human-in-the-loop** — the native LangGraph `interrupt_on` return value
+  mapped to a Temporal Query and Update.
+- **Long-lived agents** — `run_deep_agent(...)` carries messages and the result
+  cache across server-suggested (or explicitly thresholded) continue-as-new.
+- **Sub-agent durability** — sub-agents inherit the durable model object with no
+  extra wiring.
+- **Streaming** — forward model chunks to external subscribers while keeping the
+  durable result unchanged.
+- **Observability** — compose with `LangSmithPlugin` for tracing.
+
+## Related
+
+- [Temporal Deep Agents plugin](https://github.com/temporalio/sdk-python/tree/main/temporalio/contrib/deepagents)
+- [LangChain Deep Agents](https://github.com/langchain-ai/deepagents)
+- [langgraph_plugin](../langgraph_plugin) — for agents built directly as LangGraph graphs

deepagents_plugin/__init__.py

@@ -0,0 +1 @@
+"""Temporal Deep Agents plugin samples."""

deepagents_plugin/continue_as_new/README.md

@@ -0,0 +1,53 @@
+# Continue as New
+
+A long-running research agent whose conversation could outgrow Temporal's
+workflow-history limit. `run_deep_agent(agent, input, state_snapshot=...)` keeps
+the run bounded: once a turn ends with pending todos and the server recommends
+continuing (`workflow.info().is_continue_as_new_suggested()` — the default and
+recommended mode, which accounts for both history length and size), it snapshots
+the accumulated messages **and** the model/tool result cache and continues into
+a fresh run. Completed model/tool calls are reused from the carried cache rather
+than re-run. To trigger on a fixed history-event count instead, pass an explicit
+`continue_as_new_after=N`.
+
+The `@workflow.run` signature is `run(self, input, state_snapshot=None)`, where
+`input` is the messages mapping and `state_snapshot` is how `run_deep_agent`
+threads carried state into the continued run. On a continue-as-new the workflow
+is re-invoked with `args=[input, snapshot]`, so `input` must be passed straight
+into `run_deep_agent` — re-wrapping it would nest a dict where a message is
+expected and corrupt the carried conversation. Durability rides on the default
+in-workflow `InMemorySaver` (rehydrated by replay); a database-backed
+checkpointer would do I/O from workflow code and is not replay-safe.
+
+## What This Sample Demonstrates
+
+- `run_deep_agent(agent, input, state_snapshot=...)` in the default
+  server-suggested mode (with `continue_as_new_after=N` as the explicit override)
+- The `run(self, input, state_snapshot=None)` continue-as-new contract
+- Carrying both messages and the result cache across continue-as-new
+
+## Running the Sample
+
+Prerequisites: Python >= 3.11 with the [suite setup](../README.md#prerequisites)
+applied (interim plugin install), an `ANTHROPIC_API_KEY` in your
+environment, and a running Temporal dev server (`temporal server start-dev`).
+
+> The experimental plugin is not in the `deepagents` group — install it as shown
+> in the [suite README](../README.md#prerequisites) and run with `--no-sync`, or
+> a bare `uv run`/`uv sync` re-syncs the environment and uninstalls it.
+
+```bash
+# Terminal 1
+uv run --no-sync deepagents_plugin/continue_as_new/run_worker.py
+
+# Terminal 2
+uv run --no-sync deepagents_plugin/continue_as_new/run_workflow.py
+```
+
+## Files
+
+| File | Description |
+|------|-------------|
+| `workflow.py` | `LongResearchAgent` driven by `run_deep_agent` |
+| `run_worker.py` | Adds `DeepAgentsPlugin`, starts the worker |
+| `run_workflow.py` | Executes the workflow and prints the result |

deepagents_plugin/continue_as_new/__init__.py

@@ -0,0 +1,29 @@
+"""Worker for the continue-as-new sample."""
+
+import asyncio
+import os
+
+from temporalio.client import Client
+from temporalio.contrib.deepagents import DeepAgentsPlugin
+from temporalio.worker import Worker
+
+from deepagents_plugin.continue_as_new.workflow import LongResearchAgent
+
+
+async def main() -> None:
+    client = await Client.connect(
+        os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"),
+        plugins=[DeepAgentsPlugin()],
+    )
+
+    worker = Worker(
+        client,
+        task_queue="deepagents-continue-as-new",
+        workflows=[LongResearchAgent],
+    )
+    print("Worker started. Ctrl+C to exit.")
+    await worker.run()
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

deepagents_plugin/continue_as_new/run_worker.py

@@ -0,0 +1,37 @@
+"""Start the long-running research agent workflow."""
+
+import asyncio
+import os
+
+from temporalio.client import Client
+
+from deepagents_plugin.continue_as_new.workflow import LongResearchAgent
+
+
+async def main() -> None:
+    client = await Client.connect(os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"))
+
+    result = await client.execute_workflow(
+        LongResearchAgent.run,
+        # The workflow's first arg is the messages mapping (run_deep_agent's
+        # continue-as-new contract), not a bare question string.
+        {
+            "messages": [
+                {
+                    "role": "user",
+                    "content": (
+                        "Research the tradeoffs between Raft and Paxos and "
+                        "summarize them."
+                    ),
+                }
+            ]
+        },
+        id="deepagents-continue-as-new",
+        task_queue="deepagents-continue-as-new",
+    )
+
+    print(f"Result: {result}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

deepagents_plugin/continue_as_new/run_workflow.py

@@ -0,0 +1,59 @@
+"""Long-running research agent that carries state across continue-as-new.
+
+A long conversation would bloat workflow history until it hits Temporal's limit.
+``run_deep_agent(agent, input, state_snapshot=...)`` solves this: once a turn
+finishes with pending work and the server recommends continuing
+(``workflow.info().is_continue_as_new_suggested()`` — the default and
+recommended mode, accounting for both history length and size), it snapshots the
+accumulated messages **and** the model/tool result cache and continues into a
+fresh run — so completed model/tool calls are reused, not re-run, after the
+continue-as-new. Pass ``continue_as_new_after=N`` instead to trigger on a fixed
+history-event count.
+
+The contract ``run_deep_agent`` requires is that the ``@workflow.run`` method
+accepts the carried state, i.e. its signature is
+``run(self, input, state_snapshot=None)`` where ``input`` is the messages
+mapping. On a continue-as-new, ``run_deep_agent`` re-invokes the workflow with
+``args=[input, snapshot]``, so ``input`` must be passed straight through — not
+re-wrapped — or the carried conversation is corrupted. Only an in-workflow
+``InMemorySaver`` (the default) is replay-safe; a durable checkpointer would do
+I/O from workflow code.
+"""
+
+# @@@SNIPSTART python-deepagents-continue-as-new-workflow
+from typing import Any
+
+from deepagents import create_deep_agent
+from temporalio import workflow
+from temporalio.contrib.deepagents import run_deep_agent
+
+
+@workflow.defn
+class LongResearchAgent:
+    @workflow.run
+    async def run(
+        self, input: dict[str, Any], state_snapshot: dict | None = None
+    ) -> str:
+        agent = create_deep_agent(
+            model="anthropic:claude-sonnet-4-5",
+            system_prompt=(
+                "You are a research agent. Break large tasks into todos and work "
+                "through them until the research is complete."
+            ),
+        )
+        result = await run_deep_agent(
+            agent,
+            # ``input`` is the messages mapping. Pass it through unchanged: on a
+            # continue-as-new, run_deep_agent re-invokes this method with the
+            # carried input as its first arg, so re-wrapping it here would nest a
+            # dict where a message is expected and corrupt the conversation.
+            input,
+            # No threshold: continue-as-new fires when the agent still has
+            # pending todos and the server suggests continuing — the recommended
+            # mode. Pass continue_as_new_after=N to use a fixed event count.
+            state_snapshot=state_snapshot,
+        )
+        return result["messages"][-1].content
+
+
+# @@@SNIPEND

deepagents_plugin/continue_as_new/workflow.py

@@ -0,0 +1,50 @@
+# Filesystem Backend
+
+Give a Deep Agent durable, real filesystem access. The agent's built-in file
+tools (`write_file`, `read_file`, `ls`, …) delegate to a *backend*. Wrapping a
+`FilesystemBackend` with `TemporalBackend(inner, activity_options=...)` routes
+each file operation through a `deepagents.backend_op` activity, so real disk I/O
+happens in an activity worker instead of in workflow code.
+
+Contrast this with the default `StateBackend`, whose "files" live in agent state
+— that is pure workflow state and correctly stays in the workflow with no
+wrapping. `TemporalBackend` is only for backends that do real I/O.
+
+The scratch directory (`root_dir`) is chosen client-side and passed in as a
+workflow argument, so the workflow never reads the environment or the disk
+directly.
+
+## What This Sample Demonstrates
+
+- `TemporalBackend` wrapping a real-I/O `FilesystemBackend`
+- The agent's built-in file tools running their I/O as `backend_op` activities
+- Keeping the workflow deterministic by passing `root_dir` in as an argument
+
+## Running the Sample
+
+Prerequisites: Python >= 3.11 with the [suite setup](../README.md#prerequisites)
+applied (interim plugin install), an `ANTHROPIC_API_KEY` in your
+environment, and a running Temporal dev server (`temporal server start-dev`).
+
+> The experimental plugin is not in the `deepagents` group — install it as shown
+> in the [suite README](../README.md#prerequisites) and run with `--no-sync`, or
+> a bare `uv run`/`uv sync` re-syncs the environment and uninstalls it.
+
+```bash
+# Terminal 1
+uv run --no-sync deepagents_plugin/filesystem_backend/run_worker.py
+
+# Terminal 2
+uv run --no-sync deepagents_plugin/filesystem_backend/run_workflow.py
+```
+
+By default the starter creates a temporary scratch directory; set
+`DEEPAGENTS_WORKDIR` to point the agent at a directory of your choice.
+
+## Files
+
+| File | Description |
+|------|-------------|
+| `workflow.py` | `FilesystemAgent` wrapping a `FilesystemBackend` in `TemporalBackend` |
+| `run_worker.py` | Adds `DeepAgentsPlugin`, starts the worker |
+| `run_workflow.py` | Chooses a scratch dir, executes the workflow, prints the result |