0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@kai_kou(甲斐 甲)

Claude Agent SDKのSessionStore、自作して分かった罠3つ

0
Posted at

はじめに

Claude Agent SDK（Python版）に、セッションの transcript を外部ストレージへミラーリングする SessionStore という仕組みが実装されています。Redis や S3、Postgres など好きなバックエンドにセッションを永続化し、プロセスをまたいで会話を再開できるようにするための拡張ポイントです。

CHANGELOG を読んだだけでは実装イメージが掴みにくかったため、実際に pip install してソースコードを読み、自作の SessionStore アダプタを書いて動かしてみました。その過程で、CHANGELOGの説明と実際のAPIシグネチャが食い違っている ことに気づいたほか、実装時に2つの罠にハマったので、その一部始終をまとめます。

この記事で学べること

  • SessionStore プロトコルの実際のシグネチャ（CHANGELOGの記述との差分込み）
  • 公式の適合性テストハーネスで自作アダプタを検証する方法
  • 素朴な実装でハマりがちな2つの罠（async呼び出し忘れ／prefixマッチの境界バグ）

対象読者

  • Claude Agent SDK（Python）でセッション管理をカスタマイズしたい方
  • LLMエージェントの会話ログを自前のストレージに永続化したい方
  • 公式ドキュメントとソースコードの差分を検証する手法に興味がある方

前提環境

  • Python: 3.11
  • claude-agent-sdk: 0.2.111（2026-07-06リリース、MITライセンス）
  • OS: Linux

TL;DR

  • SessionStore は append/load の2メソッドが必須、list_sessions/list_session_summaries/delete/list_subkeys の4メソッドが任意という設計
  • CHANGELOG（v0.1.64エントリ）には「5メソッド・13契約」としか書かれておらず型シグネチャの記載は無かったが、実際にインストールした0.2.111のソースを読むと6メソッド・14契約に増えており、引数も key: SessionKey / entries: list[SessionStoreEntry]（どちらも TypedDict でbytesではなくJSON dict）という構造だった
  • 公式の適合性テスト run_session_store_conformance は async def なので、await を忘れると RuntimeWarning が出るだけで実質何もテストせずに完了する
  • 自作アダプタをファイル名の prefix マッチで実装すると、session_id="sess" が session_id="sess2" を巻き込んで削除してしまうバグを踏んだ

背景: SessionStoreとは何か

claude-agent-sdk（Python）は Claude Code のサブプロセスを Python から操作するための公式 SDK です。会話セッションは既定でローカルディスクに JSONL 形式で書き出されますが、複数プロセス・複数マシンでセッションを共有したい場合や、Redis/S3/Postgres のような外部ストレージに永続化したい場合、標準のローカルディスク保存だけでは足りません。

SessionStore は、この「ローカルディスクへの書き込みに加えて、外部ストレージにも同時にミラーリングする」ためのアダプタ・プロトコルです。公式リポジトリには examples/session_stores/ 配下に S3・Redis・Postgres 向けの参照実装が置かれています（wheel には含まれず、必要な分だけ自分のプロジェクトにコピーして使う設計）1。

環境構築

まず venv を切って SDK をインストールします。

python3 -m venv venv
source venv/bin/activate
pip install claude-agent-sdk
pip show claude-agent-sdk

筆者の環境では以下がインストールされました。

Name: claude-agent-sdk
Version: 0.2.111
License: MIT
Requires: anyio, mcp, sniffio

claude_agent_sdk パッケージが何をエクスポートしているか確認します。

python3 -c "
import claude_agent_sdk
print([x for x in dir(claude_agent_sdk) if 'ession' in x.lower() or 'Store' in x])
"

['ForkSessionResult', 'InMemorySessionStore', 'SDKSessionInfo', 'SessionKey',
 'SessionListSubkeysKey', 'SessionMessage', 'SessionStore', 'SessionStoreEntry',
 'SessionStoreFlushMode', 'SessionStoreListEntry', 'SessionSummaryEntry',
 'delete_session', 'delete_session_via_store', 'fold_session_summary',
 'fork_session', 'fork_session_via_store', 'get_session_info',
 'get_session_info_from_store', 'get_session_messages',
 'get_session_messages_from_store', 'import_session_to_store', 'list_sessions',
 'list_sessions_from_store', 'rename_session', 'rename_session_via_store',
 'tag_session', 'tag_session_via_store']

SessionStore 本体に加えて、SessionKey / SessionStoreEntry / SessionStoreListEntry という型、InMemorySessionStore という参照実装、そして *_via_store というヘルパー関数群が確認できます。

1. ドキュメントとソースコードの差分を確認する

CHANGELOG.md の 0.1.64 エントリを確認すると、SessionStore機能について次のように書かれています（原文）。

"SessionStore adapter: Full SessionStore support at parity with the TypeScript SDK. Includes a SessionStore protocol with 5 methods (append, load, list_sessions, delete, list_subkeys)"
同エントリには続けて "Also adds a 13-contract conformance test harness at claude_agent_sdk.testing.run_session_store_conformance" ともあります。

ここにはメソッド名の列挙と契約数（13）はありますが、各メソッドの型シグネチャは一切書かれていません。「引数は文字列なのか」「transcriptの中身は生バイト列なのかJSON なのか」は、CHANGELOGの文面だけでは分かりません。

そこで、実際に0.2.111をインストールして inspect.getsource(claude_agent_sdk.SessionStore) でソースを直接読んでみました。

import inspect
import claude_agent_sdk
print(inspect.getsource(claude_agent_sdk.SessionStore))

すると、CHANGELOGの記述時点（0.1.64）から2点進化していることが分かりました。

① メソッドが5個ではなく6個になっていた: list_session_summaries という任意メソッドが増えており、必須2（append/load）・任意4（list_sessions/list_session_summaries/delete/list_subkeys）の計6メソッド構成でした。

class SessionStore(Protocol):
    async def append(self, key: SessionKey, entries: list[SessionStoreEntry]) -> None: ...
    async def load(self, key: SessionKey) -> list[SessionStoreEntry] | None: ...
    async def list_sessions(self, project_key: str) -> list[SessionStoreListEntry]:
        raise NotImplementedError
    async def list_session_summaries(self, project_key: str) -> list[SessionSummaryEntry]:
        raise NotImplementedError
    async def delete(self, key: SessionKey) -> None:
        raise NotImplementedError
    async def list_subkeys(self, key: SessionListSubkeysKey) -> list[str]:
        raise NotImplementedError

② 引数がプレーンな型ではなく TypedDict ベースだった: CHANGELOGの文面からは読み取れなかった部分ですが、session_id は単純な文字列引数ではなく key: SessionKey（TypedDict）としてまとめて渡され、transcriptの中身も entries: list[SessionStoreEntry]（JSON化可能な辞書のリスト）という構造でした。中身は次のような定義です。

class SessionKey(TypedDict):
    project_key: str          # 呼び出し側が定義するスコープ（既定はcwdをサニタイズした値）
    session_id: str
    subpath: NotRequired[str]  # サブエージェントのtranscriptには "subagents/agent-1" 等が入る

SessionStoreEntry は生バイト列ではなく JSON化可能な辞書（TypedDict(total=False)、type フィールドのみ必須）です。

class SessionStoreEntry(TypedDict, total=False):
    type: Required[str]
    uuid: str
    timestamp: str
    # それ以外のフィールドは不透明なJSONとしてそのまま通す

さらに、適合性テスト関数のdocstringも実際に読んでみました。

from claude_agent_sdk.testing import run_session_store_conformance
import inspect
print(inspect.getdoc(run_session_store_conformance))
# => Assert the 14 :class:`SessionStore` behavioral contracts.

CHANGELOGの 0.1.64 エントリでは「13-contract」でしたが、手元の0.2.111では 14契約 に増えていました。CHANGELOGは機能追加時点のスナップショットであり、その後追加されたメソッドや契約数の変化までは追随して書き直されないため、こうした差分が生まれます。「CHANGELOGを読んで概要を掴む」だけでなく、「pip install して inspect で実物を読む」まで行って初めて、この2点の差分（メソッド数・型シグネチャ）に気づけました。

2. InMemorySessionStoreで適合性テストを動かす

SDKには claude_agent_sdk.testing.run_session_store_conformance という適合性テストハーネスが同梱されており、自作アダプタが仕様どおりに動くか14個の契約（append/loadの往復・順序保証・subpathの独立性・delete時のカスケード削除など）を検証してくれます。

まずは参照実装 InMemorySessionStore に対して実行してみます。

import asyncio
from claude_agent_sdk import InMemorySessionStore
from claude_agent_sdk.testing import run_session_store_conformance

async def main():
    run_session_store_conformance(lambda: InMemorySessionStore())  # ← 実はこれ動いていない
    print("done")

asyncio.run(main())

これを実行すると、done は表示されるものの、以下の警告が出ます。

RuntimeWarning: coroutine 'run_session_store_conformance' was never awaited

inspect.signature() だけを見ると戻り値の型注釈が None なので同期関数に見えますが、inspect.iscoroutinefunction() で確認すると True です。await を付け忘れると 例外もエラーメッセージも出さずに「テストした気になれてしまう」 のが厄介なところです。正しくは次のように書きます。

async def main():
    await run_session_store_conformance(lambda: InMemorySessionStore())
    print("InMemorySessionStore: 14 contracts PASSED")

asyncio.run(main())
# => InMemorySessionStore: 14 contracts PASSED

CIに組み込む場合は pytest-asyncio 等でこの関数を確実に await する形にしておかないと、テストが常に green のまま実質何も検証していない、という事故につながります。

3. 自作SessionStoreアダプタを実装する

InMemory実装だけでは芸がないので、JSONL形式でディスクに永続化する FileSessionStore を自作してみます。最終的に動いた実装は以下のとおりです。

import json
from pathlib import Path


class FileSessionStore:
    """SessionStore を「本編1ファイル + サブキー用ディレクトリ」で永続化する自作アダプタ。"""

    def __init__(self, root: str):
        self.root = Path(root)
        self.root.mkdir(parents=True, exist_ok=True)

    def _base(self, key) -> str:
        return f"{key['project_key']}__{key['session_id']}"

    def _path(self, key) -> Path:
        base = self._base(key)
        subpath = key.get("subpath")
        if subpath:
            p = self.root / f"{base}.subkeys" / f"{subpath}.jsonl"
            p.parent.mkdir(parents=True, exist_ok=True)
            return p
        return self.root / f"{base}.jsonl"

    async def append(self, key, entries):
        with self._path(key).open("a") as f:
            for entry in entries:
                f.write(json.dumps(entry) + "\n")

    async def load(self, key):
        p = self._path(key)
        if not p.exists():
            return None
        return [json.loads(line) for line in p.read_text().splitlines() if line.strip()]

    async def list_sessions(self, project_key):
        results = []
        for p in self.root.glob(f"{project_key}__*.jsonl"):
            results.append({
                "session_id": p.stem[len(project_key) + 2:],
                "mtime": int(p.stat().st_mtime * 1000),
            })
        return results

    async def delete(self, key):
        if key.get("subpath"):
            self._path(key).unlink(missing_ok=True)
            return
        base = self._base(key)
        (self.root / f"{base}.jsonl").unlink(missing_ok=True)
        subdir = self.root / f"{base}.subkeys"
        if subdir.exists():
            for p in subdir.rglob("*.jsonl"):
                p.unlink()

    async def list_subkeys(self, key):
        subdir = self.root / f"{self._base(key)}.subkeys"
        if not subdir.exists():
            return []
        return [str(p.relative_to(subdir))[: -len(".jsonl")] for p in subdir.rglob("*.jsonl")]

これを適合性テストにかけると、全契約をパスします。

import asyncio
import tempfile
from claude_agent_sdk.testing import run_session_store_conformance
from file_store import FileSessionStore

async def main():
    await run_session_store_conformance(lambda: FileSessionStore(tempfile.mkdtemp()))
    print("FileSessionStore(自作JSONLアダプタ): 14 contracts PASSED")

asyncio.run(main())
# => FileSessionStore(自作JSONLアダプタ): 14 contracts PASSED

ここに至るまでに、2つの罠を踏んでいます。

ハマりポイント1: prefixマッチが別セッションを巻き込む

最初の実装では、ファイル名を f"{project_key}__{session_id}" という素朴な文字列にし、delete/list_sessions を glob(f"{prefix}*.jsonl") のような prefix マッチで実装していました。

これで適合性テストを流すと、以下のアサーションで失敗しました。

await store.delete(_KEY)  # _KEY = {"project_key": "proj", "session_id": "sess"}
...
loaded_other = await store.load(other)  # other = {"project_key": "proj", "session_id": "sess2"}
assert loaded_other is not None and len(loaded_other) == 1

原因は単純で、"proj__sess" という文字列は "proj__sess2" の prefixとして一致してしまう ため、session_id="sess" を消したつもりが session_id="sess2" まで巻き込んで消していました。ファイル名やキーを prefix マッチで扱う実装は、区切り文字の有無まで含めて境界を明示しないと、意図しない他セッションへの副作用を生みます。今回は「本編は {base}.jsonl というちょうど1つのファイル名」に固定し、prefix マッチが必要なのはサブキー探索だけに絞ることで解消しました。

ハマりポイント2: subpathに含まれる"/"の扱い

SessionKey の subpath フィールドには、サブエージェントの transcript を表す "subagents/agent-1" のような スラッシュ入りの文字列 が入ります。最初は subpath.replace("/", "_") でファイル名に押し込めていましたが、list_subkeys で元の文字列に復元できず、次のアサーションで失敗しました。

# run_session_store_conformance 内部からの抜粋（async関数内で実行される）
async def _check(store, _KEY):
    assert await store.list_subkeys(_KEY) == ["subagents/agent-2"]

"subagents_agent-2" のように潰した文字列からは、区切りが - なのか / を置換したものなのか復元できません。最終的には、サブキー用に {base}.subkeys/ というディレクトリを切り、その配下に subpath をそのままディレクトリ構造として保存する設計に変更して解決しました（subagents/agent-1.jsonl のようなパスがそのままディレクトリ階層になる）。

実運用で使うときの注意点

公式ドキュメントによると、SessionStore を使う際はいくつか運用上の前提があります。

  • ローカルディスクへの書き込みは SessionStore の有無に関わらず継続される。SessionStore はあくまで「セカンダリコピー」を受け取る位置づけで、ゼロデータ保持（ZDR）等の要件でローカルコピーが不要な場合は CLAUDE_CONFIG_DIR=/tmp を指定する
  • append は 1ターンあたり約100msの間隔でバッチ到着する。失敗時は3回までリトライされ、それでも失敗すると MirrorErrorMessage として表面化する（タイムアウトはリトライされない）
  • delete はSDK側から自動では呼ばれず、delete_session_via_store() を明示的に呼んだ場合のみ実行される。保持期間（TTL）の管理はアダプタ側の責務

CLIからは --session-mirror オプションで、この SessionStore アダプタへのリアルタイム転送（Transcript Mirroring）を有効化できます。

著者視点の発見ポイント

CHANGELOGはメソッド名や契約数の列挙はしていても、型シグネチャまでは書いていませんでした。この「書かれていない部分」を実装イメージだけで補って進めていたら、後から TypedDict ベースの構造に書き直す手戻りが発生していたはずです。pip install して inspect.getsource()／inspect.getdoc() で実物を確認する、公式の適合性テストハーネスがあればまずそれを走らせる、という2ステップを踏むだけで、こうしたドキュメントの記載漏れ・古い情報とのズレは高い確率で検出できます。特にLLM関連のSDKは開発速度が速く、CHANGELOGの記述時点からマイナーバージョンが何度も上がっているため、「動くコードで検証してから書く」習慣が効いてくる領域だと感じました。

まとめ

  • Claude Agent SDK（Python）の SessionStore は append/load が必須、他4メソッドが任意という設計
  • CHANGELOG（v0.1.64時点）は「5メソッド・13契約」だったが、実際にインストールした0.2.111では6メソッド・14契約に増えており、引数も SessionKey/SessionStoreEntry（ともにTypedDict）ベースだった（CHANGELOGには型の記載自体が無かった）
  • 公式の適合性テスト run_session_store_conformance は async def なので await を忘れると無言でスキップされる
  • 自作アダプタを書く際は、prefixマッチの境界と subpath 内の / の扱いに注意が必要

参考リンク

  1. claude-agent-sdk-python CHANGELOG ↩

0
0
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?