Add Deep Agents external storage sample

Author: brianstrauch

deepagents_plugin/README.md

@@ -26,6 +26,7 @@ one side.
 | [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. |
+| [external_storage](external_storage) | Carry a 6 MiB tool result and the next model input through native S3 External Storage; scripted model, no API keys. |
 
 ## Prerequisites
 
@@ -45,8 +46,10 @@ one side.
    > Python >= 3.11 — on older interpreters the group resolves to nothing
    > and the samples are skipped.
 
-2. Configure a model provider. The samples use
-   `anthropic:claude-sonnet-4-5`, which needs an Anthropic API key:
+2. Configure a model provider. Most samples use
+   `anthropic:claude-sonnet-4-5`, which needs an Anthropic API key. The
+   [external_storage](external_storage) sample uses a scripted model and needs
+   no provider credentials:
 
    ```bash
    export ANTHROPIC_API_KEY=...
@@ -85,11 +88,13 @@ uv run deepagents_plugin/hello_world/run_worker.py
 uv run deepagents_plugin/hello_world/run_workflow.py
 ```
 
-The `langsmith_tracing` sample instead bundles the worker and starter into a
-single driver:
+The `langsmith_tracing` and `external_storage` samples instead bundle the worker
+and starter into a single driver. The storage sample also needs a mock S3 service;
+see its [setup instructions](external_storage):
 
 ```bash
 uv run deepagents_plugin/langsmith_tracing/main.py
+uv run deepagents_plugin/external_storage/main.py
 ```
 
 ## Key Features Demonstrated
@@ -109,6 +114,8 @@ uv run deepagents_plugin/langsmith_tracing/main.py
 - **Streaming** — forward model chunks to external subscribers while keeping the
   durable result unchanged.
 - **Observability** — compose with `LangSmithPlugin` for tracing.
+- **Large payload storage** — compose native `ExternalStorage` with the plugin's
+  data converter, keeping full tool results in S3 and small references in history.
 
 ## Related
 

deepagents_plugin/external_storage/README.md

@@ -0,0 +1,121 @@
+# External Storage
+
+Run a Deep Agent whose `read_file` tool returns **6 MiB** of text, exceeding
+Temporal's default 2 MiB payload limit. The SDK's native `ExternalStorage`
+offloads the tool output and the next model activity's input to S3, recording
+small references in workflow history and retrieving the full bytes on decode.
+The workflow returns only a byte count, SHA-256, and a short final answer.
+
+Native External Storage is in **Public Preview**. This sample uses the SDK's
+`S3StorageDriver` with the local mock S3 service from the
+[general External Storage sample](../../external_storage). Neither AWS credentials,
+Docker, nor an LLM API key is needed. The model replies are scripted; the real
+Deep Agents loop, tool activity, model activities, and S3 transfers still run.
+
+## Running the sample
+
+Use Python **3.11 or later**. Run these commands from the repository root:
+
+```bash
+uv sync --python 3.13 --group deepagents --group external-storage
+```
+
+Start Temporal with its default payload limits in one terminal:
+
+```bash
+temporal server start-dev
+```
+
+In a second terminal, start the existing mock S3 service. It listens on port
+5000 and creates the `temporal-payloads` bucket:
+
+```bash
+uv run external_storage/s3.py
+```
+
+In a third terminal, run the demo. It starts a worker, executes one workflow,
+checks the result's integrity, and shuts down the worker:
+
+```bash
+uv run deepagents_plugin/external_storage/main.py
+```
+
+The output includes:
+
+```text
+Tool result: 6,291,456 bytes (6 MiB), integrity verified
+Agent: Received the complete document.
+```
+
+The script also prints the workflow ID and complete SHA-256. Inspect the history
+in the Temporal UI or with:
+
+```bash
+temporal workflow show --workflow-id <printed-workflow-id>
+```
+
+The completed `deepagents.invoke_tool` activity and the following
+`deepagents.invoke_model` input contain native storage references. The first
+model input and final workflow result remain small and inline.
+
+To view externally stored payloads in the UI, reuse the
+[storage-aware codec server](../../external_storage#5-optional-run-the-codec-server):
+run `uv run external_storage/codec_server.py` and set the UI's Remote Codec
+Endpoint to `http://localhost:8081`. Its gzip decoder passes uncompressed payloads
+through, so it can also read this sample's references.
+
+## How the configuration works
+
+`client.py` creates `ExternalStorage(drivers=[driver],
+payload_size_threshold=256 * 1024)`. It uses the SDK's `SimplePlugin` converter
+hook to apply this configuration **after** `DeepAgentsPlugin` installs its
+LangChain-aware payload converter:
+
+```python
+def configure(converter: DataConverter | None) -> DataConverter:
+    return replace(converter or DataConverter.default, external_storage=storage)
+
+client = await Client.connect(
+    "localhost:7233",
+    plugins=[
+        create_plugin(),
+        SimplePlugin("ExternalStorage", data_converter=configure),
+    ],
+)
+```
+
+This order also supports the older plugin version in this repository's lockfile,
+which otherwise replaces the client's converter. The worker inherits the final
+converter. The S3 client stays open while the worker runs and the starter decodes
+results. No application code uploads or downloads payloads manually.
+
+This demo deliberately uses **no compression codec**: repeated `x` characters
+would compress well below the storage threshold. It overrides the built-in
+`read_file` tool with an activity-backed mock bulk reader. Deep Agents excludes
+`read_file` from tool-result eviction, so the complete result crosses both
+activity boundaries. Applications using real models should page documents or
+retrieve excerpts to manage context separately. Storage reduces transport and
+history bytes; it does not reduce model tokens or decoded workflow memory.
+
+With storage removed, this tool result fails with `PayloadsTooLarge` at the
+default limits. Raising the blob limit alone still leaves the gRPC message
+limit described in [Hanyu Liu's post](https://hanyuliu.me/posts/temporal-deep-agent-payload-limits/).
+Native storage avoids transmitting the large bytes through either limit.
+
+For real S3, use an existing bucket and normal AWS credentials instead of this
+sample's local endpoint and mock credentials. Configure compatible storage
+drivers on every client, worker, and replayer that reads the history. Retain
+objects for as long as execution, reset, or replay may need them; Temporal does
+not delete stored payloads when an activity finishes.
+
+## Tests
+
+The integration test starts an isolated mock S3 service and verifies the full
+result, both native references, activity routing, and replay with a new S3
+client after worker shutdown.
+No manually running S3 service or model credentials are required:
+
+```bash
+uv sync --python 3.13 --group deepagents --group external-storage --group dev
+uv run pytest tests/deepagents_plugin/external_storage_test.py
+```

deepagents_plugin/external_storage/client.py

@@ -0,0 +1,64 @@
+"""Compose native S3 storage with the Deep Agents data converter."""
+
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+from dataclasses import replace
+
+import aioboto3
+from temporalio.client import Client
+from temporalio.contrib.aws.s3driver import S3StorageDriver
+from temporalio.contrib.aws.s3driver.aioboto3 import new_aioboto3_client
+from temporalio.contrib.deepagents import DeepAgentsPlugin
+from temporalio.converter import DataConverter, ExternalStorage
+from temporalio.envconfig import ClientConfig
+from temporalio.plugin import SimplePlugin
+
+S3_ENDPOINT = "http://localhost:5000"
+S3_BUCKET = "temporal-payloads"
+
+
+# @@@SNIPSTART python-deepagents-external-storage-converter
+def storage_plugin(storage: ExternalStorage) -> SimplePlugin:
+    """Add storage after the agent plugin has installed its payload converter."""
+
+    def configure(converter: DataConverter | None) -> DataConverter:
+        return replace(converter or DataConverter.default, external_storage=storage)
+
+    # Applying this after DeepAgentsPlugin also supports older plugin versions
+    # that replace the converter rather than composing with the client's one.
+    return SimplePlugin("ExternalStorage", data_converter=configure)
+
+
+@asynccontextmanager
+async def connect_client(
+    plugin: DeepAgentsPlugin,
+    *,
+    target_host: str | None = None,
+    s3_endpoint: str = S3_ENDPOINT,
+) -> AsyncIterator[Client]:
+    """Keep the S3 client alive throughout workflow execution and payload reads."""
+    config = ClientConfig.load_client_connect_config()
+    config.setdefault("target_host", "localhost:7233")
+    if target_host is not None:
+        config["target_host"] = target_host
+
+    session = aioboto3.Session()
+    async with session.client(
+        "s3",
+        endpoint_url=s3_endpoint,
+        aws_access_key_id="test",
+        aws_secret_access_key="test",
+        region_name="us-east-1",
+    ) as s3_client:
+        driver = S3StorageDriver(
+            client=new_aioboto3_client(s3_client),
+            bucket=S3_BUCKET,
+        )
+        storage = ExternalStorage(drivers=[driver], payload_size_threshold=256 * 1024)
+        yield await Client.connect(
+            **config,
+            plugins=[plugin, storage_plugin(storage)],
+        )
+
+
+# @@@SNIPEND

deepagents_plugin/external_storage/main.py

@@ -0,0 +1,71 @@
+"""Run one Deep Agent with a 6 MiB tool result and native S3 storage.
+
+Both the worker and starter run in this process. The two model replies are
+scripted, so the demo needs no LLM credentials or provider context budget.
+The model and tool still run as real Temporal activities.
+"""
+
+import asyncio
+import hashlib
+import uuid
+
+from langchain_core.messages import AIMessage
+from temporalio.contrib.deepagents import DeepAgentsPlugin
+from temporalio.contrib.deepagents.testing import mock_model_provider
+from temporalio.worker import Worker
+
+from deepagents_plugin.external_storage.client import connect_client
+from deepagents_plugin.external_storage.workflow import (
+    PAYLOAD_BYTES,
+    TASK_QUEUE,
+    ExternalStorageAgent,
+)
+
+
+def create_plugin() -> DeepAgentsPlugin:
+    """Script a tool request and acknowledgment for this single-workflow demo."""
+    return DeepAgentsPlugin(
+        model_provider=mock_model_provider(
+            [
+                AIMessage(
+                    content="",
+                    tool_calls=[
+                        {
+                            "name": "read_file",
+                            "args": {"file_path": "large-document.txt"},
+                            "id": "call-read",
+                        }
+                    ],
+                ),
+                AIMessage(content="Received the complete document."),
+            ]
+        )
+    )
+
+
+async def main() -> None:
+    async with connect_client(create_plugin()) as client:
+        async with Worker(
+            client,
+            task_queue=TASK_QUEUE,
+            workflows=[ExternalStorageAgent],
+            max_cached_workflows=0,
+        ):
+            workflow_id = f"deepagents-external-storage-{uuid.uuid4()}"
+            result = await client.execute_workflow(
+                ExternalStorageAgent.run,
+                "Read large-document.txt and acknowledge receiving it.",
+                id=workflow_id,
+                task_queue=TASK_QUEUE,
+            )
+
+        assert result.result_bytes == PAYLOAD_BYTES
+        assert result.sha256 == hashlib.sha256(b"x" * PAYLOAD_BYTES).hexdigest()
+        print(f"Workflow: {workflow_id}")
+        print(f"Tool result: {result.result_bytes:,} bytes (6 MiB), integrity verified")
+        print(f"SHA-256: {result.sha256}")
+        print(f"Agent: {result.answer}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

deepagents_plugin/external_storage/workflow.py

@@ -0,0 +1,75 @@
+"""A Deep Agent whose tool returns 6 MiB, carried by native External Storage."""
+
+import hashlib
+from dataclasses import dataclass
+from datetime import timedelta
+
+from langchain_core.messages import ToolMessage
+from langchain_core.tools import tool
+from temporalio import workflow
+from temporalio.common import RetryPolicy
+from temporalio.contrib.deepagents import (
+    create_temporal_deep_agent,
+    tool_as_activity,
+)
+
+PAYLOAD_BYTES = 6 * 1024 * 1024
+MODEL = "fake:external-storage"
+TASK_QUEUE = "deepagents-external-storage"
+
+
+@dataclass
+class AgentResult:
+    """Small integrity evidence; the large document stays out of the final result."""
+
+    result_bytes: int
+    sha256: str
+    answer: str
+
+
+@tool
+async def read_file(file_path: str) -> str:
+    """Read a mock document containing exactly 6 MiB of text."""
+    # Override the built-in read_file with an activity-backed bulk reader.
+    # Deep Agents excludes read_file from tool-result eviction, so its complete
+    # content reaches the next model activity. Real file I/O would go here.
+    del file_path
+    return "x" * PAYLOAD_BYTES
+
+
+# @@@SNIPSTART python-deepagents-external-storage-workflow
+@workflow.defn
+class ExternalStorageAgent:
+    @workflow.run
+    async def run(self, question: str) -> AgentResult:
+        reader = tool_as_activity(
+            read_file,
+            start_to_close_timeout=timedelta(seconds=30),
+            activity_options={"retry_policy": RetryPolicy(maximum_attempts=1)},
+        )
+        agent = create_temporal_deep_agent(
+            model=MODEL,
+            tools=[reader],
+            system_prompt="Read the document, then acknowledge receiving the result.",
+            activity_options={"start_to_close_timeout": timedelta(seconds=30)},
+        )
+        result = await agent.ainvoke(
+            {"messages": [{"role": "user", "content": question}]}
+        )
+        tool_result = next(
+            message
+            for message in result["messages"]
+            if isinstance(message, ToolMessage) and message.name == "read_file"
+        )
+        assert isinstance(tool_result.content, str)
+        data = tool_result.content.encode()
+        answer = result["messages"][-1].content
+        assert isinstance(answer, str)
+        return AgentResult(
+            result_bytes=len(data),
+            sha256=hashlib.sha256(data).hexdigest(),
+            answer=answer,
+        )
+
+
+# @@@SNIPEND