strands.vended_plugins.context_offloader.storage

Storage backends for offloaded tool result content.

This module defines the Storage protocol and provides three built-in implementations: file-based, in-memory, and S3 storage. Each content block from a tool result is stored individually with its content type preserved.

Example:

from strands.vended_plugins.context_offloader import (
    FileStorage,
    InMemoryStorage,
    S3Storage,
)

# File-based storage
storage = FileStorage(artifact_dir="./artifacts")
ref = storage.store("tool_123_0", b"large output content...", "text/plain")
content, content_type = storage.retrieve(ref)

# In-memory storage (useful for testing and serverless)
storage = InMemoryStorage()

# S3 storage
storage = S3Storage(bucket="my-bucket", prefix="artifacts/")

Storage

@runtime_checkable
class Storage(Protocol)

Defined in: src/strands/vended_plugins/context_offloader/storage.py:61

Backend for storing and retrieving offloaded content blocks.

Each content block from a tool result is stored individually with its content type preserved. The SDK ships three built-in implementations: InMemoryStorage, FileStorage, and S3Storage. Implement this protocol to create custom storage backends (e.g., Redis, DynamoDB).

Lifecycle: This protocol intentionally does not include eviction or deletion methods. Stored content accumulates for the lifetime of the storage instance. For long-running agents, create a new storage instance per session or use a backend with built-in lifecycle management (e.g., S3 lifecycle policies).

store

def store(key: str, content: bytes, content_type: str = "text/plain") -> str

Defined in: src/strands/vended_plugins/context_offloader/storage.py:76

Store content and return a reference identifier.

Arguments:

• key - A unique key for this content block.
• content - The raw content bytes to store.
• content_type - MIME type of the content (e.g., “text/plain”, “application/json”, “image/png”, “application/pdf”).

Returns:

A reference string that can be used to retrieve the content later.

retrieve

def retrieve(reference: str) -> tuple[bytes, str]

Defined in: src/strands/vended_plugins/context_offloader/storage.py:90

Retrieve stored content by reference.

Arguments:

• reference - The reference returned by a previous store() call.

Returns:

A tuple of (content bytes, content type).

Raises:

• KeyError - If the reference is not found.

FileStorage

class FileStorage()

Defined in: src/strands/vended_plugins/context_offloader/storage.py:105

Store offloaded content as files on disk.

Files are written to the configured artifact directory with unique names. File extensions are derived from the content type. A .metadata.json sidecar file tracks content types so they survive process restarts.

Arguments:

• artifact_dir - Directory path where artifact files will be stored.

__init__

def __init__(artifact_dir: str = "./artifacts") -> None

Defined in: src/strands/vended_plugins/context_offloader/storage.py:118

Initialize file-based storage.

Arguments:

• artifact_dir - Directory path where artifact files will be stored.

store

def store(key: str, content: bytes, content_type: str = "text/plain") -> str

Defined in: src/strands/vended_plugins/context_offloader/storage.py:136

Store content as a file and return the path as reference.

The returned path preserves the form of artifact_dir passed to the constructor: a relative artifact_dir yields a relative reference, an absolute one yields an absolute reference.

Arguments:

• key - A unique key for this content block.
• content - The raw content bytes to store.
• content_type - MIME type of the content.

Returns:

The file path (e.g., ./artifacts/1234_1_key.txt).

retrieve

def retrieve(reference: str) -> tuple[bytes, str]

Defined in: src/strands/vended_plugins/context_offloader/storage.py:168

Retrieve content from a stored file.

Accepts both full paths (as returned by store()) and bare filenames for backward compatibility.

Arguments:

• reference - The file path or filename returned by store().

Returns:

A tuple of (content bytes, content type).

Raises:

• KeyError - If the file does not exist.

InMemoryStorage

class InMemoryStorage()

Defined in: src/strands/vended_plugins/context_offloader/storage.py:213

Store offloaded content in memory.

Useful for testing and serverless environments where disk access is not available or not desired. Thread-safe.

Supports turn-based eviction: entries not accessed (stored or retrieved) within evict_after_turns agent loop cycles are automatically removed. The ContextOffloader plugin triggers eviction on each model invocation cycle. Eviction is enabled by default (20 cycles). Pass None to disable.

Notes:

Content does not survive process restarts. For multi-session persistence, use FileStorage or S3Storage. Each agent should use its own InMemoryStorage instance — sharing one across multiple agents is not supported when eviction is enabled.

Evicted entries are permanently deleted from memory. The agent will receive an error if it attempts to retrieve evicted content. The original tool result is not preserved in the conversation history after offloading — only the preview and references remain in context.

Arguments:

• evict_after_turns - Number of cycles of inactivity before an entry is evicted. Defaults to 20. None disables eviction.

__init__

def __init__(
        evict_after_turns: int | None = _DEFAULT_EVICT_AFTER_TURNS) -> None

Defined in: src/strands/vended_plugins/context_offloader/storage.py:242

Initialize in-memory storage.

Arguments:

• evict_after_turns - Number of cycles of inactivity before an entry is evicted. Defaults to 20. None disables eviction.

Raises:

• ValueError - If evict_after_turns is not a positive integer.

store

def store(key: str, content: bytes, content_type: str = "text/plain") -> str

Defined in: src/strands/vended_plugins/context_offloader/storage.py:262

Store content in memory and return a reference.

Arguments:

• key - A unique key for this content block.
• content - The raw content bytes to store.
• content_type - MIME type of the content.

Returns:

A unique reference string.

retrieve

def retrieve(reference: str) -> tuple[bytes, str]

Defined in: src/strands/vended_plugins/context_offloader/storage.py:279

Retrieve content from memory.

Refreshes the last-accessed turn so the entry stays alive longer when eviction is enabled.

Arguments:

• reference - The reference returned by store().

Returns:

A tuple of (content bytes, content type).

Raises:

• KeyError - If the reference is not found (or was evicted).

clear

def clear() -> None

Defined in: src/strands/vended_plugins/context_offloader/storage.py:339

Remove all stored content.

Call this to free memory when offloaded results are no longer needed, e.g., between sessions or after an invocation completes.

S3Storage

class S3Storage()

Defined in: src/strands/vended_plugins/context_offloader/storage.py:349

Store offloaded content in Amazon S3.

Objects are stored with unique keys under the configured prefix. Content type is preserved as S3 object metadata.

Arguments:

• bucket - S3 bucket name.
• prefix - S3 key prefix for organizing stored artifacts.
• boto_session - Optional boto3 session. If not provided, a new session is created using the given region_name.
• boto_client_config - Optional botocore client configuration.
• region_name - AWS region. Used only when boto_session is not provided.

Example:

from strands.vended_plugins.context_offloader import S3Storage

storage = S3Storage(
    bucket="my-agent-artifacts",
    prefix="tool-results/",
)

__init__

def __init__(bucket: str,
             prefix: str = "",
             boto_session: boto3.Session | None = None,
             boto_client_config: BotocoreConfig | None = None,
             region_name: str | None = None) -> None

Defined in: src/strands/vended_plugins/context_offloader/storage.py:374

Initialize S3-based storage.

Arguments:

• bucket - S3 bucket name.
• prefix - S3 key prefix for organizing stored artifacts.
• boto_session - Optional boto3 session. If not provided, a new session is created using the given region_name.
• boto_client_config - Optional botocore client configuration.
• region_name - AWS region. Used only when boto_session is not provided.

store

def store(key: str, content: bytes, content_type: str = "text/plain") -> str

Defined in: src/strands/vended_plugins/context_offloader/storage.py:410

Store content as an S3 object and return an s3:// URI as reference.

Arguments:

• key - A unique key for this content block.
• content - The raw content bytes to store.
• content_type - MIME type of the content.

Returns:

An S3 URI (e.g., s3://bucket/prefix/1234_1_key).

Raises:

• botocore.exceptions.ClientError - If the S3 operation fails (e.g., bucket does not exist, permission denied).

retrieve

def retrieve(reference: str) -> tuple[bytes, str]

Defined in: src/strands/vended_plugins/context_offloader/storage.py:441

Retrieve content from an S3 object.

Accepts both s3:// URIs (as returned by store()) and raw S3 keys for backward compatibility.

Arguments:

• reference - The S3 URI or object key returned by store().

Returns:

A tuple of (content bytes, content type).

Raises:

• KeyError - If the object does not exist.