Add Deep Agents plugin samples

README.md

@@ -65,6 +65,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,126 @@
+# 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 an explicit `TemporalModel`. |
+| [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
+
+1. Install dependencies:
+
+   ```bash
+   uv sync --group deepagents
+   ```
+
+   > The `temporalio-contrib-deepagents` plugin is experimental and not yet
+   > published to PyPI, so it is not part of the `deepagents` group above.
+   > Until it publishes, install it from a local checkout of the SDK (adjust
+   > the path to wherever your `sdk-python` checkout lives):
+   >
+   > ```bash
+   > uv pip install ../sdk-python/temporalio/contrib/deepagents
+   > ```
+   >
+   > The plugin uses a namespace-package overlay layout, so install it
+   > non-editable (no `-e`) — an editable install cannot map its sources onto
+   > `temporalio.contrib.deepagents`. Once it is on PyPI this step goes away
+   > and you can add `temporalio-contrib-deepagents` to the `deepagents` group.
+
+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 experimental plugin is installed
+> out-of-band (`uv pip install …` above) and is not 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 the
+> plugin publishes and joins the `deepagents` group, the flag is 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 `TemporalModel(...)` explicitly.
+- **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(continue_as_new_after=...)` carries
+  messages and the result cache across 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,48 @@
+# Continue as New
+
+A long-running research agent whose conversation could outgrow Temporal's
+workflow-history limit. `run_deep_agent(..., continue_as_new_after=N)` keeps the
+run bounded: once history passes `N` events and the agent still has pending
+todos, 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.
+
+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, continue_as_new_after=..., state_snapshot=...)`
+- 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: `uv sync --group deepagents`, 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` with `continue_as_new_after` |
+| `run_worker.py` | Adds `DeepAgentsPlugin`, starts the worker |
+| `run_workflow.py` | Executes the workflow and prints the result |

deepagents_plugin/continue_as_new/run_worker.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_workflow.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/workflow.py

@@ -0,0 +1,58 @@
+"""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, continue_as_new_after=N, state_snapshot=...)``
+solves this: once the current turn finishes past the ``N``-event threshold and
+there is still pending work, 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.
+
+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 temporalio import workflow
+
+with workflow.unsafe.imports_passed_through():
+    from deepagents import create_deep_agent
+    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,
+            # Continue-as-new once history passes this many events and the agent
+            # still has pending todos. Tune to your model's turn size.
+            continue_as_new_after=10_000,
+            state_snapshot=state_snapshot,
+        )
+        return result["messages"][-1].content
+
+
+# @@@SNIPEND

deepagents_plugin/filesystem_backend/README.md

@@ -0,0 +1,49 @@
+# 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: `uv sync --group deepagents`, 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 |

deepagents_plugin/filesystem_backend/run_worker.py

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