diff --git a/CHANGELOG.md b/CHANGELOG.md
index be3444177..f4cdf8d89 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -20,6 +20,11 @@ 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
+  live Workflow Streams output.
 ### Changed
 
 ### Deprecated
diff --git a/pyproject.toml b/pyproject.toml
index d05d2f8c6..126d95813 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -46,7 +46,7 @@ lambda-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.47.0"]
 
 [project.urls]
 Homepage = "https://github.com/temporalio/sdk-python"
@@ -97,7 +97,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.47.0",
   "strands-agents-tools>=0.5.2",
   "mcp>=1.9.4,<2",
 ]
diff --git a/temporalio/contrib/strands/README.md b/temporalio/contrib/strands/README.md
index 126f4bd95..aa7d67db9 100644
--- a/temporalio/contrib/strands/README.md
+++ b/temporalio/contrib/strands/README.md
@@ -170,6 +170,154 @@ 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.
+
+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.
+Eviction only drops the local adapter. 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` and any `FileNotFoundError` — including its
+`SandboxPathNotFoundError` subclass — cross the activity boundary as
+non-retryable failures and are re-raised inside the workflow with the sandbox's
+own message, so a command that exceeds its `timeout` or a path that does not
+exist surfaces to the agent on the first attempt instead of retrying. Other
+sandbox failures, including the `OSError` that Strands documents for a failed
+`write_file`, are retried under the `retry_policy` you pass to
+`TemporalSandbox`.
+
+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_bash` 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 strands.sandbox import StreamChunk
+from temporalio.contrib.strands import 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=StreamChunk
+):
+    print(item.data.stream_type, item.data.data)
+```
+
+The topic is an observer-facing merged log. If sandbox executions overlap,
+their chunks may interleave. Use different `streaming_topic` values when the
+consumer needs separate logs; 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 that
+Temporal retries may leave chunks in the topic before the 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 arguments and results cross Temporal's payload boundary and enter workflow
+history. Keep command output and files within the server's configured payload
+size limits; use external storage for large artifacts. In particular, `env`
+values are recorded in history and must not contain secrets.
+
 ## 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`):
diff --git a/temporalio/contrib/strands/__init__.py b/temporalio/contrib/strands/__init__.py
index 39a8e7401..35d839e39 100644
--- a/temporalio/contrib/strands/__init__.py
+++ b/temporalio/contrib/strands/__init__.py
@@ -2,12 +2,17 @@
 
 from . import workflow
 from ._plugin import StrandsPlugin
+from ._sandbox_activity import SandboxWorkflowChain, SandboxWorkflowContext
 from ._temporal_agent import TemporalAgent
 from ._temporal_mcp_client import TemporalMCPClient
+from ._temporal_sandbox import TemporalSandbox
 
 __all__ = [
     "StrandsPlugin",
+    "SandboxWorkflowChain",
+    "SandboxWorkflowContext",
     "TemporalAgent",
     "TemporalMCPClient",
+    "TemporalSandbox",
     "workflow",
 ]
diff --git a/temporalio/contrib/strands/_plugin.py b/temporalio/contrib/strands/_plugin.py
index 0f1972666..75e035b51 100644
--- a/temporalio/contrib/strands/_plugin.py
+++ b/temporalio/contrib/strands/_plugin.py
@@ -1,9 +1,10 @@
-from collections.abc import AsyncGenerator, Callable
+from collections.abc import AsyncGenerator, Awaitable, Callable
 from contextlib import asynccontextmanager
 from dataclasses import replace
 from datetime import timedelta
 
 from strands.models import BedrockModel, Model
+from strands.sandbox import Sandbox
 from strands.tools.mcp import MCPClient
 
 from temporalio.contrib.pydantic import pydantic_data_converter
@@ -14,6 +15,10 @@
 
 from ._failure_converter import StrandsFailureConverter
 from ._model_activity import ModelActivity
+from ._sandbox_activity import (
+    SandboxActivities,
+    SandboxWorkflowContext,
+)
 from ._temporal_mcp_client import (
     _evict_connection,
     build_call_tool_activity,
@@ -39,6 +44,13 @@ 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.
     """
 
     def __init__(
@@ -46,9 +58,18 @@ def __init__(
         *,
         models: dict[str, Callable[[], Model]] | None = None,
         mcp_clients: dict[str, Callable[[], MCPClient]] | None = None,
+        sandboxes: dict[
+            str,
+            Callable[
+                [SandboxWorkflowContext],
+                Sandbox | Awaitable[Sandbox],
+            ],
+        ]
+        | None = None,
         mcp_connection_idle_timeout: timedelta | None = None,
+        sandbox_cache_idle_timeout: timedelta | None = None,
     ) -> 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"``, matching Strands' own implicit default.
@@ -62,6 +83,14 @@ 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)
+            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(
@@ -80,6 +109,8 @@ async def run_context() -> AsyncGenerator[None, None]:
             try:
                 yield
             finally:
+                if sandbox_activities is not None:
+                    await sandbox_activities.aclose()
                 for server in mcp_clients:
                     await _evict_connection(server)
 
diff --git a/temporalio/contrib/strands/_sandbox_activity.py b/temporalio/contrib/strands/_sandbox_activity.py
new file mode 100644
index 000000000..1654d3afd
--- /dev/null
+++ b/temporalio/contrib/strands/_sandbox_activity.py
@@ -0,0 +1,410 @@
+from __future__ import annotations
+
+import asyncio
+import base64
+import inspect
+from collections.abc import AsyncGenerator, Awaitable, Callable
+from contextlib import asynccontextmanager
+from dataclasses import dataclass, field
+from datetime import timedelta
+from typing import Any
+
+from strands.sandbox import (
+    ExecutionResult,
+    FileInfo,
+    Sandbox,
+    StreamChunk,
+)
+from strands.sandbox.errors import SandboxTimeoutError
+
+from temporalio import activity
+from temporalio.contrib.workflow_streams import WorkflowStreamClient
+from temporalio.exceptions import ApplicationError
+
+from ._heartbeat_decorator import auto_heartbeater
+
+SANDBOX_TIMEOUT_ERROR_TYPE = "StrandsSandboxTimeoutError"
+SANDBOX_PATH_NOT_FOUND_ERROR_TYPE = "StrandsSandboxPathNotFoundError"
+SANDBOX_NOT_FOUND_ERROR_TYPE = "StrandsSandboxNotFoundError"
+_SANDBOX_CACHE_IDLE_TIMEOUT = timedelta(minutes=5)
+
+
+@dataclass(frozen=True)
+class SandboxWorkflowChain:
+    """Identity shared by every run in a Workflow chain."""
+
+    namespace: str
+    workflow_id: str
+    first_execution_run_id: str
+
+
+@dataclass(frozen=True)
+class SandboxWorkflowContext:
+    """Workflow execution requesting a worker-side sandbox."""
+
+    chain: SandboxWorkflowChain
+    run_id: str
+
+    @property
+    def namespace(self) -> str:
+        """Namespace containing the Workflow."""
+        return self.chain.namespace
+
+    @property
+    def workflow_id(self) -> str:
+        """Workflow ID shared by the execution chain."""
+        return self.chain.workflow_id
+
+    @property
+    def first_execution_run_id(self) -> str:
+        """Run ID identifying the execution chain."""
+        return self.chain.first_execution_run_id
+
+
+SandboxFactory = Callable[[SandboxWorkflowContext], Sandbox | Awaitable[Sandbox]]
+_SandboxKey = tuple[str, SandboxWorkflowChain]
+
+
+@dataclass
+class _WorkflowScopedInput:
+    sandbox_name: str = field(default="", kw_only=True)
+    first_execution_run_id: str = field(default="", kw_only=True)
+
+
+@dataclass
+class _ExecuteInput(_WorkflowScopedInput):
+    command: str
+    timeout: float | None = None
+    cwd: str | None = None
+    env: dict[str, str] | None = None
+    kwargs: dict[str, Any] = field(default_factory=dict)
+    streaming_topic: str | None = None
+    streaming_batch_interval_seconds: float = 0.1
+
+
+@dataclass
+class _ExecuteCodeInput(_WorkflowScopedInput):
+    code: str
+    language: str
+    timeout: float | None = None
+    cwd: str | None = None
+    env: dict[str, str] | None = None
+    kwargs: dict[str, Any] = field(default_factory=dict)
+    streaming_topic: str | None = None
+    streaming_batch_interval_seconds: float = 0.1
+
+
+@dataclass
+class _PathInput(_WorkflowScopedInput):
+    path: str
+    kwargs: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class _WriteFileInput(_PathInput):
+    content_base64: str = ""
+
+
+@dataclass
+class _StreamItem:
+    value: dict[str, Any]
+
+
+class _SandboxRecord:
+    def __init__(
+        self,
+        owner: SandboxActivities,
+        key: _SandboxKey,
+        context: SandboxWorkflowContext,
+        factory: SandboxFactory,
+        idle_timeout: timedelta,
+    ) -> None:
+        self._owner = owner
+        self._key = key
+        self._context = context
+        self._idle_timeout = idle_timeout
+        self._inflight = 0
+        self._idle_handle: asyncio.TimerHandle | None = None
+        self._sandbox_task = asyncio.create_task(self._create(factory))
+
+    async def _create(self, factory: SandboxFactory) -> Sandbox:
+        sandbox = factory(self._context)
+        if inspect.isawaitable(sandbox):
+            return await sandbox
+        return sandbox
+
+    def acquire(self) -> None:
+        self._inflight += 1
+        if self._idle_handle is not None:
+            self._idle_handle.cancel()
+            self._idle_handle = None
+
+    def release(self) -> None:
+        self._inflight -= 1
+        if self._inflight == 0 and self._owner._has_record(self._key, self):
+            self._idle_handle = asyncio.get_running_loop().call_later(
+                self._idle_timeout.total_seconds(), self._on_idle
+            )
+
+    def _on_idle(self) -> None:
+        self._idle_handle = None
+        if self._inflight == 0:
+            self._owner._evict(self._key, self)
+
+    async def sandbox(self) -> Sandbox:
+        return await asyncio.shield(self._sandbox_task)
+
+    def creation_done(self) -> bool:
+        return self._sandbox_task.done()
+
+    async def aclose(self) -> None:
+        if self._idle_handle is not None:
+            self._idle_handle.cancel()
+            self._idle_handle = None
+        if not self._sandbox_task.done():
+            self._sandbox_task.cancel()
+        try:
+            await self._sandbox_task
+        except BaseException:
+            pass
+
+
+class SandboxActivities:
+    """Lazily resolves Workflow-scoped sandboxes and exposes their activities."""
+
+    def __init__(
+        self,
+        factories: dict[str, SandboxFactory],
+        idle_timeout: timedelta | None = None,
+    ) -> None:
+        """Store named Workflow-scoped worker-side sandbox factories."""
+        self._factories = dict(factories)
+        self._idle_timeout = (
+            idle_timeout if idle_timeout is not None else _SANDBOX_CACHE_IDLE_TIMEOUT
+        )
+        if self._idle_timeout <= timedelta(0):
+            raise ValueError("Sandbox cache idle timeout must be positive")
+        self._records: dict[_SandboxKey, _SandboxRecord] = {}
+
+    @asynccontextmanager
+    async def _sandbox(
+        self, input: _WorkflowScopedInput
+    ) -> AsyncGenerator[Sandbox, None]:
+        info = activity.info()
+        if (
+            not info.workflow_id
+            or not info.workflow_run_id
+            or not input.first_execution_run_id
+        ):
+            raise RuntimeError("Sandbox activities must be started by a Workflow")
+        context = SandboxWorkflowContext(
+            chain=SandboxWorkflowChain(
+                namespace=info.namespace,
+                workflow_id=info.workflow_id,
+                first_execution_run_id=input.first_execution_run_id,
+            ),
+            run_id=info.workflow_run_id,
+        )
+        factory = self._factories.get(input.sandbox_name)
+        if factory is None:
+            raise ApplicationError(
+                f"Unknown sandbox name {input.sandbox_name!r}. "
+                f"Known: {sorted(self._factories)}",
+                type=SANDBOX_NOT_FOUND_ERROR_TYPE,
+                non_retryable=True,
+            )
+        key = (input.sandbox_name, context.chain)
+        record = self._records.get(key)
+        if record is None:
+            record = _SandboxRecord(self, key, context, factory, self._idle_timeout)
+            self._records[key] = record
+        record.acquire()
+        try:
+            try:
+                sandbox = await record.sandbox()
+            except asyncio.CancelledError:
+                # One cancelled activity must not cancel or evict initialization
+                # that another activity for the same Workflow is awaiting.
+                if record.creation_done():
+                    self._evict(key, record)
+                raise
+            except BaseException:
+                self._evict(key, record)
+                raise
+            yield sandbox
+        finally:
+            record.release()
+
+    def _has_record(self, key: _SandboxKey, record: _SandboxRecord) -> bool:
+        return self._records.get(key) is record
+
+    def _evict(self, key: _SandboxKey, record: _SandboxRecord) -> None:
+        if self._has_record(key, record):
+            del self._records[key]
+
+    async def aclose(self) -> None:
+        """Cancel cache timers and discard all worker-local sandbox adapters."""
+        records = list(self._records.values())
+        self._records.clear()
+        for record in records:
+            await record.aclose()
+
+    def activities(self) -> list[Callable[..., Any]]:
+        """Build one stable activity set that dispatches by sandbox name."""
+
+        @activity.defn(name=_activity_name("execute"))
+        @auto_heartbeater
+        async def execute(input: _ExecuteInput) -> list[_StreamItem]:
+            async with self._sandbox(input) as sandbox:
+                return await self._run_stream(
+                    sandbox.execute_streaming(
+                        input.command,
+                        timeout=input.timeout,
+                        cwd=input.cwd,
+                        env=input.env,
+                        **input.kwargs,
+                    ),
+                    timeout=input.timeout,
+                    streaming_topic=input.streaming_topic,
+                    streaming_batch_interval_seconds=input.streaming_batch_interval_seconds,
+                )
+
+        @activity.defn(name=_activity_name("execute-code"))
+        @auto_heartbeater
+        async def execute_code(
+            input: _ExecuteCodeInput,
+        ) -> list[_StreamItem]:
+            async with self._sandbox(input) as sandbox:
+                return await self._run_stream(
+                    sandbox.execute_code_streaming(
+                        input.code,
+                        input.language,
+                        timeout=input.timeout,
+                        cwd=input.cwd,
+                        env=input.env,
+                        **input.kwargs,
+                    ),
+                    timeout=input.timeout,
+                    streaming_topic=input.streaming_topic,
+                    streaming_batch_interval_seconds=input.streaming_batch_interval_seconds,
+                )
+
+        @activity.defn(name=_activity_name("read-file"))
+        @auto_heartbeater
+        async def read_file(input: _PathInput) -> bytes:
+            try:
+                async with self._sandbox(input) as sandbox:
+                    return await sandbox.read_file(input.path, **input.kwargs)
+            except FileNotFoundError as err:
+                raise _path_not_found_error(err, input.path) from err
+
+        @activity.defn(name=_activity_name("write-file"))
+        @auto_heartbeater
+        async def write_file(input: _WriteFileInput) -> None:
+            try:
+                async with self._sandbox(input) as sandbox:
+                    await sandbox.write_file(
+                        input.path,
+                        base64.b64decode(input.content_base64),
+                        **input.kwargs,
+                    )
+            except FileNotFoundError as err:
+                raise _path_not_found_error(err, input.path) from err
+
+        @activity.defn(name=_activity_name("remove-file"))
+        @auto_heartbeater
+        async def remove_file(input: _PathInput) -> None:
+            try:
+                async with self._sandbox(input) as sandbox:
+                    await sandbox.remove_file(input.path, **input.kwargs)
+            except FileNotFoundError as err:
+                raise _path_not_found_error(err, input.path) from err
+
+        @activity.defn(name=_activity_name("list-files"))
+        @auto_heartbeater
+        async def list_files(input: _PathInput) -> list[FileInfo]:
+            try:
+                async with self._sandbox(input) as sandbox:
+                    return await sandbox.list_files(input.path, **input.kwargs)
+            except FileNotFoundError as err:
+                raise _path_not_found_error(err, input.path) from err
+
+        return [execute, execute_code, read_file, write_file, remove_file, list_files]
+
+    async def _run_stream(
+        self,
+        stream: AsyncGenerator[StreamChunk | ExecutionResult, None],
+        *,
+        timeout: float | None,
+        streaming_topic: str | None,
+        streaming_batch_interval_seconds: float,
+    ) -> list[_StreamItem]:
+        items: list[_StreamItem] = []
+        try:
+            if streaming_topic is None:
+                async for item in stream:
+                    items.append(_StreamItem(_item_to_json(item)))
+                return items
+
+            client = WorkflowStreamClient.from_within_activity(
+                batch_interval=timedelta(seconds=streaming_batch_interval_seconds),
+            )
+            topic = client.topic(streaming_topic, type=StreamChunk)
+            async with client:
+                async for item in stream:
+                    items.append(_StreamItem(_item_to_json(item)))
+                    if isinstance(item, StreamChunk):
+                        topic.publish(item)
+            return items
+        except SandboxTimeoutError as err:
+            raise _timeout_error(err, timeout) from err
+
+
+def _activity_name(operation: str) -> str:
+    return f"strands-sandbox-{operation}"
+
+
+def _timeout_error(err: SandboxTimeoutError, timeout: float | None) -> ApplicationError:
+    # A timeout is the deterministic outcome the caller asked for, so retrying
+    # just repeats it. Surface it to workflow code on the first attempt instead.
+    return ApplicationError(
+        str(err),
+        timeout,
+        type=SANDBOX_TIMEOUT_ERROR_TYPE,
+        non_retryable=True,
+    )
+
+
+def _path_not_found_error(err: FileNotFoundError, path: str) -> ApplicationError:
+    # Strands documents FileNotFoundError, not SandboxPathNotFoundError, for
+    # read/remove/list; only list_files raises the sandbox-specific subclass.
+    # Either way the path is missing on every attempt, so retrying is futile.
+    return ApplicationError(
+        str(err),
+        path,
+        type=SANDBOX_PATH_NOT_FOUND_ERROR_TYPE,
+        non_retryable=True,
+    )
+
+
+def _item_to_json(item: StreamChunk | ExecutionResult) -> dict[str, Any]:
+    if isinstance(item, StreamChunk):
+        return {
+            "kind": "stream_chunk",
+            "data": item.data,
+            "stream_type": item.stream_type,
+        }
+    return {
+        "kind": "execution_result",
+        "exit_code": item.exit_code,
+        "stdout": item.stdout,
+        "stderr": item.stderr,
+        "output_files": [
+            {
+                "name": output.name,
+                "content_base64": base64.b64encode(output.content).decode("ascii"),
+                "mime_type": output.mime_type,
+            }
+            for output in item.output_files
+        ],
+    }
diff --git a/temporalio/contrib/strands/_temporal_agent.py b/temporalio/contrib/strands/_temporal_agent.py
index c2f9f14c7..41b13afa4 100644
--- a/temporalio/contrib/strands/_temporal_agent.py
+++ b/temporalio/contrib/strands/_temporal_agent.py
@@ -9,6 +9,7 @@
 
 from ._temporal_mcp_client import TemporalMCPClient
 from ._temporal_model import TemporalModel
+from ._temporal_sandbox import TemporalSandbox
 
 _SNAPSHOT_DISABLED = (
     "TemporalAgent disables take_snapshot()/load_snapshot(). Temporal "
@@ -23,8 +24,9 @@ class TemporalAgent(Agent):
 
     ``model`` is the name of a factory registered in
     ``StrandsPlugin(models={...})``. The activity options apply to every model
-    invocation this agent makes. All other keyword arguments are forwarded to
-    Strands' ``Agent`` (``tools``, ``hooks``, ``system_prompt``,
+    invocation this agent makes. ``sandbox`` is a workflow-side
+    ``TemporalSandbox`` whose name selects a worker-side factory. All other
+    keyword arguments are forwarded to Strands' ``Agent`` (``tools``, ``hooks``, ``system_prompt``,
     ``structured_output_model``, ``messages``, etc.).
 
     Strands' ``retry_strategy`` is disabled; configure retries via
@@ -48,6 +50,7 @@ def __init__(
         priority: Priority = Priority.default,
         streaming_topic: str | None = None,
         streaming_batch_interval: timedelta = timedelta(milliseconds=100),
+        sandbox: TemporalSandbox | None = None,
         **agent_kwargs: Any,
     ) -> None:
         """Build a TemporalAgent from a registered model name and activity options."""
@@ -76,7 +79,7 @@ def __init__(
             streaming_topic=streaming_topic,
             streaming_batch_interval=streaming_batch_interval,
         )
-        super().__init__(model=temporal_model, **agent_kwargs)
+        super().__init__(model=temporal_model, sandbox=sandbox, **agent_kwargs)
 
         # Strands invokes ToolProvider.load_tools() once at construction on a
         # separate run_async thread that has no workflow runtime, so a
diff --git a/temporalio/contrib/strands/_temporal_sandbox.py b/temporalio/contrib/strands/_temporal_sandbox.py
new file mode 100644
index 000000000..e06e086c2
--- /dev/null
+++ b/temporalio/contrib/strands/_temporal_sandbox.py
@@ -0,0 +1,205 @@
+import base64
+from collections.abc import AsyncGenerator
+from datetime import timedelta
+from typing import Any, TypeVar
+
+from strands.sandbox import ExecutionResult, FileInfo, OutputFile, Sandbox, StreamChunk
+from strands.sandbox.errors import SandboxPathNotFoundError, SandboxTimeoutError
+from strands.types.tools import AgentTool
+from strands.vended_tools.bash import make_bash
+from strands.vended_tools.file_editor import make_file_editor
+
+from temporalio import workflow
+from temporalio.common import Priority, RetryPolicy
+from temporalio.exceptions import ActivityError, ApplicationError
+from temporalio.workflow import ActivityCancellationType, VersioningIntent
+
+from ._sandbox_activity import (
+    SANDBOX_PATH_NOT_FOUND_ERROR_TYPE,
+    SANDBOX_TIMEOUT_ERROR_TYPE,
+    _activity_name,
+    _ExecuteCodeInput,
+    _ExecuteInput,
+    _PathInput,
+    _StreamItem,
+    _WriteFileInput,
+)
+
+_ErrorT = TypeVar("_ErrorT", bound=OSError)
+
+
+class TemporalSandbox(Sandbox):
+    """Workflow-side sandbox that dispatches operations as Temporal activities."""
+
+    def __init__(
+        self,
+        name: str,
+        *,
+        task_queue: str | None = None,
+        schedule_to_close_timeout: timedelta | None = None,
+        schedule_to_start_timeout: timedelta | None = None,
+        start_to_close_timeout: timedelta | None = None,
+        heartbeat_timeout: timedelta | None = None,
+        retry_policy: RetryPolicy | None = None,
+        cancellation_type: ActivityCancellationType = ActivityCancellationType.TRY_CANCEL,
+        versioning_intent: VersioningIntent | None = None,
+        summary: str | None = None,
+        priority: Priority = Priority.default,
+        streaming_topic: str | None = None,
+        streaming_batch_interval: timedelta = timedelta(milliseconds=100),
+    ) -> None:
+        """Configure a registered sandbox name and its activity options."""
+        self._name = name
+        self._streaming_topic = streaming_topic
+        self._streaming_batch_interval = streaming_batch_interval
+        self._options: dict[str, Any] = {
+            "task_queue": task_queue,
+            "schedule_to_close_timeout": schedule_to_close_timeout,
+            "schedule_to_start_timeout": schedule_to_start_timeout,
+            "start_to_close_timeout": start_to_close_timeout,
+            "heartbeat_timeout": heartbeat_timeout,
+            "retry_policy": retry_policy,
+            "cancellation_type": cancellation_type,
+            "versioning_intent": versioning_intent,
+            "summary": summary,
+            "priority": priority,
+        }
+
+    async def execute_streaming(
+        self,
+        command: str,
+        *,
+        timeout: float | None = None,
+        cwd: str | None = None,
+        env: dict[str, str] | None = None,
+        **kwargs: Any,
+    ) -> AsyncGenerator[StreamChunk | ExecutionResult, None]:
+        """Execute a command in the registered worker-side sandbox."""
+        items = await self._execute(
+            "execute",
+            _ExecuteInput(
+                command=command,
+                timeout=timeout,
+                cwd=cwd,
+                env=env,
+                kwargs=kwargs,
+                streaming_topic=self._streaming_topic,
+                streaming_batch_interval_seconds=self._streaming_batch_interval.total_seconds(),
+            ),
+            result_type=list[_StreamItem],
+        )
+        for item in items:
+            yield _item_from_json(item.value)
+
+    async def execute_code_streaming(
+        self,
+        code: str,
+        language: str,
+        *,
+        timeout: float | None = None,
+        cwd: str | None = None,
+        env: dict[str, str] | None = None,
+        **kwargs: Any,
+    ) -> AsyncGenerator[StreamChunk | ExecutionResult, None]:
+        """Execute code in the registered worker-side sandbox."""
+        items = await self._execute(
+            "execute-code",
+            _ExecuteCodeInput(
+                code=code,
+                language=language,
+                timeout=timeout,
+                cwd=cwd,
+                env=env,
+                kwargs=kwargs,
+                streaming_topic=self._streaming_topic,
+                streaming_batch_interval_seconds=self._streaming_batch_interval.total_seconds(),
+            ),
+            result_type=list[_StreamItem],
+        )
+        for item in items:
+            yield _item_from_json(item.value)
+
+    async def read_file(self, path: str, **kwargs: Any) -> bytes:
+        """Read bytes from the registered worker-side sandbox."""
+        return await self._execute(
+            "read-file", _PathInput(path, kwargs), result_type=bytes
+        )
+
+    async def write_file(self, path: str, content: bytes, **kwargs: Any) -> None:
+        """Write bytes to the registered worker-side sandbox."""
+        await self._execute(
+            "write-file",
+            _WriteFileInput(path, kwargs, base64.b64encode(content).decode("ascii")),
+        )
+
+    async def remove_file(self, path: str, **kwargs: Any) -> None:
+        """Remove a file from the registered worker-side sandbox."""
+        await self._execute("remove-file", _PathInput(path, kwargs))
+
+    async def list_files(self, path: str, **kwargs: Any) -> list[FileInfo]:
+        """List a directory in the registered worker-side sandbox."""
+        return await self._execute(
+            "list-files", _PathInput(path, kwargs), result_type=list[FileInfo]
+        )
+
+    def get_tools(self) -> list[AgentTool]:
+        """Vend Strands' standard bash and file-editor sandbox tools."""
+        return [
+            make_file_editor(sandbox=self, name="sandbox_file_editor"),
+            make_bash(sandbox=self, name="sandbox_bash"),
+        ]
+
+    async def _execute(
+        self, operation: str, input: Any, *, result_type: type | None = None
+    ) -> Any:
+        input.sandbox_name = self._name
+        input.first_execution_run_id = workflow.info().first_execution_run_id
+        try:
+            return await workflow.execute_activity(
+                _activity_name(operation),
+                input,
+                result_type=result_type,
+                **self._options,
+            )
+        except ActivityError as err:
+            cause = err.__cause__
+            if isinstance(cause, ApplicationError):
+                if cause.type == SANDBOX_TIMEOUT_ERROR_TYPE:
+                    seconds = cause.details[0] if cause.details else None
+                    raise _with_message(SandboxTimeoutError(seconds), cause) from err
+                if cause.type == SANDBOX_PATH_NOT_FOUND_ERROR_TYPE:
+                    path = cause.details[0] if cause.details else ""
+                    raise _with_message(SandboxPathNotFoundError(path), cause) from err
+            raise
+
+
+def _with_message(error: _ErrorT, cause: ApplicationError) -> _ErrorT:
+    # The details only carry what the workflow needs to rebuild the error type.
+    # Restore the sandbox's own message so a timeout reports the duration the
+    # sandbox actually enforced, not the one the caller requested, and a missing
+    # path keeps whatever the backing environment said about it.
+    if cause.message:
+        error.args = (cause.message,)
+    return error
+
+
+def _item_from_json(value: Any) -> StreamChunk | ExecutionResult:
+    if not isinstance(value, dict):
+        raise TypeError("Sandbox stream item must be an object")
+    if value.get("kind") == "stream_chunk":
+        return StreamChunk(value["data"], value["stream_type"])
+    if value.get("kind") == "execution_result":
+        return ExecutionResult(
+            exit_code=value["exit_code"],
+            stdout=value["stdout"],
+            stderr=value["stderr"],
+            output_files=[
+                OutputFile(
+                    name=output["name"],
+                    content=base64.b64decode(output["content_base64"]),
+                    mime_type=output["mime_type"],
+                )
+                for output in value["output_files"]
+            ],
+        )
+    raise ValueError(f"Unknown sandbox stream item kind: {value.get('kind')!r}")
diff --git a/tests/contrib/strands/test_sandbox.py b/tests/contrib/strands/test_sandbox.py
new file mode 100644
index 000000000..132ce489c
--- /dev/null
+++ b/tests/contrib/strands/test_sandbox.py
@@ -0,0 +1,687 @@
+import asyncio
+from collections.abc import AsyncGenerator
+from dataclasses import dataclass
+from datetime import timedelta
+from typing import Any
+from uuid import uuid4
+
+import pytest
+from strands import SandboxPathNotFoundError, SandboxTimeoutError, tool
+from strands.sandbox import ExecutionResult, FileInfo, OutputFile, Sandbox, StreamChunk
+
+from temporalio import workflow
+from temporalio.client import Client
+from temporalio.common import RetryPolicy
+from temporalio.contrib.strands import (
+    SandboxWorkflowChain,
+    SandboxWorkflowContext,
+    StrandsPlugin,
+    TemporalAgent,
+    TemporalSandbox,
+)
+from temporalio.contrib.workflow_streams import WorkflowStream, WorkflowStreamClient
+from temporalio.worker import Replayer, Worker
+from tests.contrib.strands.common import get_activities
+
+
+class RecordingSandbox(Sandbox):
+    def __init__(self) -> None:
+        self.calls: list[tuple[Any, ...]] = []
+        self.files = {"/binary": b"\x00\xff"}
+
+    async def execute_streaming(
+        self,
+        command: str,
+        *,
+        timeout: float | None = None,
+        cwd: str | None = None,
+        env: dict[str, str] | None = None,
+        **kwargs: Any,
+    ) -> AsyncGenerator[StreamChunk | ExecutionResult, None]:
+        self.calls.append(("execute", command, timeout, cwd, env, kwargs))
+        yield StreamChunk("out")
+        yield StreamChunk("err", "stderr")
+        yield ExecutionResult(
+            0,
+            "out",
+            "err",
+            [OutputFile("artifact.bin", b"\x80\xff")],
+        )
+
+    async def execute_code_streaming(
+        self,
+        code: str,
+        language: str,
+        *,
+        timeout: float | None = None,
+        cwd: str | None = None,
+        env: dict[str, str] | None = None,
+        **kwargs: Any,
+    ) -> AsyncGenerator[StreamChunk | ExecutionResult, None]:
+        self.calls.append(("execute_code", code, language, timeout, cwd, env, kwargs))
+        yield ExecutionResult(0, code, "")
+
+    async def read_file(self, path: str, **kwargs: Any) -> bytes:
+        self.calls.append(("read_file", path, kwargs))
+        return self.files[path]
+
+    async def write_file(self, path: str, content: bytes, **kwargs: Any) -> None:
+        self.calls.append(("write_file", path, content, kwargs))
+        self.files[path] = content
+
+    async def remove_file(self, path: str, **kwargs: Any) -> None:
+        self.calls.append(("remove_file", path, kwargs))
+        del self.files[path]
+
+    async def list_files(self, path: str, **kwargs: Any) -> list[FileInfo]:
+        self.calls.append(("list_files", path, kwargs))
+        return [FileInfo("binary", False, len(self.files["/binary"]))]
+
+
+@dataclass
+class SandboxWorkflowResult:
+    command_items_match: bool
+    code_result: ExecutionResult
+    binary_values_match: bool
+    files: list[FileInfo]
+
+
+@workflow.defn
+class SandboxWorkflow:
+    @workflow.run
+    async def run(self) -> SandboxWorkflowResult:
+        sandbox = TemporalSandbox(
+            "recording", start_to_close_timeout=timedelta(seconds=15)
+        )
+        command_items = [
+            item
+            async for item in sandbox.execute_streaming(
+                "echo hi",
+                timeout=2,
+                cwd="/work",
+                env={"VISIBLE": "history"},
+                future_option=True,
+            )
+        ]
+        expected_command_items = [
+            StreamChunk("out"),
+            StreamChunk("err", "stderr"),
+            ExecutionResult(
+                0,
+                "out",
+                "err",
+                [OutputFile("artifact.bin", b"\x80\xff")],
+            ),
+        ]
+        code_result = await sandbox.execute_code(
+            "print('hi')", "python3", future_option=2
+        )
+        original = await sandbox.read_file("/binary", future_option=3)
+        await sandbox.write_file("/other", b"\x01\xfe", future_option=4)
+        written = await sandbox.read_file("/other")
+        await sandbox.remove_file("/other", future_option=5)
+        files = await sandbox.list_files("/", future_option=6)
+        return SandboxWorkflowResult(
+            command_items == expected_command_items,
+            code_result,
+            original == b"\x00\xff" and written == b"\x01\xfe",
+            files,
+        )
+
+
+async def test_sandbox_operations_are_durable_and_cached(client: Client):
+    task_queue = f"test_sandbox-{uuid4()}"
+    constructed: list[RecordingSandbox] = []
+    contexts: list[SandboxWorkflowContext] = []
+
+    def factory(context: SandboxWorkflowContext) -> RecordingSandbox:
+        sandbox = RecordingSandbox()
+        contexts.append(context)
+        constructed.append(sandbox)
+        return sandbox
+
+    plugin = StrandsPlugin(models={}, sandboxes={"recording": factory})
+    async with Worker(
+        client,
+        task_queue=task_queue,
+        workflows=[SandboxWorkflow],
+        plugins=[plugin],
+        max_cached_workflows=0,
+    ):
+        handle = await client.start_workflow(
+            SandboxWorkflow.run,
+            id=f"test_sandbox-{uuid4()}",
+            task_queue=task_queue,
+        )
+        result = await handle.result()
+
+    assert result.command_items_match
+    assert result.code_result == ExecutionResult(0, "print('hi')", "")
+    assert result.binary_values_match
+    assert result.files == [FileInfo("binary", False, 2)]
+    assert len(constructed) == 1
+    assert contexts == [
+        SandboxWorkflowContext(
+            chain=SandboxWorkflowChain(
+                namespace=client.namespace,
+                workflow_id=handle.id,
+                first_execution_run_id=handle.first_execution_run_id or "",
+            ),
+            run_id=handle.result_run_id or "",
+        )
+    ]
+    assert constructed[0].calls == [
+        (
+            "execute",
+            "echo hi",
+            2,
+            "/work",
+            {"VISIBLE": "history"},
+            {"future_option": True},
+        ),
+        (
+            "execute_code",
+            "print('hi')",
+            "python3",
+            None,
+            None,
+            None,
+            {"future_option": 2},
+        ),
+        ("read_file", "/binary", {"future_option": 3}),
+        ("write_file", "/other", b"\x01\xfe", {"future_option": 4}),
+        ("read_file", "/other", {}),
+        ("remove_file", "/other", {"future_option": 5}),
+        ("list_files", "/", {"future_option": 6}),
+    ]
+
+    history = await handle.fetch_history()
+    assert get_activities(history) == [
+        "strands-sandbox-execute",
+        "strands-sandbox-execute-code",
+        "strands-sandbox-read-file",
+        "strands-sandbox-write-file",
+        "strands-sandbox-read-file",
+        "strands-sandbox-remove-file",
+        "strands-sandbox-list-files",
+    ]
+    await Replayer(workflows=[SandboxWorkflow], plugins=[plugin]).replay_workflow(
+        history
+    )
+
+
+@workflow.defn
+class IsolatedSandboxWorkflow:
+    @workflow.run
+    async def run(self, value: bytes) -> bytes:
+        sandbox = TemporalSandbox(
+            "recording", start_to_close_timeout=timedelta(seconds=15)
+        )
+        await sandbox.write_file("/value", value)
+        return await sandbox.read_file("/value")
+
+
+async def test_sandbox_isolated_per_workflow_with_async_factory(client: Client):
+    task_queue = f"test_sandbox_isolation-{uuid4()}"
+    sandboxes: dict[SandboxWorkflowContext, RecordingSandbox] = {}
+
+    async def factory(context: SandboxWorkflowContext) -> RecordingSandbox:
+        await asyncio.sleep(0)
+        sandbox = RecordingSandbox()
+        sandboxes[context] = sandbox
+        return sandbox
+
+    plugin = StrandsPlugin(models={}, sandboxes={"recording": factory})
+    async with Worker(
+        client,
+        task_queue=task_queue,
+        workflows=[IsolatedSandboxWorkflow],
+        plugins=[plugin],
+        max_cached_workflows=0,
+    ):
+        handles = [
+            await client.start_workflow(
+                IsolatedSandboxWorkflow.run,
+                value,
+                id=f"test_sandbox_isolation-{uuid4()}",
+                task_queue=task_queue,
+            )
+            for value in (b"first", b"second")
+        ]
+        assert await asyncio.gather(*(handle.result() for handle in handles)) == [
+            b"first",
+            b"second",
+        ]
+
+    assert len(sandboxes) == 2
+    assert {context.workflow_id for context in sandboxes} == {
+        handle.id for handle in handles
+    }
+    assert {sandbox.files["/value"] for sandbox in sandboxes.values()} == {
+        b"first",
+        b"second",
+    }
+
+
+@workflow.defn
+class ContinueAsNewSandboxWorkflow:
+    @workflow.run
+    async def run(self, continued: bool = False) -> bytes:
+        sandbox = TemporalSandbox(
+            "recording", start_to_close_timeout=timedelta(seconds=15)
+        )
+        if not continued:
+            await sandbox.write_file("/continued", b"same sandbox")
+            workflow.continue_as_new(True)
+        return await sandbox.read_file("/continued")
+
+
+async def test_sandbox_reused_across_continue_as_new(client: Client):
+    task_queue = f"test_sandbox_continue_as_new-{uuid4()}"
+    contexts: list[SandboxWorkflowContext] = []
+
+    def factory(context: SandboxWorkflowContext) -> RecordingSandbox:
+        contexts.append(context)
+        return RecordingSandbox()
+
+    plugin = StrandsPlugin(models={}, sandboxes={"recording": factory})
+    async with Worker(
+        client,
+        task_queue=task_queue,
+        workflows=[ContinueAsNewSandboxWorkflow],
+        plugins=[plugin],
+        max_cached_workflows=0,
+    ):
+        handle = await client.start_workflow(
+            ContinueAsNewSandboxWorkflow.run,
+            id=f"test_sandbox_continue_as_new-{uuid4()}",
+            task_queue=task_queue,
+        )
+        assert await handle.result() == b"same sandbox"
+
+    assert len(contexts) == 1
+    assert contexts[0].first_execution_run_id == handle.first_execution_run_id
+    assert contexts[0].run_id == handle.first_execution_run_id
+
+
+@workflow.defn
+class RetriedSandboxWorkflow:
+    @workflow.run
+    async def run(self) -> bytes:
+        sandbox = TemporalSandbox(
+            "recording", start_to_close_timeout=timedelta(seconds=15)
+        )
+        if workflow.info().attempt == 1:
+            await sandbox.write_file("/retried", b"same sandbox")
+            raise RuntimeError("retry workflow")
+        return await sandbox.read_file("/retried")
+
+
+async def test_sandbox_reused_across_workflow_retry(client: Client):
+    task_queue = f"test_sandbox_workflow_retry-{uuid4()}"
+    contexts: list[SandboxWorkflowContext] = []
+
+    def factory(context: SandboxWorkflowContext) -> RecordingSandbox:
+        contexts.append(context)
+        return RecordingSandbox()
+
+    plugin = StrandsPlugin(models={}, sandboxes={"recording": factory})
+    async with Worker(
+        client,
+        task_queue=task_queue,
+        workflows=[RetriedSandboxWorkflow],
+        plugins=[plugin],
+        max_cached_workflows=0,
+        workflow_failure_exception_types=[RuntimeError],
+    ):
+        handle = await client.start_workflow(
+            RetriedSandboxWorkflow.run,
+            id=f"test_sandbox_workflow_retry-{uuid4()}",
+            task_queue=task_queue,
+            retry_policy=RetryPolicy(
+                initial_interval=timedelta(milliseconds=1), maximum_attempts=2
+            ),
+        )
+        assert await handle.result() == b"same sandbox"
+
+    assert len(contexts) == 1
+    assert contexts[0].first_execution_run_id == handle.first_execution_run_id
+    assert contexts[0].run_id == handle.first_execution_run_id
+
+
+@workflow.defn
+class IdleSandboxWorkflow:
+    @workflow.run
+    async def run(self) -> None:
+        sandbox = TemporalSandbox(
+            "recording", start_to_close_timeout=timedelta(seconds=15)
+        )
+        await sandbox.read_file("/binary")
+        await workflow.sleep(0.25)
+        await sandbox.read_file("/binary")
+
+
+async def test_sandbox_cache_evicts_when_idle(client: Client):
+    task_queue = f"test_sandbox_idle-{uuid4()}"
+    contexts: list[SandboxWorkflowContext] = []
+
+    def factory(context: SandboxWorkflowContext) -> RecordingSandbox:
+        contexts.append(context)
+        return RecordingSandbox()
+
+    plugin = StrandsPlugin(
+        models={},
+        sandboxes={"recording": factory},
+        sandbox_cache_idle_timeout=timedelta(milliseconds=50),
+    )
+    async with Worker(
+        client,
+        task_queue=task_queue,
+        workflows=[IdleSandboxWorkflow],
+        plugins=[plugin],
+        max_cached_workflows=0,
+    ):
+        await client.execute_workflow(
+            IdleSandboxWorkflow.run,
+            id=f"test_sandbox_idle-{uuid4()}",
+            task_queue=task_queue,
+        )
+
+    assert len(contexts) == 2
+    assert contexts[0] == contexts[1]
+
+
+class SlowSandbox(RecordingSandbox):
+    async def read_file(self, path: str, **kwargs: Any) -> bytes:
+        await asyncio.sleep(0.15)
+        return await super().read_file(path, **kwargs)
+
+
+@workflow.defn
+class ConcurrentSandboxWorkflow:
+    @workflow.run
+    async def run(self) -> list[bytes]:
+        sandbox = TemporalSandbox(
+            "recording", start_to_close_timeout=timedelta(seconds=15)
+        )
+        return list(
+            await asyncio.gather(
+                sandbox.read_file("/binary"), sandbox.read_file("/binary")
+            )
+        )
+
+
+async def test_sandbox_factory_is_single_flight_and_not_evicted_in_use(
+    client: Client,
+):
+    task_queue = f"test_sandbox_single_flight-{uuid4()}"
+    contexts: list[SandboxWorkflowContext] = []
+
+    async def factory(context: SandboxWorkflowContext) -> SlowSandbox:
+        contexts.append(context)
+        await asyncio.sleep(0.05)
+        return SlowSandbox()
+
+    plugin = StrandsPlugin(
+        models={},
+        sandboxes={"recording": factory},
+        sandbox_cache_idle_timeout=timedelta(milliseconds=25),
+    )
+    async with Worker(
+        client,
+        task_queue=task_queue,
+        workflows=[ConcurrentSandboxWorkflow],
+        plugins=[plugin],
+        max_cached_workflows=0,
+    ):
+        result = await client.execute_workflow(
+            ConcurrentSandboxWorkflow.run,
+            id=f"test_sandbox_single_flight-{uuid4()}",
+            task_queue=task_queue,
+        )
+
+    assert result == [b"\x00\xff", b"\x00\xff"]
+    assert len(contexts) == 1
+
+
+def test_sandbox_cache_idle_timeout_must_be_positive() -> None:
+    with pytest.raises(ValueError, match="must be positive"):
+        StrandsPlugin(
+            models={},
+            sandboxes={"recording": lambda _: RecordingSandbox()},
+            sandbox_cache_idle_timeout=timedelta(0),
+        )
+
+
+def test_sandbox_factories_share_one_activity_set() -> None:
+    plugin = StrandsPlugin(
+        models={},
+        sandboxes={
+            "first": lambda _: RecordingSandbox(),
+            "second": lambda _: RecordingSandbox(),
+        },
+    )
+
+    assert plugin.activities is not None
+    assert not callable(plugin.activities)
+    assert len(plugin.activities) == 6
+
+
+def test_sandbox_workflow_context_separates_run_and_chain_identity() -> None:
+    chain = SandboxWorkflowChain("namespace", "workflow", "first-run")
+    first = SandboxWorkflowContext(chain, "run-1")
+    second = SandboxWorkflowContext(chain, "run-2")
+
+    assert first != second
+    assert first.chain == second.chain
+    assert first.namespace == "namespace"
+    assert first.workflow_id == "workflow"
+    assert first.first_execution_run_id == "first-run"
+
+
+@tool(name="sandbox_bash")
+def custom_bash(command: str) -> str:
+    return command
+
+
+def test_sandbox_default_tools_and_override() -> None:
+    default_agent = TemporalAgent(
+        model="mock",
+        sandbox=TemporalSandbox("recording"),
+        start_to_close_timeout=timedelta(seconds=15),
+    )
+    assert "sandbox_bash" in default_agent.tool_registry.registry
+    assert "sandbox_file_editor" in default_agent.tool_registry.registry
+
+    override_agent = TemporalAgent(
+        model="mock",
+        sandbox=TemporalSandbox("recording"),
+        tools=[custom_bash],
+        start_to_close_timeout=timedelta(seconds=15),
+    )
+    assert override_agent.tool_registry.registry["sandbox_bash"] is custom_bash
+    assert "sandbox_file_editor" in override_agent.tool_registry.registry
+
+
+@workflow.defn
+class StreamingSandboxWorkflow:
+    def __init__(self) -> None:
+        self.stream = WorkflowStream()
+
+    @workflow.run
+    async def run(self) -> bool:
+        sandbox = TemporalSandbox(
+            "recording",
+            start_to_close_timeout=timedelta(seconds=15),
+            streaming_topic="sandbox-events",
+        )
+        result = [item async for item in sandbox.execute_streaming("echo hi")]
+        return result[-1] == ExecutionResult(
+            0,
+            "out",
+            "err",
+            [OutputFile("artifact.bin", b"\x80\xff")],
+        )
+
+
+async def test_sandbox_streaming_publishes_raw_chunks(client: Client):
+    task_queue = f"test_sandbox_streaming-{uuid4()}"
+    workflow_id = f"test_sandbox_streaming-{uuid4()}"
+    plugin = StrandsPlugin(
+        models={}, sandboxes={"recording": lambda _: RecordingSandbox()}
+    )
+    async with Worker(
+        client,
+        task_queue=task_queue,
+        workflows=[StreamingSandboxWorkflow],
+        plugins=[plugin],
+        max_cached_workflows=0,
+    ):
+        handle = await client.start_workflow(
+            StreamingSandboxWorkflow.run,
+            id=workflow_id,
+            task_queue=task_queue,
+        )
+        stream = WorkflowStreamClient.create(client, workflow_id)
+        events: list[StreamChunk] = []
+
+        async def collect() -> None:
+            async for stream_item in stream.subscribe(
+                ["sandbox-events"],
+                result_type=StreamChunk,
+                poll_cooldown=timedelta(milliseconds=50),
+            ):
+                events.append(stream_item.data)
+                if len(events) == 2:
+                    break
+
+        collect_task = asyncio.create_task(collect())
+        assert await handle.result()
+        await asyncio.wait_for(collect_task, timeout=10)
+
+    assert events == [
+        StreamChunk("out"),
+        StreamChunk("err", "stderr"),
+    ]
+    await Replayer(
+        workflows=[StreamingSandboxWorkflow], plugins=[plugin]
+    ).replay_workflow(await handle.fetch_history())
+
+
+class ErrorSandbox(RecordingSandbox):
+    def __init__(self, *, always_timeout: bool = False) -> None:
+        super().__init__()
+        self.attempts = 0
+        self.always_timeout = always_timeout
+
+    async def execute_streaming(
+        self,
+        command: str,
+        *,
+        timeout: float | None = None,
+        cwd: str | None = None,
+        env: dict[str, str] | None = None,
+        **kwargs: Any,
+    ) -> AsyncGenerator[StreamChunk | ExecutionResult, None]:
+        self.attempts += 1
+        if self.always_timeout:
+            # The backing sandbox enforces its own limit, not the requested one.
+            raise SandboxTimeoutError(90)
+        if self.attempts == 1:
+            raise RuntimeError("transient")
+        yield ExecutionResult(0, "retried", "")
+
+    async def read_file(self, path: str, **kwargs: Any) -> bytes:
+        raise FileNotFoundError(f"cat: {path}: No such file or directory")
+
+    async def list_files(self, path: str, **kwargs: Any) -> list[FileInfo]:
+        raise SandboxPathNotFoundError(path)
+
+
+@workflow.defn
+class SandboxErrorWorkflow:
+    @workflow.run
+    async def run(self) -> tuple[str, bool, bool, str, str]:
+        retried = TemporalSandbox(
+            "retried",
+            start_to_close_timeout=timedelta(seconds=15),
+            retry_policy=RetryPolicy(
+                initial_interval=timedelta(milliseconds=1), maximum_attempts=3
+            ),
+        )
+        result = await retried.execute("command", timeout=3)
+        try:
+            await retried.list_files("/missing")
+        except SandboxPathNotFoundError:
+            path_error = True
+        else:
+            path_error = False
+
+        # A plain FileNotFoundError from the sandbox arrives as the sandbox
+        # subclass, keeping the backing environment's own message.
+        try:
+            await retried.read_file("/missing")
+        except SandboxPathNotFoundError as err:
+            read_message = str(err)
+        else:
+            read_message = ""
+
+        # No retry policy: a timeout must surface on the first attempt rather
+        # than retrying under Temporal's unlimited-attempt default. The
+        # schedule-to-close timeout bounds the failure if that ever regresses.
+        failing = TemporalSandbox(
+            "failing",
+            start_to_close_timeout=timedelta(seconds=5),
+            schedule_to_close_timeout=timedelta(seconds=15),
+        )
+        try:
+            await failing.execute("command", timeout=4)
+        except SandboxTimeoutError as err:
+            timeout_error = True
+            timeout_message = str(err)
+        else:
+            timeout_error = False
+            timeout_message = ""
+        return result.stdout, path_error, timeout_error, read_message, timeout_message
+
+
+async def test_sandbox_retries_and_reconstructs_errors(client: Client):
+    task_queue = f"test_sandbox_errors-{uuid4()}"
+    retried = ErrorSandbox()
+    failing = ErrorSandbox(always_timeout=True)
+    factory_attempts = 0
+
+    def retried_factory(_: SandboxWorkflowContext) -> ErrorSandbox:
+        nonlocal factory_attempts
+        factory_attempts += 1
+        if factory_attempts == 1:
+            raise RuntimeError("transient factory failure")
+        return retried
+
+    plugin = StrandsPlugin(
+        models={},
+        sandboxes={"retried": retried_factory, "failing": lambda _: failing},
+    )
+    async with Worker(
+        client,
+        task_queue=task_queue,
+        workflows=[SandboxErrorWorkflow],
+        plugins=[plugin],
+        max_cached_workflows=0,
+    ):
+        result = await client.execute_workflow(
+            SandboxErrorWorkflow.run,
+            id=f"test_sandbox_errors-{uuid4()}",
+            task_queue=task_queue,
+        )
+
+    assert result == (
+        "retried",
+        True,
+        True,
+        "cat: /missing: No such file or directory",
+        "Execution timed out after 90 seconds",
+    )
+    assert factory_attempts == 2
+    assert retried.attempts == 2
+    assert failing.attempts == 1
diff --git a/uv.lock b/uv.lock
index 6e6ccc4f4..7692e7354 100644
--- a/uv.lock
+++ b/uv.lock
@@ -4827,7 +4827,7 @@ requires-dist = [
     { name = "protobuf", specifier = ">=3.20,<8.0.0" },
     { name = "pydantic", marker = "extra == 'pydantic'", specifier = ">=2.0.0,<3" },
     { name = "python-dateutil", marker = "python_full_version < '3.11'", specifier = ">=2.8.2,<3" },
-    { name = "strands-agents", marker = "extra == 'strands-agents'", specifier = ">=1.39.0" },
+    { name = "strands-agents", marker = "extra == 'strands-agents'", specifier = ">=1.47.0" },
     { name = "types-aioboto3", extras = ["s3"], marker = "extra == 'aioboto3'", specifier = ">=10.4.0" },
     { name = "types-protobuf", specifier = ">=3.20,<8.0.0" },
     { name = "typing-extensions", specifier = ">=4.2.0,<5" },
@@ -4876,7 +4876,7 @@ dev = [
     { name = "pytest-xdist", specifier = ">=3.6,<4" },
     { name = "ruff", specifier = ">=0.15.12,<0.16" },
     { name = "setuptools", specifier = "<82" },
-    { name = "strands-agents", specifier = ">=1.39.0" },
+    { name = "strands-agents", specifier = ">=1.47.0" },
     { name = "strands-agents-tools", specifier = ">=0.5.2" },
     { name = "toml", specifier = ">=0.10.2,<0.11" },
     { name = "twine", specifier = ">=4.0.1,<5" },