feat(py): add unified storage interface

[lizradway]

Description

Ports the unified Storage primitive from PR #3099 (TypeScript SDK) to the Python SDK — a single persistence interface (write/read/delete/list over opaque bytes) that all SDK subsystems can share. This eliminates per-subsystem persistence re-invention and gives every consumer the same backend options (local file, S3, in-memory, or custom) for free.

The context offloader plugin is adapted to accept both the new unified Storage and the legacy offloader Storage, selecting the code path at runtime via duck-type detection. Unified storage gets content framing (2-byte BE content-type header) and cycle-based eviction; the legacy path is unchanged.

Session integration is deferred — Python's SessionRepository is architecturally different from TS's snapshot storage and deserves its own design.

Public API

New module: strands.storage

from strands.storage import Storage, InMemoryStorage, LocalFileStorage, S3Storage

Storage Protocol (generic on ListQuery, defaults to str)

@runtime_checkable
class Storage(Protocol[ListQuery]):
    async def write(self, key: str, data: bytes) -> None: ...
    async def read(self, key: str) -> bytes | None: ...
    async def delete(self, key: str) -> None: ...
    async def list(self, query: ListQuery) -> list[str]: ...

All keys are opaque, '/'-separated strings. All data is raw bytes. read returns None for missing keys. delete is a no-op for missing keys. list returns keys sorted ascending; empty string matches all.

InMemoryStorage

storage = InMemoryStorage()
await storage.write("key", b"value")
storage.clear()  # testing helper

Thread-safe dict backend. Unbounded — consumers manage eviction.

LocalFileStorage

storage = LocalFileStorage("./.strands/")
await storage.write("sessions/abc/state.json", data)

# Sandbox routing
sandbox_storage = storage.for_sandbox(agent.sandbox)

# Namespacing
scoped = storage.namespace("offloader")

Maps key segments to filesystem paths. Writes are atomic (temp file + os.replace). Supports sandbox routing via for_sandbox() — the returned view routes I/O through the sandbox's file API. namespace() returns a _NamespacedLocalFileStorage that preserves for_sandbox through the namespace layer.

S3Storage

storage = S3Storage(
    "my-bucket",
    prefix="agents/",
    region_name="us-east-1",
    # OR
    boto_session=my_session,
    boto_client_config=Config(read_timeout=30),
)

• Lazy client creation (no import cost until first use)
• All boto3 calls wrapped in asyncio.to_thread (non-blocking)
• Paginated listing via list_objects_v2
• User-agent tagging (strands-agents)
• namespace() for key prefix scoping

Key normalization

All implementations share _normalize_key / _normalize_prefix:

• Collapses // → /, strips leading/trailing /
• Rejects empty keys, .. segments → raises StorageError

Namespace support

scoped = storage.namespace("offloader")
await scoped.write("key", data)
# equivalent to: await storage.write("offloader/key", data)

Composable — storage.namespace("a").namespace("b") scopes to "a/b/". _NAMESPACED sentinel allows SDK constructs to detect pre-scoped storage and skip auto-prefixing.

Context Offloader Changes

The ContextOffloader constructor now accepts storage: Storage | _LegacyStorage:

from strands.storage import LocalFileStorage
from strands.vended_plugins.context_offloader import ContextOffloader

agent = Agent(plugins=[
    ContextOffloader(
        storage=LocalFileStorage("./.strands/"),
        evict_after_cycles=20,
    )
])

New parameters:

• evict_after_cycles: int | None = 20 — entries stored more than N cycles ago are deleted on BeforeModelCallEvent. Set to None to disable.

Internal changes:

• Content framing: [2-byte BE content-type length][content-type UTF-8][content bytes]
• Per-agent eviction tracking via WeakKeyDictionary[Agent, dict[str, int]]
• Duck-type detection for sandbox routing (hasattr(storage, "for_sandbox"))
• Auto-namespaces unified storage under "offloader/" if not already scoped

Developer Experience

# Minimal — in-memory, no persistence
from strands.storage import InMemoryStorage
storage = InMemoryStorage()

# Local file — atomic writes, sandbox-aware
from strands.storage import LocalFileStorage
storage = LocalFileStorage("./.strands/")

# S3 — non-blocking, paginated, lazy client
from strands.storage import S3Storage
storage = S3Storage("my-bucket", prefix="agents/")

# Custom backend — just implement the 4-method Protocol
class MyStorage:
    async def write(self, key: str, data: bytes) -> None: ...
    async def read(self, key: str) -> bytes | None: ...
    async def delete(self, key: str) -> None: ...
    async def list(self, query: str) -> list[str]: ...

# isinstance checks work at runtime
from strands.storage import Storage
assert isinstance(InMemoryStorage(), Storage)  # True (runtime_checkable)

Related Issues

#3099 (TS implementation), #3258 (TS bug fixes ported here)

Type of Change

New feature

Testing

• 90 storage unit tests (InMemory, LocalFile, S3 via moto)
• 13 unified-storage plugin tests (framing, eviction, per-agent scoping)
• All existing offloader tests continue to pass (legacy path unchanged)
• Lint (ruff + mypy strict) passes
• Python 3.10–3.14, Linux + macOS + Windows CI

Checklist

• I have read the CONTRIBUTING document
• I have reviewed and understand every line of code in this PR, including any generated by AI tools, and I can explain why it works
• My change is focused and reasonably small; I have split unrelated work into separate PRs
• I have added any necessary tests that prove my fix is effective or my feature works
• I have updated the documentation accordingly
• I have added an appropriate example to the documentation to outline the feature, or no new docs are needed
• My changes generate no new warnings
• Any dependent changes have been merged and published

---

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of your choice.

[github-actions]

Assessment: Request Changes

Clean, well-documented storage primitive with solid coverage of the happy paths, key normalization, and path-traversal rejection. The main concerns are around the interaction between per-agent (sandbox-bound) storage and the new eviction bookkeeping, binary-safety on the sandbox path, and API-review process — details are in the inline comments.

Review Categories

• Correctness: Eviction bookkeeping and the sandbox file path don't fully account for per-agent, sandbox-bound backends and non-UTF-8 binary payloads (images/documents), which the offloader stores.
• Code quality: S3Storage.read has a convoluted/effectively-dead not-found fallback; async methods make blocking boto3 calls.
• Robustness: Frame decoding and best-effort eviction silently absorb corruption/errors.
• Testing: Strong overall; one test uses a deprecated event-loop pattern, and per-agent eviction/binary round-tripping aren't covered.
• API bar raising: This adds a public API surface (strands.storage, Storage, three impls) but lacks the needs-api-review/completed-api-review label and the structured API doc (signatures + defaults, exports, use cases) in the description. If the parent TS PR feat: add unified storage interface #3099 already cleared API review, linking that would help; otherwise it should go through review.

Nice work porting this over — the interface is minimal and the docstrings are genuinely helpful.

[codecov]

Codecov Report

❌ Patch coverage is 95.25000% with 19 lines in your changes missing coverage. Please review.

Files with missing lines	Patch %	Lines
...rands-py/src/strands/storage/local_file_storage.py	89.84%	10 Missing and 3 partials ⚠️
...strands/vended_plugins/context_offloader/plugin.py	94.62%	2 Missing and 3 partials ⚠️
strands-py/src/strands/storage/s3_storage.py	98.88%	0 Missing and 1 partial ⚠️

📢 Thoughts on this report? Let us know!

[github-actions]

Issue (Testing): Codecov reports 69.76% patch coverage with 117 uncovered lines — concentrated exactly in the risk areas: local_file_storage.py (59%, the sandbox I/O paths) and plugin.py (48%, the new unified-storage eviction/framing path). These are the branches most likely to regress silently.

Suggestion: Add tests for (1) the sandbox-bound LocalFileStorage write/read/list/delete round-trip (with binary bytes), and (2) cycle-based eviction against a unified Storage, including the multi-agent sharing case flagged above. That would cover the newly-added behavior rather than just the host-filesystem happy path.

[github-actions]

Assessment: Comment (re-review)

Thanks for the quick turnaround — the substantive items from the first pass are addressed: the sandbox path now round-trips raw bytes, S3Storage.read has a clean NoSuchKey check, boto3 calls are off the event loop via asyncio.to_thread, test_clear is async, and eviction now deletes through the per-agent storage.

Remaining (non-blocking) items

• Multi-agent eviction: _stored_cycles is still a single shared dict, so a plugin shared across agents with sandbox-bound storage can drop another agent's tracking entry / delete against the wrong backend. Scoping it per agent would close the loop.
• Frame validation: _unframe_content slices on an untrusted length header without bounds checks.
• Coverage: patch coverage is 69.76% — the sandbox I/O and unified-eviction paths are largely untested.
• API review: still no needs-api-review/completed-api-review label for this public strands.storage surface; linking the TS review (feat: add unified storage interface #3099) would resolve it.

None of these are blockers on their own — the core design is solid.

[github-actions]

Assessment: Approve (pending API-review process)

Re-reviewed at a940835. Every code item from the previous rounds has been addressed cleanly:

Resolved

• Multi-agent eviction — _stored_cycles is now a WeakKeyDictionary[Agent, dict[str, int]] and eviction reads agent_cycles = self._stored_cycles.get(event.agent), so tracking/eviction are isolated per agent. 👍
• Frame validation — _unframe_content now checks len(frame) < 2 and len(frame) < 2 + ct_len and raises an explicit ValueError (with a documented Raises).
• Silent eviction errors — failures now log key=<%s> | failed to evict stale entry at debug instead of pass.
• Binary safety / S3 read / blocking boto3 / async test — all fixed in earlier pushes.
• Coverage — patch coverage went from 69.76% → 94.77% (local file 90%, plugin 92%, S3 99%).
• Bonus: the new _resolve_storage auto-namespacing under offloader is a nice isolation improvement.

Sole remaining item (process, not code): this PR still lacks a needs-api-review/completed-api-review label despite introducing the public strands.storage surface (Storage, InMemoryStorage, LocalFileStorage, S3Storage). Since it's a port of the already-reviewed TS PR #3099, linking that review in the description and applying completed-api-review should be enough to close this out.

Great iteration — the design and test coverage are in solid shape now.

[github-actions]

Re-review @ 8e8b4d2 — Assessment: Approve (pending API-review label)

Reviewed the harden namespace prefix handling and eviction logic commit. All four changes are solid and I verified them locally (93 storage + 160 offloader tests pass):

• Trailing-slash normalization (_normalize_prefix(prefix).rstrip("/")) fixes the "sessions//" double-slash corruption, and test_trailing_slash_prefix_does_not_corrupt_keys locks in the regression. 👍
• Eviction retry-on-failure — tracking entries are now only dropped after a successful delete() (via continue), so transient backend failures retry next cycle instead of orphaning data. This is stricter than the version I previously approved — nice catch.
• Collapsing _NamespacedLocalFileStorage into a generic _NamespacedStorage.for_sandbox removes a single-purpose subclass with no loss of coverage (test_namespace_preserves_for_sandbox still exercises the delegation path).
• _storage_for_agent simplification — I confirmed legacy FileStorage still exposes for_sandbox, so dropping the isinstance(FileStorage) special-case is safe; the hasattr check subsumes it with no behavior change.

No new code concerns. The only outstanding item remains the process one: applying completed-api-review (or linking TS PR #3099) for the public strands.storage surface.

[github-actions]

Re-review @ 8fb5d87 — Assessment: Approve (pending API-review label)

Reviewed remove stale type: ignore comments caught by mypy. This is a trivial, safe cleanup:

• Removed the now-unused # type: ignore[union-attr] on _NamespacedStorage.for_sandbox (storage.py) and _storage_for_agent (plugin.py) — I confirmed mypy reports no union-attr errors on those lines, so the suppressions were genuinely stale.
• Remaining diff is pure line-length reformatting (collapsing wrapped statements) with no behavior change.

253 storage + offloader tests still pass. No new concerns — the code side has been in solid shape for several rounds now, so the only thing gating merge is the completed-api-review label (or linking TS PR #3099) for the public strands.storage surface.