[Strands Agents] Storage インターフェースで DynamoDB バックエンドを自作して、会話履歴と退避データを 1 テーブルにまとめてみました

1 はじめに

製造ビジネステクノロジー部の平内（SIN）です。

Strands Agents に、SDK 全体で共通利用できる Storage インターフェース（unified Storage）が追加されました。それまで Context Offloader プラグインが独自に持っていたストレージ機構が、write / read / delete / list の 4 メソッドだけの共通インターフェースとして切り出され、組み込みで InMemoryStorage / LocalFileStorage / S3Storage の 3 実装が提供されています。

組み込みバックエンドの基本的な使い方については、下記の記事で詳しく紹介されています。

• Strands Agents に統一 Storage インターフェースが追加されました

そこで本記事では上記を参考にさせていただいて、インターフェースが統一されたからこそ可能になった使い方を試してみました。具体的には次の 3 点です。

1. 公式には存在しない DynamoDB バックエンドを自作する（実装するのはたった 4 メソッド）
2. 自作した 1 つのバックエンドを SessionManager と Context Offloader で共有し、会話履歴と退避データを 1 つの DynamoDB テーブルで管理する
3. 別プロセスから同じセッションを再開し、前のプロセスが退避した大容量データをエージェント自身に検索させる

サンプルコードは、下記のリポジトリに置きました。

GitHub: strands-agents-dynamodb-storage

2 統一 Storage インターフェースとは

(1) 4 メソッドの契約

TypeScript 版の Storage インターフェースは以下のとおりです（storage.ts）。

export interface Storage<ListQuery = string> {
  write(key: string, data: Uint8Array): Promise<void>   // 上書き保存
  read(key: string): Promise<Uint8Array | null>         // 無ければ null
  delete(key: string): Promise<void>                    // 無くても no-op
  list(query: ListQuery): Promise<string[]>             // プレフィックス一致、フルキーを昇順で返す
  namespace?(prefix: string): Storage                   // optional
}

扱う値は生のバイト列（Uint8Array）のみ、キーは session/abc/... のような / 区切りのパス風文字列です。この 4 メソッドを実装するだけで、SDK のプラグインが利用できるカスタムバックエンドになります。

(2) 自動ネームスペースが「共有」を可能にする

このインターフェースの面白いところは、SessionManager と ContextOffloader という 2 つの機能に同じ保存先を渡しても衝突しない点です。SDK 側に「未スコープの Storage を受け取ったら自動でプレフィックスを付ける」仕組みが組み込まれており、ソースコードでは次のように処理されています。

• SessionManager: session/ を自動付与（session-manager.ts の _resolveSnapshotStorage）
• ContextOffloader: offloader/ を自動付与（plugin.ts）

つまり「バックエンドを 1 回選ぶだけで、SDK 内の永続化がすべてそこに乗る」という設計です。この自動プレフィックスは SDK 側のラッパーが行うため、自作バックエンドに namespace()（(1) のインターフェースでは optional）を実装する必要はありません。実装すれば、キー体系を自分で制御することもできます。

なお、本記事で扱うのは SessionManager と ContextOffloader の 2 つですが、この Storage を共有できる仕組みはこれらに限りません。SDK のソース（storage/index.ts）のコメントによれば、セッション・メモリ・コンテキストのオフロード・トランスクリプトなど、永続化を必要とする複数のサブシステムが同じインターフェースを利用できます。

(3) TypeScript 版を使う理由

なお、本記事の構成が成立するのは 2026 年 8 月 3 日時点では TypeScript 版のみです。TypeScript 版の SessionManager は unified Storage を直接受け取れますが、Python 版の SessionManager は従来どおり SessionRepository ベースで、unified Storage を渡せるのは ContextOffloader だけでした（harness-sdk の strands-py/src/strands/session/ で確認）。このため、本記事は TypeScript（@strands-agents/sdk 1.11.2）で実装しています。

3 DynamoDB バックエンドの設計

(1) キー設計

Storage のキーはパス風文字列、list はプレフィックス一致という契約なので、DynamoDB のテーブルは次のように設計しました。

sk にフルキーをそのまま入れるのがポイントです。list('offloader/') のようなプレフィックス検索は、pk の完全一致 + begins_with(sk, prefix) の Query で実現でき、Scan に頼らず効率的に一覧できます。また、Query の結果はソートキー順（昇順）で返るため、「昇順で返す」という list の契約とも相性が良い形になります。

(2) TTL と 400KB 上限

書き込み時に ttl 属性（本サンプルでは 24 時間後）を付与し、DynamoDB の TTL 機能で自動削除させます。検証データの消し忘れがなくなります。

この TTL は Storage インターフェースの機能ではなく、DynamoDB ネイティブの TTL をバックエンド実装の内部で利用しているものです。そしてこれはサンプル限定の便宜ではなく、実運用でも妥当な設計となりえます。会話セッションや退避データは一時的な性格のものが多く、「一定期間経ったら自動で消す」という要件は現実によくあるためです（TTL が不要なら ttlSeconds を渡さなければ無効になります）。

一方で、DynamoDB のアイテムは 400KB が上限です。Context Offloader はツール結果 1 ブロックを 1 キーとして保存するため、本サンプルの退避データ（42KB）は問題ありませんが、400KB を超えるツール結果を扱う場合は S3Storage の方が適しています。逆に言えば、用途に応じてバックエンドを差し替えられること自体が unified Storage の利点です。

4 実装

Github dynamodb-storage.ts

実装は 90 行ほどです。以下、メソッドごとに設計意図を説明します。

(1) クラスの骨組みとキーの分解

まず、implements Storage を宣言してクラスを定義します。コンストラクタでは DynamoDB クライアントとテーブル名、そして TTL の秒数を受け取るだけです。

export class DynamoDBStorage implements Storage {
  private readonly client: DynamoDBClient
  private readonly tableName: string
  private readonly ttlSeconds?: number

  constructor(tableName: string, options: DynamoDBStorageOptions = {}) {
    this.tableName = tableName
    this.ttlSeconds = options.ttlSeconds
    this.client = new DynamoDBClient({ region: options.region })
  }

  // 'session/abc/...' → pk='session', sk='session/abc/...'
  private keyOf(key: string) {
    return { pk: { S: key.split('/')[0] }, sk: { S: key } }
  }

この実装の要になるのが keyOf です。SDK から渡ってくるキーは session/demo-session-001/... や offloader/tooluse_xxx_0 のような / 区切りの文字列なので、その先頭セグメント（session や offloader）をパーティションキー（pk）に、フルキーをそのままソートキー（sk）に割り当てます。write / read / delete の 3 つはこの keyOf でアイテムを一意に特定でき、list は「同じ pk の中を sk のプレフィックスで絞り込む」という DynamoDB が得意とする形に落とし込めます。

(2) write: TTL を付けて上書き保存する

write は、keyOf で作った pk / sk に、バイト列を data（バイナリ型 B）として加えて PutItem するだけです。PutItem は同一キーがあれば上書きするので、Storage の「上書き保存」という契約とそのまま一致します。

  async write(key: string, data: Uint8Array): Promise<void> {
    const item: Record<string, any> = { ...this.keyOf(key), data: { B: data } }
    if (this.ttlSeconds !== undefined) {
      item.ttl = { N: String(Math.floor(Date.now() / 1000) + this.ttlSeconds) }
    }
    await this.client.send(new PutItemCommand({ TableName: this.tableName, Item: item }))
  }

ポイントは ttl 属性です。コンストラクタで ttlSeconds が渡されているときだけ、「現在のエポック秒 + 指定秒数」を数値型（N）で付与します。この属性名は CDK 側でテーブルの TTL 属性として指定した ttl と一致させる必要があります（第 5 章）。これにより、書き込まれたアイテムは指定時刻を過ぎると DynamoDB 側で自動削除されます。

(3) read / delete: null と冪等性

read は GetItem の結果からバイナリ値を取り出し、アイテムが存在しなければ null を返します。Storage の契約は「無ければ例外ではなく null」なので、?. と ?? null でそのまま表現できます。

  async read(key: string): Promise<Uint8Array | null> {
    const res = await this.client.send(
      new GetItemCommand({ TableName: this.tableName, Key: this.keyOf(key) })
    )
    return res.Item?.data?.B ?? null
  }

  async delete(key: string): Promise<void> {
    await this.client.send(
      new DeleteItemCommand({ TableName: this.tableName, Key: this.keyOf(key) })
    )
  }

delete も DeleteItem を投げるだけです。DynamoDB の DeleteItem は対象キーが存在しなくてもエラーにならないため、Storage の「無くても no-op（冪等）」という契約を追加のコードなしで満たせます。

(4) list: Query と Scan の使い分け

4 メソッドの中で唯一ロジックがあるのが list です。SDK からは list('offloader/') のようにプレフィックスが渡ってきます。プレフィックスに / が含まれていれば先頭セグメント（= pk）が確定するので、pk の完全一致 + begins_with(sk, ...) の Query で効率よく絞り込めます。

  async list(prefix: string): Promise<string[]> {
    const keys: string[] = []
    let startKey: Record<string, any> | undefined
    do {
      const command = prefix.includes('/')
        ? new QueryCommand({    // pk が確定 → Query + begins_with
            TableName: this.tableName,
            KeyConditionExpression: 'pk = :pk AND begins_with(sk, :prefix)',
            ExpressionAttributeValues: {
              ':pk': { S: prefix.split('/')[0] }, ':prefix': { S: prefix },
            },
            ProjectionExpression: 'sk',
            ExclusiveStartKey: startKey,
          })
        : new ScanCommand({     // pk 不明（空文字など）→ Scan にフォールバック
            TableName: this.tableName,
            ProjectionExpression: 'sk',
            ExclusiveStartKey: startKey,
          })
      const res = await this.client.send(command)
      keys.push(...(res.Items ?? []).map((item) => item.sk.S!))
      startKey = res.LastEvaluatedKey    // ページネーション
    } while (startKey)
    return keys.sort()
  }

一方、list('')（全件）のようにプレフィックスに / が無い場合は pk を特定できないため、Scan にフォールバックします。どちらの場合も LastEvaluatedKey が返る限りループしてページネーションし、最後に sort() で昇順に整えて返します。Storage の契約が「フルキーを昇順で返す」なので、この sort() で仕上げています（Query はソートキー順で返るため実運用ではほぼソート済みですが、Scan 経路も含めて明示的に揃えています）。

エラーの再送や細かい最適化は省略した最小実装ですが、implements Storage と書いて 4 メソッドを埋めるだけで、型チェックが契約への適合を確認してくれます。これだけで SDK のプラグインから利用できます。

5 セットアップ

(1) DynamoDB テーブル（CDK）

Github storage-stack.ts

作成する AWS リソースは DynamoDB テーブル 1 つだけです。

const table = new dynamodb.Table(this, 'StorageTable', {
  tableName, // strands-agents-dynamodb-storage-{アカウントID}
  partitionKey: { name: 'pk', type: dynamodb.AttributeType.STRING },
  sortKey: { name: 'sk', type: dynamodb.AttributeType.STRING },
  billingMode: dynamodb.BillingMode.PAY_PER_REQUEST,
  timeToLiveAttribute: 'ttl',
  removalPolicy: cdk.RemovalPolicy.DESTROY,
})

課金方式には PAY_PER_REQUEST（オンデマンド）を指定しています。キャパシティを予約する PROVISIONED と違い、リクエストが無ければ課金されないため、サンプル用途での待機コストを抑えられます。

(2) デプロイと実行準備

git clone https://github.com/furuya02/strands-agents-dynamodb-storage.git
cd strands-agents-dynamodb-storage

cd cdk && pnpm install && pnpm cdk deploy && cd ..
# Output: TableName = strands-agents-dynamodb-storage-<ACCOUNT_ID>

pnpm install
export TABLE_NAME=strands-agents-dynamodb-storage-<ACCOUNT_ID>

ここで設定した環境変数 TABLE_NAME は、後続のスクリプト（chat1 / chat2 / list-keys）が参照します。

モデルは ap-northeast-1 の Claude Haiku 4.5（jp.anthropic.claude-haiku-4-5-20251001-v1:0）を使用しました。環境変数 MODEL_ID で差し替えできます。

6 デモ 1: 会話履歴と退避データが 1 テーブルに同居する

(1) 同じ Storage を 2 箇所に渡す

Github chat1.ts

自作した DynamoDBStorage のインスタンスを 1 つ作り、SessionManager と ContextOffloader の両方に渡します。エージェントには、大きな結果（300 名分の社員データ、約 42KB）を返すダミーツール get_all_employees を持たせました。

const storage = new DynamoDBStorage(tableName, { region, ttlSeconds: 24 * 60 * 60 })

const agent = new Agent({
  model: new BedrockModel({ modelId: MODEL_ID, region: REGION }),
  systemPrompt: 'あなたは社員情報を扱うアシスタントです。日本語で簡潔に回答してください。',
  tools: [getAllEmployees],
  sessionManager: new SessionManager({ sessionId: 'demo-session-001', storage }),
  plugins: [new ContextOffloader({ storage, maxResultTokens: 2000, previewTokens: 500 })],
})

await agent.invoke('全社員データを取得して、総件数と最初の社員の名前を教えてください。')

ここで ContextOffloader に渡した maxResultTokens: 2000 は、「ツール結果をそのまま会話に残してよい上限のトークン数」です。会話履歴（コンテキスト）は LLM に毎回まるごと送られるため、42KB のツール結果をそのまま残すと、以降のやり取りのたびに巨大なデータを送り続けることになり、トークン消費もコンテキストの上限も圧迫します。

そこで ContextOffloader は、閾値を超えたツール結果を会話から抜き出して DynamoDB へ退避（オフロード）し、会話には短いプレビュー（previewTokens: 500 の分）と「本体は offloader/... にある」という参照情報だけを残します。退避した本体は失われず、必要になれば後から取り出せます（第 7 章）。

実行すると、今回のツール結果（約 42KB）は閾値の 2000 トークンを大きく超えるため自動的にオフロードされ、会話コンテキストにはプレビューだけが残ります。

(2) テーブルの中身を確認する

Github list-keys.ts

確認スクリプトも Storage インターフェース（list と read）だけで書けます。session/ と offloader/ の 2 つのプレフィックスで list し、見つかったキーごとに read してバイト数を表示するだけです。

const storage = buildStorage(requireTableName())

for (const prefix of ['session/', 'offloader/']) {
  const keys = await storage.list(prefix)                 // プレフィックスでキー一覧
  console.log(`--- ${prefix} (${keys.length} 件) ---`)
  for (const key of keys) {
    const data = await storage.read(key)                  // 本体を読んでサイズ確認
    console.log(`${key}  (${data?.length ?? 0} bytes)`)
  }
}

実行結果は次のとおりです。

SessionManager が書いた会話スナップショット（session/...）と、ContextOffloader が退避した 42KB のツール結果（offloader/...）が、自動ネームスペースによって衝突せずに 1 つのテーブルへ同居していることが確認できます。アプリケーション側でプレフィックスの管理は一切していません。

7 デモ 2: 別プロセスからセッションを再開して退避データを検索する

Github chat2.ts

次に、chat1 とは別のプロセスで、同じ sessionId を指定してエージェントを作ります。ツール定義（get_all_employees）はあえて渡していません。

const agent = new Agent({
  model,
  systemPrompt: 'あなたは社員情報を扱うアシスタントです。日本語で簡潔に回答してください。',
  sessionManager: new SessionManager({ sessionId: 'demo-session-001', storage }),
  plugins: [new ContextOffloader({ storage })],
})

await agent.invoke(
  '先ほど取得した社員データの中から、部署が「経理部」の社員を2名探して、名前と社員IDを教えてください。'
)

実行結果です。

ここでは 2 つのことが同時に起きています。

1. SessionManager が DynamoDB のスナップショットから会話を復元し、「先ほど取得した社員データ」という指示語が通じている
2. ContextOffloader が提供する retrieve_offloaded_content ツールを使って、エージェント自身が DynamoDB 上の退避データをパターン検索し、経理部の社員を特定している

回答の EMP-0004 / EMP-0008 は、ダミーデータの部署割り当て（4 部署を順繰り）と一致する正しい結果でした。会話履歴と退避データの両方が共有バックエンドにあるため、「前のプロセスの続き」が成立しています。

8 最後に

今回は、Strands Agents の unified Storage インターフェースを使って DynamoDB バックエンドを自作し、SessionManager と ContextOffloader で共有してみました。

• カスタムバックエンドの実装は write / read / delete / list の 4 メソッドだけで、90 行ほどの最小実装でも SDK のプラグインからそのまま利用できました
• 自動ネームスペース（session/ と offloader/）のおかげで、1 つのバックエンドを複数プラグインで共有でき、会話履歴と退避データを 1 テーブルで管理できました
• 共有バックエンドに会話と退避データの両方があるため、別プロセスからのセッション再開と、retrieve_offloaded_content による退避データの検索がそのまま成立しました

「バックエンドを 1 回選ぶだけで、SDK の永続化がすべてそこに乗る」という設計は、DynamoDB に限りません。他のデータストア（インメモリキャッシュの Redis、リレーショナルデータベースの PostgreSQL など）を保存先にしたバックエンドも、同じ 4 メソッドを実装すれば自作できます。

本記事は @strands-agents/sdk 1.11.2 で検証しています。unified Storage は追加されて間もない機能のため、最新の仕様は公式ドキュメントもあわせてご確認ください。

サンプルコードは、下記のリポジトリで公開しています。

GitHub: strands-agents-dynamodb-storage

9 参考リンク

• Strands Agents Changelog
• Storage | Strands Agents
• Session Management | Strands Agents
• Context Offloader | Strands Agents
• strands-agents/harness-sdk
• Strands Agents に統一 Storage インターフェースが追加されました