Add durable Strands sandbox support

CHANGELOG.md

@@ -20,6 +20,12 @@ to include examples, links to docs, or any other relevant information.
 
 ### Added
 
+- **Experimental**: `temporalio.contrib.strands` now supports durable,
+  Workflow-isolated Strands sandboxes through `TemporalSandbox` and
+  worker-side factories with run and Workflow-chain context registered with
+  `StrandsPlugin(sandboxes=...)`, using one shared activity set with optional
+  correlated live Workflow Streams output, worker-side environment references,
+  and async-context-manager factories for adapter cleanup.
 - Added GCP Cloud Run serverless-worker OpenTelemetry plugin in `temporalio.contrib.opentelemetry`.
 - Added new options to ActivityHandle.describe() to retrieve associated payloads, such as activity input and outcome.
 - New properties and methods in ActivityExecution and ActivityExecutionDescription.

pyproject.toml

@@ -56,7 +56,7 @@ cloud-run-worker-otel = [
 ]
 aioboto3 = ["aioboto3>=10.4.0", "types-aioboto3[s3]>=10.4.0"]
 google-genai = ["google-genai>=2.10.0,<3.0.0"]
-strands-agents = ["strands-agents>=1.39.0"]
+strands-agents = ["strands-agents>=1.51.0"]
 
 [project.urls]
 Homepage = "https://github.com/temporalio/sdk-python"
@@ -107,7 +107,7 @@ dev = [
   "opentelemetry-sdk-extension-aws>=2.0.0,<3",
   "pytest-flakefinder>=1.1.0",
   "async-timeout>=4.0,<6; python_version < '3.11'",
-  "strands-agents>=1.39.0",
+  "strands-agents>=1.51.0",
   "strands-agents-tools>=0.5.2",
   "mcp>=1.9.4,<2",
 ]

temporalio/contrib/strands/README.md

@@ -175,6 +175,206 @@ async for item in WorkflowStreamClient.create(client, workflow_id).subscribe(
     print(item.data)
 ```
 
+## Sandboxes
+
+`TemporalSandbox` implements Strands' sandbox API by scheduling every command,
+code, and filesystem operation as a Temporal Activity. Register the real
+worker-side sandbox under a name, then select that name in workflow code:
+
+```python
+from strands.sandbox.docker import DockerSandbox
+from temporalio.contrib.strands import (
+    SandboxWorkflowContext,
+    StrandsPlugin,
+    TemporalAgent,
+    TemporalSandbox,
+)
+
+async def build_sandbox(context: SandboxWorkflowContext) -> DockerSandbox:
+    # Application-specific and idempotent: return the existing container when
+    # another activity worker has already provisioned this Workflow's sandbox.
+    container = await get_or_create_build_container(
+        context.chain.first_execution_run_id
+    )
+    return DockerSandbox(container.name)
+
+# workflow
+agent = TemporalAgent(
+    sandbox=TemporalSandbox(
+        "build",
+        start_to_close_timeout=timedelta(minutes=5),
+    ),
+)
+
+# worker
+Worker(
+    ...,
+    plugins=[StrandsPlugin(sandboxes={
+        "build": build_sandbox,
+    })],
+)
+```
+
+The plugin registers one shared set of sandbox activities regardless of how
+many factories are configured. Each operation carries the selected sandbox
+name in its activity input so the worker can dispatch it to the matching
+factory.
+
+The factory is called lazily with a `SandboxWorkflowContext` containing the
+current `run_id` and a `chain` identity with the Workflow's namespace, Workflow
+ID, and first execution Run ID. The worker-local cache uses the sandbox name and
+chain identity, so each Workflow chain gets a separate sandbox for each
+registered name. Retries, Continue-As-New, Reset, and Cron runs belong to the
+same chain and therefore use the same sandbox; unrelated Workflow chains do not
+share one. Multiple `TemporalSandbox` objects with the same name in one chain
+intentionally share that chain's sandbox.
+
+The factory receives the current Run ID only when a worker-local cache entry is
+created. A later run in the same chain reuses a warm entry without calling the
+factory again. After eviction, the next factory call receives the Run ID of the
+run that recreates the entry.
+
+The first execution Run ID is supplied by workflow code because Activity
+metadata only identifies the current run. Treat sandbox activities as trusted
+worker endpoints: workflow code with access to their task queue can call them
+directly and choose this value. Do not share the task queue between mutually
+untrusted workflows. If Workflow IDs can be reused, the backing environment
+lookup must also prevent a new execution from reconnecting to an old execution's
+environment, for example by expiring old environments before ID reuse.
+
+Factories may be synchronous or asynchronous. Synchronous factories must only
+construct a lightweight adapter and must not block the activity event loop;
+use an asynchronous factory for remote lookup or provisioning. A factory may
+run more than once for the same context after cache eviction or on different
+workers, so provisioning must be idempotent. Strands' `DockerSandbox` only
+connects to an already-running container; it does not create one.
+
+Worker-local adapters are reused until they have been idle for five minutes.
+Set `sandbox_cache_idle_timeout` on `StrandsPlugin` to change that duration.
+Return an async context manager from the factory when an adapter owns clients,
+sockets, subprocess handles, tunnels, or leases. Its exit method runs after
+cache eviction and Worker shutdown. Returning a plain `Sandbox` only drops the
+adapter from the cache:
+
+```python
+from contextlib import asynccontextmanager
+
+@asynccontextmanager
+async def build_managed_sandbox(context: SandboxWorkflowContext):
+    adapter = await connect_to_sandbox(context)
+    try:
+        yield adapter
+    finally:
+        await adapter.aclose()
+```
+
+In either case, provisioning, teardown, and cleanup of orphaned backing
+environments remain the application's responsibility; use a backend TTL or
+reaper for workflows that are terminated before normal cleanup.
+
+That cache is per worker *process*, while successive sandbox activities from one
+workflow are routed independently across the task queue. With more than one
+worker on the queue, a `write-file` can land on one worker and the following
+`read-file` on another. The context factory must therefore reconnect every
+worker to the same Workflow-scoped backing environment rather than relying on
+per-process state. A single worker on the queue also satisfies this.
+
+Reset does not roll back commands or filesystem mutations already performed in
+the external sandbox, just as it does not roll back other Activity side effects.
+Account for that when resetting a Workflow that uses a sandbox.
+
+`SandboxTimeoutError` is serialized as an Activity failure and retried under the
+`retry_policy` you pass to `TemporalSandbox`. If retries are exhausted, it is
+reconstructed inside the workflow with the sandbox's own message. Any
+`FileNotFoundError` raised by a sandbox filesystem operation — including its
+`SandboxPathNotFoundError` subclass — is non-retryable because the requested path
+is absent, and is reconstructed in the same way. Factory failures and other
+sandbox failures, including the `OSError` that Strands documents for a failed
+`write_file`, are also retryable.
+
+Like all Temporal Activities, sandbox operations have at-least-once execution
+semantics. A worker can finish a command or filesystem mutation and fail before
+recording its result, causing a retry to perform the operation again. Use a
+bounded `retry_policy`, and make commands and mutations idempotent when repeated
+execution would be unsafe.
+
+By default, `TemporalSandbox.get_tools()` vends `sandbox_shell` and
+`sandbox_file_editor`. A tool passed explicitly through `TemporalAgent(tools=...)`
+with either name takes precedence, following Strands' normal sandbox-tool
+override behavior.
+
+Execution output is always buffered into the activity result so workflow replay
+observes the same ordered `StreamChunk` and `ExecutionResult` values. For live,
+observer-facing output, set `streaming_topic` and host a `WorkflowStream` on the
+workflow. The activity publishes each `StreamChunk` as it arrives; the final
+`ExecutionResult` is returned only through the buffered activity result:
+
+```python
+from datetime import timedelta
+
+from temporalio.contrib.strands import SandboxStreamEvent, TemporalSandbox
+from temporalio.contrib.workflow_streams import WorkflowStream, WorkflowStreamClient
+
+# workflow __init__
+self.stream = WorkflowStream()
+self.sandbox = TemporalSandbox(
+    "build",
+    start_to_close_timeout=timedelta(minutes=5),
+    streaming_topic="sandbox-events",
+)
+
+# external client
+async for item in WorkflowStreamClient.create(client, workflow_id).subscribe(
+    ["sandbox-events"], result_type=SandboxStreamEvent
+):
+    event = item.data
+    print(event.execution_id, event.sequence, event.chunk.data)
+```
+
+Each `SandboxStreamEvent` includes the sandbox name, an execution ID composed of
+the Workflow Run ID and Activity ID, the Activity attempt, chunk sequence, and
+`StreamChunk`. Events from concurrent executions may interleave on a shared
+topic; group them by `execution_id` and `attempt`, then order them by `sequence`.
+Workflow code still receives the correctly separated, complete buffered result
+for each call. Because publications are observer-facing side effects of an
+Activity attempt, a failed attempt may leave chunks in the topic before a retry
+publishes its own output.
+
+Streaming is disabled by default. When `streaming_topic=None`, sandbox
+activities do not construct a `WorkflowStreamClient` and the workflow does not
+need to host a `WorkflowStream`.
+
+All sandbox Activity arguments and results are serialized into workflow history.
+Keep command output and files within the server's configured payload size limits;
+use external storage for large artifacts.
+
+Do not put secret values directly in `env`, because those values are serialized
+into workflow history. Use `temporal_worker_env_ref()` to serialize only the
+name of a worker environment variable, then allow that name on every worker that
+runs the sandbox activities:
+
+```python
+from temporalio.contrib.strands import temporal_worker_env_ref
+
+# workflow
+await sandbox.execute(
+    "build",
+    env={"API_KEY": temporal_worker_env_ref("BUILD_API_KEY")},
+)
+
+# worker
+plugin = StrandsPlugin(
+    sandboxes={"build": build_sandbox},
+    resolvable_worker_env_vars=["BUILD_API_KEY"],
+)
+```
+
+Names are matched exactly. A reference to a name the worker does not allow is
+passed to the sandbox unchanged; an allowed but unset variable resolves to an
+empty string. `AllowAllWorkerEnvVars()` permits any name, but should only be used
+when all workflow code on the task queue is trusted to read every environment
+variable available to that worker.
+
 ## Tools
 
 Decorate non-deterministic tools with `@activity.defn`, or if you're importing tools from `strands_tools`, wrap them in a thin async function. Then, register the activity on the worker via `Worker(activities=[...])` and pass it to the agent with `workflow.activity_as_tool(activity, **options)` along with any activity options (e.g. `start_to_close_timeout`):

temporalio/contrib/strands/__init__.py

@@ -2,12 +2,25 @@
 
 from . import workflow
 from ._plugin import StrandsPlugin
+from ._sandbox_activity import (
+    SandboxStreamEvent,
+    SandboxWorkflowChain,
+    SandboxWorkflowContext,
+)
 from ._temporal_agent import TemporalAgent
 from ._temporal_mcp_client import TemporalMCPClient
+from ._temporal_sandbox import TemporalSandbox
+from ._worker_env_ref import AllowAllWorkerEnvVars, temporal_worker_env_ref
 
 __all__ = [
+    "AllowAllWorkerEnvVars",
+    "SandboxStreamEvent",
     "StrandsPlugin",
+    "SandboxWorkflowChain",
+    "SandboxWorkflowContext",
     "TemporalAgent",
     "TemporalMCPClient",
+    "TemporalSandbox",
+    "temporal_worker_env_ref",
     "workflow",
 ]

temporalio/contrib/strands/_plugin.py

@@ -1,4 +1,4 @@
-from collections.abc import AsyncGenerator, Callable
+from collections.abc import AsyncGenerator, Callable, Collection
 from contextlib import asynccontextmanager
 from dataclasses import replace
 from datetime import timedelta
@@ -16,11 +16,13 @@
 
 from ._failure_converter import StrandsFailureConverter
 from ._model_activity import ModelActivity
+from ._sandbox_activity import SandboxActivities, SandboxFactory
 from ._temporal_mcp_client import (
     _evict_connection,
     build_call_tool_activity,
     build_list_tools_activity,
 )
+from ._worker_env_ref import AllowAllWorkerEnvVars
 
 
 def _default_bedrock_model() -> Model:
@@ -51,19 +53,35 @@ class StrandsPlugin(SimplePlugin):
     ``mcp_connection_idle_timeout`` controls how long a worker-process MCP
     connection is kept open between ``call-tool`` activities before it is
     disconnected; the timer resets on every reuse. Defaults to 5 minutes.
+
+    When ``sandboxes`` is supplied, registers one stable set of activities that
+    dispatches each operation by sandbox name. Each factory receives the
+    requesting Workflow run's context and may return a sandbox directly or
+    awaitably. Worker-local adapters are cached by sandbox name and Workflow
+    chain until ``sandbox_cache_idle_timeout`` elapses. Use the same name in
+    workflow-side ``TemporalSandbox(name)`` instances.
+
+    ``resolvable_worker_env_vars`` controls which worker environment variables
+    sandbox command ``env`` references may resolve immediately before execution.
     """
 
     def __init__(
         self,
         *,
         models: dict[str, Callable[[], Model]] | None = None,
         mcp_clients: dict[str, Callable[[], MCPClient]] | None = None,
+        sandboxes: dict[str, SandboxFactory] | None = None,
         mcp_connection_idle_timeout: timedelta | None = None,
+        sandbox_cache_idle_timeout: timedelta | None = None,
+        resolvable_worker_env_vars: Collection[str] | AllowAllWorkerEnvVars = (),
     ) -> None:
-        """Build the plugin from optional model and MCP transport factories.
+        """Build the plugin from optional model, MCP, and sandbox factories.
 
         If ``models`` is omitted, registers a single ``BedrockModel`` factory
         under the name ``"bedrock"`` with Botocore retries disabled.
+
+        A sandbox factory may return an async context manager when its worker-local
+        adapter needs cleanup after cache eviction or Worker shutdown.
         """
         default_name: str | None = None
         if models is None:
@@ -74,6 +92,18 @@ def __init__(
             ma = ModelActivity(models, default_name=default_name)
             activities.extend([ma.invoke_model, ma.invoke_model_streaming])
 
+        sandbox_activities = (
+            SandboxActivities(
+                sandboxes,
+                sandbox_cache_idle_timeout,
+                resolvable_worker_env_vars,
+            )
+            if sandboxes
+            else None
+        )
+        if sandbox_activities is not None:
+            activities.extend(sandbox_activities.activities())
+
         mcp_clients = mcp_clients or {}
         for server, client_factory in mcp_clients.items():
             activities.append(
@@ -87,11 +117,21 @@ def __init__(
                 )
             )
 
+        sandbox_run_contexts = 0
+
         @asynccontextmanager
         async def run_context() -> AsyncGenerator[None, None]:
+            nonlocal sandbox_run_contexts
+            if sandbox_activities is not None:
+                sandbox_run_contexts += 1
             try:
                 yield
             finally:
+                if sandbox_activities is not None:
+                    sandbox_run_contexts -= 1
+                    # One plugin instance can be shared by multiple Workers.
+                    if sandbox_run_contexts == 0:
+                        await sandbox_activities.aclose()
                 for server in mcp_clients:
                     await _evict_connection(server)