Add Deep Agents external storage sample

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) | Native External Storage for a 6 MiB tool result; scripted model, no API keys. |
 
 ## Prerequisites
 
@@ -45,8 +46,10 @@ one side.
    > Temporal 1.34.0 or later and 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** — configure native `ExternalStorage` on the client's
+  data converter to offload a 6 MiB tool result and the following model input.
 
 ## Related
 

deepagents_plugin/external_storage/README.md

@@ -0,0 +1,63 @@
+# External Storage
+
+This sample runs a Deep Agent whose activity-backed `read_file` tool returns a
+**6 MiB** document. Temporal's native `ExternalStorage` stores the tool result
+and the following model activity input in S3, leaving small references in
+workflow history. The workflow returns only the document's byte count, SHA-256,
+and final answer.
+
+The client configures `ExternalStorage` on a data converter and passes it to
+`DeepAgentsPlugin`, which composes it with its LangChain-aware converter. The
+worker inherits that configuration. Payloads above the default **256 KiB**
+threshold are uploaded and downloaded automatically; the workflow input, first
+model input, and final result stay inline. The agent loop runs in the workflow,
+and each model and tool call runs as its own plugin activity.
+
+The S3 driver uses the local mock service from the
+[general External Storage sample](../../external_storage). No AWS credentials,
+Docker, or LLM API key is needed. The model replies are scripted; the real Deep
+Agents loop, tool activity, model activities, and S3 transfers still run.
+
+## Run it
+
+Use Python **3.11 or later**:
+
+```bash
+uv sync --python 3.13 --group deepagents --group external-storage
+```
+
+Start Temporal in one terminal:
+
+```bash
+temporal server start-dev
+```
+
+Start the mock S3 service in another terminal. It listens on port 5000 and
+creates the `temporal-payloads` bucket:
+
+```bash
+uv run external_storage/s3.py
+```
+
+Run the demo:
+
+```bash
+uv run deepagents_plugin/external_storage/main.py
+```
+
+The output includes the workflow ID, verified document size and SHA-256, and
+final answer. Inspect the workflow history with:
+
+```bash
+temporal workflow show --workflow-id <printed-workflow-id>
+```
+
+Activity inputs and results above the threshold appear as native storage
+references. To view externally stored payloads in the UI, run the
+[storage-aware codec server](../../external_storage#5-optional-run-the-codec-server)
+and set the UI's Remote Codec Endpoint to `http://localhost:8081`.
+
+For real S3, use an existing bucket and normal AWS credentials. Configure
+compatible storage drivers on every client, worker, and replayer that reads the
+history. Keep payload objects for as long as execution, reset, or replay may
+need them. The local mock S3 service is in memory.

deepagents_plugin/external_storage/client.py

@@ -0,0 +1,66 @@
+"""Configure native S3 External Storage for the Deep Agents plugin."""
+
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+from dataclasses import replace
+
+import aioboto3
+from langchain_core.messages import AIMessage
+from temporalio.client import Client
+from temporalio.contrib.aws.s3driver import S3StorageDriver
+from temporalio.contrib.aws.s3driver.aioboto3 import new_aioboto3_client
+from temporalio.converter import DataConverter, ExternalStorage
+from temporalio.deepagents import DeepAgentsPlugin
+from temporalio.deepagents.testing import mock_model_provider
+from temporalio.envconfig import ClientConfig
+
+S3_ENDPOINT = "http://localhost:5000"
+S3_BUCKET = "temporal-payloads"
+
+
+@asynccontextmanager
+async def connect_client(
+    *,
+    target_host: str | None = None,
+    s3_endpoint: str = S3_ENDPOINT,
+) -> AsyncIterator[Client]:
+    """Keep the S3 client alive while Temporal reads and writes payloads."""
+    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])
+        data_converter = replace(DataConverter.default, external_storage=storage)
+        # Script the model so the demo needs no LLM provider or API key.
+        plugin = DeepAgentsPlugin(
+            data_converter=data_converter,
+            model_provider=mock_model_provider(
+                [
+                    AIMessage(
+                        content="",
+                        tool_calls=[
+                            {
+                                "name": "read_file",
+                                "args": {"file_path": "large-document.txt"},
+                                "id": "call-read",
+                            }
+                        ],
+                    ),
+                    AIMessage(content="I received the complete document."),
+                ]
+            ),
+        )
+        yield await Client.connect(**config, plugins=[plugin])

deepagents_plugin/external_storage/main.py

@@ -0,0 +1,40 @@
+"""Run an offline agent with native External Storage for a 6 MiB tool result."""
+
+import asyncio
+import hashlib
+import uuid
+
+from temporalio.worker import Worker
+
+from deepagents_plugin.external_storage.client import connect_client
+from deepagents_plugin.external_storage.workflow import (
+    PAYLOAD_BYTES,
+    ExternalStorageAgent,
+)
+
+
+async def main() -> None:
+    workflow_id = f"deepagents-external-storage-{uuid.uuid4()}"
+    async with connect_client() as client:
+        async with Worker(
+            client,
+            task_queue=workflow_id,
+            workflows=[ExternalStorageAgent],
+        ):
+            result = await client.execute_workflow(
+                ExternalStorageAgent.run,
+                "Read large-document.txt and acknowledge receiving it.",
+                id=workflow_id,
+                task_queue=workflow_id,
+            )
+
+    assert result.tool_result_bytes == PAYLOAD_BYTES
+    assert result.tool_result_sha256 == hashlib.sha256(b"x" * PAYLOAD_BYTES).hexdigest()
+    print(f"Workflow: {workflow_id}")
+    print(f"Tool result: {result.tool_result_bytes:,} bytes (6 MiB)")
+    print(f"SHA-256: {result.tool_result_sha256}")
+    print(f"Final answer: {result.last_answer}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())

deepagents_plugin/external_storage/workflow.py

@@ -0,0 +1,64 @@
+"""Exercise native External Storage with a 6 MiB tool result."""
+
+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.deepagents import create_temporal_deep_agent, tool_as_activity
+
+PAYLOAD_BYTES = 6 * 1024 * 1024
+
+
+@dataclass
+class ExternalStorageResult:
+    tool_result_bytes: int
+    tool_result_sha256: str
+    last_answer: str
+
+
+@tool
+async def read_file(file_path: str) -> str:
+    """Read a mock document containing exactly 6 MiB of text."""
+    del file_path
+    return "x" * PAYLOAD_BYTES
+
+
+@workflow.defn
+class ExternalStorageAgent:
+    @workflow.run
+    async def run(self, question: str) -> ExternalStorageResult:
+        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="fake:external-storage",
+            tools=[reader],
+            system_prompt="Read the requested document and acknowledge receiving it.",
+            activity_options={"start_to_close_timeout": timedelta(seconds=30)},
+        )
+
+        result = await agent.ainvoke(
+            {"messages": [{"role": "user", "content": question}]}
+        )
+        message = next(
+            message
+            for message in result["messages"]
+            if isinstance(message, ToolMessage) and message.name == "read_file"
+        )
+        if not isinstance(message.content, str):
+            raise TypeError("Expected the document tool result to be text")
+        answer = result["messages"][-1].content
+        if not isinstance(answer, str):
+            raise TypeError("Expected a text response from the model")
+        tool_bytes = message.content.encode()
+        return ExternalStorageResult(
+            tool_result_bytes=len(tool_bytes),
+            tool_result_sha256=hashlib.sha256(tool_bytes).hexdigest(),
+            last_answer=answer,
+        )