Using DynamoDB as a storage backend for Strands Agents

Strands Agents is an open source SDK that takes a model-driven approach to building AI agents in a few lines of code, with first-class support for Amazon Bedrock and other model providers. Among its building blocks is a unified storage interface: a byte-oriented contract (write, read, delete, list) that every stateful subsystem in the SDK speaks. The Session Manager persists conversation snapshots through it, the Memory Manager stores long-term memories through it, and the context offloader and transcripts use the same operations.

The strands-dynamodb-storage package implements that contract on a single DynamoDB table, for both Python and TypeScript. The SDK's /-separated storage keys map directly onto the DynamoDB key model: a key's first two segments become the partition key and the remainder becomes the sort key, so point operations are single-item calls and listing a prefix is a native partition Query, never a table scan.

Key features

Prerequisites

Installation

Install the package from PyPI:

pip install strands-dynamodb-storage "strands-agents>=1.48.0"

Or from npm for TypeScript (the TypeScript package is a feature-parity mirror):

npm install strands-dynamodb-storage

Create the table

The package holds no CreateTable permission and never creates infrastructure: you create the table in advance, with your own tagging, backup, and encryption settings applied. A table with a string partition key pk and a string sort key sk is the only requirement:

aws dynamodb create-table \
  --table-name agent-storage \
  --attribute-definitions AttributeName=pk,AttributeType=S AttributeName=sk,AttributeType=S \
  --key-schema AttributeName=pk,KeyType=HASH AttributeName=sk,KeyType=RANGE \
  --billing-mode PAY_PER_REQUEST

If you plan to use semantic memory, declare the vector index in this same command: an index's name, dimensions, and distance function cannot be changed after creation, so size the dimensions to your embedding model. The package README shows the full call that declares the index.

Persist agent sessions

Hand the storage to your agent's session manager. The SDK namespaces the keys, snapshots the conversation on every invocation, and restores it when the same session returns:

from strands import Agent
from strands.session import SnapshotSessionManager
from strands_dynamodb_storage import DynamoDBStorage

storage = DynamoDBStorage("agent-storage", region_name="us-east-1")
session = SnapshotSessionManager(session_id="user-42", storage=storage)

agent = Agent(session_manager=session)
agent("Where did we leave off?")

You can also set storage once on the Agent itself. Every subsystem that accepts a storage then inherits it, each namespaced under its own key prefix, so one table carries the whole agent's state:

from strands.vended_plugins.context_offloader import ContextOffloader

agent = Agent(
    storage=storage,
    session_manager=SnapshotSessionManager(),  # persists under session/
    plugins=[ContextOffloader()],              # offloads oversized tool results under offloader/
)

The byte contract is also available directly. The contract is async: inside an agent the SDK drives it for you, and in a plain script you wrap the calls in asyncio.run:

import asyncio


async def main():
    await storage.write("session/user-42/notes", b"prefers aisle seats")
    keys = await storage.list("session/user-42/")


asyncio.run(main())

Semantic long-term memory with vector indexes

Session persistence solves half of the memory problem: your agent survives a restart and resumes the conversation. The harder half is recalling something a user told the agent weeks ago, in a new conversation that shares no keys with the old one, and that requires searching memories by meaning rather than by key. DynamoDB vector indexes bring nearest-neighbor search to the same table that holds your agent's state (see Using vector indexes in DynamoDB).

The following examples embed text with Amazon Titan Text Embeddings V2 through Amazon Bedrock. The model returns 1,024-dimension vectors by default, so the index for these examples is created with 1,024 dimensions. The following code defines the embed() function that the remaining examples use:

import json

import boto3

bedrock = boto3.client("bedrock-runtime", region_name="us-east-1")


def embed(text):
    response = bedrock.invoke_model(
        modelId="amazon.titan-embed-text-v2:0",
        body=json.dumps({"inputText": text, "dimensions": 1024}),
    )
    return json.loads(response["body"].read())["embedding"]

Writing a memory attaches an embedding and optional metadata alongside the bytes. The following code constructs the store with a per-tenant prefix, so the key memories/m1 lands in the physical partition user/u1:

from strands_dynamodb_storage import DynamoDBStorage, SearchQuery

storage = DynamoDBStorage("agent-storage", region_name="us-east-1", prefix="user/u1")

await storage.write(
    "memories/m1",
    b"prefers window seats on long flights",
    vector=embed("prefers window seats on long flights"),
    metadata={"kind": "preference"},
)

Recalling by meaning is one call, scoped to the same partition:

results = await storage.search(SearchQuery(
    vector=embed("what are this user's seating preferences?"),
    top_k=5,
    pk="user/u1",  # the physical partition: the full key's first two segments
    filter={"kind": "preference"},
))

Notice the pk argument. The vector index is partitioned the same way the table is, and every search is scoped to one partition, so a tenant's search never ranges over another tenant's memories, and the work each search performs tracks the size of that tenant's memory rather than the whole table. Keep in mind that the partition value is supplied by the caller, so this is query scoping rather than an authorization boundary: a principal holding dynamodb:SearchVectors on the table can search any partition, and tenant access control belongs in IAM and your application layer.

Results return most similar first. The raw score's direction follows the index's distance function: lower is nearer for cosine and Euclidean distance, and higher is more similar for dot product.

Wire memory into an agent

In a real agent you want retrieved memories to reach the model automatically, and the SDK's Memory Manager handles that: it retrieves relevant entries before each model call and folds them into the model input, and it registers a search_memory tool the model can call on demand. The Memory Manager accepts any object implementing the SDK's MemoryStore protocol. The package does not ship one, so you define a small class in your own application that embeds on write, embeds on search, and returns MemoryEntry values:

import uuid

from strands.memory import MemoryEntry
from strands_dynamodb_storage import DynamoDBStorage, SearchQuery


class DynamoDBMemoryStore:
    def __init__(self, storage, partition, embed):
        self.storage = storage
        self.partition = partition
        self.embed = embed  # the embed() function defined earlier
        self.name = "dynamodb"
        self.description = "Long-term memories in DynamoDB, searched by meaning"
        self.max_search_results = 3
        self.writable = True
        self.extraction = None

    async def add(self, content, metadata=None):
        await self.storage.write(
            f"memories/{uuid.uuid4().hex[:8]}",
            content.encode(),
            vector=self.embed(content),
            metadata=metadata,
        )

    async def search(self, query, options=None):
        results = await self.storage.search(SearchQuery(
            vector=self.embed(query),
            top_k=self.max_search_results,
            pk=self.partition,
            include_values=True,
        ))
        return [
            MemoryEntry(content=r.data.decode(), metadata=r.metadata)
            for r in results
            if r.data is not None
        ]

The following code seeds three memories and wires the store into the agent, so retrieved memories reach the model with no orchestration code on your side:

import asyncio

from strands import Agent
from strands.memory import MemoryManager

storage = DynamoDBStorage("agent-storage", region_name="us-east-1", prefix="user/u1")
store = DynamoDBMemoryStore(storage, partition="user/u1", embed=embed)


async def seed():
    await store.add("Prefers window seats on long flights")
    await store.add("Planning a trip to Tokyo in December")
    await store.add("Allergic to peanuts")


asyncio.run(seed())

memory = MemoryManager(stores=[store], add_tool_config=True)
agent = Agent(memory_manager=memory)
agent("Book me a flight seat for my December trip. Which seat should I pick?")

Running this against a live table, the agent calls its search_memory tool, matches the Tokyo trip and the window-seat preference from DynamoDB, and recommends a window seat for the flight. Passing add_tool_config=True also registers an add_memory tool, so the model can store new facts through the same table it recalls from.

Required IAM permissions

At runtime the package issues four DynamoDB operations, plus SearchVectors when you use semantic search, and Amazon S3 operations only when you configure offload, so the least-privilege IAM policy is short. Replace 111122223333 with your AWS account ID and update the Region to match your environment:

{
  "Version": "2012-10-17",
  "Statement": [
    {
      "Effect": "Allow",
      "Action": [
        "dynamodb:PutItem",
        "dynamodb:GetItem",
        "dynamodb:DeleteItem",
        "dynamodb:Query"
      ],
      "Resource": "arn:aws:dynamodb:us-east-1:111122223333:table/agent-storage"
    }
  ]
}

The package README carries the complete policy, including the additional statements for semantic search, Amazon S3 offload, and embedding model invocation.

Considerations

Additional resources