[アップデート] Strands Agents に Storage インターフェースが追加されました

uv init --python 3.13
uv add strands-agents boto3

from strands.vended_plugins.context_offloader import FileStorage

from strands.storage import InMemoryStorage, LocalFileStorage, S3Storage

70 bytes      ./artifacts-legacy/.metadata.json
189270 bytes  ./artifacts-legacy/1785117571419_1_tooluse_uAzZomKfAU1SlfKN6CTgfD_0.txt

189282 bytes  ./artifacts-unified/offloader/tooluse_b209HdLjoMKbt9KAtNZA2J_0

storage = InMemoryStorage()
await storage.write("users/alice.json", b'{"name": "Alice"}')
await storage.read("users/alice.json")   # b'{"name": "Alice"}'
await storage.read("users/carol.json")   # None
await storage.list("users/")             # ['users/alice.json']
await storage.delete("users/alice.json") # 存在しないキーでもエラーにならない

# namespace: キーの先頭にプレフィックスを付けたビューを作る
ns = storage.namespace("tenant-a/")
await ns.write("config.json", b"{}")     # 実体は tenant-a/config.json

"""検証用に、わざと大きな結果を返すツール。"""

import json

from strands import tool

ROLES = ["user", "viewer", "editor", "user", "viewer"]

@tool
def list_users() -> str:
    """社内システムのユーザー一覧を取得する。"""
    users = [
        {
            "id": i,
            "name": f"user{i:03d}",
            "email": f"user{i:03d}@example.com",
            "role": "admin" if i in (742, 987) else ROLES[i % len(ROLES)],
            "department": f"dept-{i % 12}",
            "note": "このユーザーは社内システムの標準的な利用者です。",
        }
        for i in range(1, 1001)
    ]
    return json.dumps({"users": users}, ensure_ascii=False)

"""ContextOffloader + InMemoryStorage で大きなツール結果を退避させる。"""

import asyncio

from strands import Agent
from strands.models import BedrockModel
from strands.storage import InMemoryStorage
from strands.vended_plugins.context_offloader import ContextOffloader

from tools import list_users

async def main() -> None:
    storage = InMemoryStorage()

    agent = Agent(
        model=BedrockModel(model_id="us.anthropic.claude-haiku-4-5-20251001-v1:0"),
        tools=[list_users],
        plugins=[ContextOffloader(storage=storage)],
    )

    result = await agent.invoke_async("ユーザー一覧から、admin ロールのユーザーを全員挙げてください。")

    print("最終回答:", result)
    for key in await storage.list():
        data = await storage.read(key)
        print(f"  {key}: {len(data):,} bytes")

if __name__ == "__main__":
    asyncio.run(main())

uv run 02_offloader_inmemory.py

Tool #1: list_users
Tool #2: retrieve_offloaded_content
Tool #3: retrieve_offloaded_content

最終回答: 全体を見ると、adminロールのユーザーは **2名** です：

| ID | 名前 | メールアドレス | 部門 |
|-----|------|--------------|------|
| 742 | user742 | user742@example.com | dept-10 |
| 987 | user987 | user987@example.com | dept-3 |

  offloader/tooluse_uEtG8LApvokM29jkIojFsa_0: 189,282 bytes

export OFFLOAD_BUCKET=strands-offload-demo-$(aws sts get-caller-identity --query Account --output text)
aws s3api create-bucket --bucket "$OFFLOAD_BUCKET" --region us-east-1
aws s3api put-public-access-block --bucket "$OFFLOAD_BUCKET" \
  --public-access-block-configuration "BlockPublicAcls=true,IgnorePublicAcls=true,BlockPublicPolicy=true,RestrictPublicBuckets=true"

- from strands.storage import InMemoryStorage
+ from strands.storage import S3Storage

- storage = InMemoryStorage()
+ storage = S3Storage(BUCKET, prefix="tool-results", region_name="us-east-1")

  agent = Agent(
      model=BedrockModel(model_id="us.anthropic.claude-haiku-4-5-20251001-v1:0"),
      tools=[list_users],
-     plugins=[ContextOffloader(storage=storage)],
+     plugins=[ContextOffloader(storage=storage, evict_after_cycles=None)],
  )

uv run 03_offloader_s3.py
aws s3 ls "s3://$OFFLOAD_BUCKET/" --recursive --human-readable

2026-07-27 08:27:43  184.8 KiB tool-results/offloader/tooluse_84QjVEktPe9TtPT56xnGWu_0

"""エージェントを動かした別プロセスから、S3 に残った退避データを読み直す。"""

import asyncio
import json
import os

from strands.storage import S3Storage

BUCKET = os.environ["OFFLOAD_BUCKET"]

def unframe(frame: bytes) -> tuple[bytes, str]:
    """先頭 2 バイトの長さ + Content-Type を剥がして本体を取り出す。"""
    ct_len = int.from_bytes(frame[:2], "big")
    content_type = frame[2 : 2 + ct_len].decode("utf-8")
    return frame[2 + ct_len :], content_type

async def main() -> None:
    storage = S3Storage(BUCKET, prefix="tool-results", region_name="us-east-1")

    for key in await storage.list():
        frame = await storage.read(key)
        body, content_type = unframe(frame)
        users = json.loads(body)["users"]
        print(f"{key}")
        print(f"  content-type : {content_type}")
        print(f"  users        : {len(users)}")
        print(f"  admins       : {[u['name'] for u in users if u['role'] == 'admin']}")

if __name__ == "__main__":
    asyncio.run(main())

uv run 04_read_from_s3.py

offloader/tooluse_84QjVEktPe9TtPT56xnGWu_0
  content-type : text/plain
  users        : 1000
  admins       : ['user742', 'user987']

aws s3 rm "s3://$OFFLOAD_BUCKET" --recursive
aws s3api delete-bucket --bucket "$OFFLOAD_BUCKET"