Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions .github/CODEOWNERS
Original file line number Diff line number Diff line change
Expand Up @@ -14,12 +14,14 @@

# 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
/langfuse_tracing/ @temporalio/sdk @temporalio/ai-sdk
/langgraph_plugin/ @temporalio/sdk @temporalio/ai-sdk
/langsmith_tracing/ @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/langfuse_tracing/ @temporalio/sdk @temporalio/ai-sdk
/tests/langgraph_plugin/ @temporalio/sdk @temporalio/ai-sdk
Expand Down
1 change: 1 addition & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
132 changes: 132 additions & 0 deletions deepagents_plugin/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,132 @@
# 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 `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 `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
1 change: 1 addition & 0 deletions deepagents_plugin/__init__.py
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@
"""Temporal Deep Agents plugin samples."""
52 changes: 52 additions & 0 deletions deepagents_plugin/continue_as_new/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,52 @@
# 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: `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` |
| `run_worker.py` | Adds `DeepAgentsPlugin`, starts the worker |
| `run_workflow.py` | Executes the workflow and prints the result |
Empty file.
29 changes: 29 additions & 0 deletions deepagents_plugin/continue_as_new/run_worker.py
Original file line number Diff line number Diff line change
@@ -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())
37 changes: 37 additions & 0 deletions deepagents_plugin/continue_as_new/run_workflow.py
Original file line number Diff line number Diff line change
@@ -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())
59 changes: 59 additions & 0 deletions deepagents_plugin/continue_as_new/workflow.py
Original file line number Diff line number Diff line change
@@ -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
49 changes: 49 additions & 0 deletions deepagents_plugin/filesystem_backend/README.md
Original file line number Diff line number Diff line change
@@ -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 |
Empty file.
Loading
Loading