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?

@K5K(松田 啓佑)

ClaudeCodeの会話履歴をObsidianに取り込む (Hooksだけで実現)

Posted at

ClaudeCodeで日報を書きたい意欲が高まっている今日この頃、「ClaudeCodeとの会話履歴をObsidianに保存しておきたい」と思うようになりました。

なぜかというと、ClaudeCodeとの会話内容は自分の業務実績のソースになからです。どんなコードを書いたか、どんな問題を解決したか、どんなことを考えたか。これらはすべて日報を書くときのネタになります。それに、貯めておけば後々何かに使えるかもしれないという期待もありました。(まだ明確に思いついてはいませんが)

「LaunchAgentベースの実装」から「Hooksのみでの実装」へ

ということでZennの記事（LaunchAgentを使ってClaude Codeの会話履歴をObsidianに取り込む）を参考にして実現していました。こちらの方法はLaunchAgentを使ってClaudeCodeのセッションファイルを監視し、定期的にObsidianメモに同期するというものでした。

これはこれでうまく動いていおり、特に不満は感じておりませんでした。ただ、ClaudeCode外の機能（LaunchAgent）に依存するのが少しネックだなとは感じていました。

ClaudeCodeの機能だけで、もっとシンプルにできたらいいなーと思っていたところ、「あれ？これってHooksで実現できるんじゃね？」と閃き、やってみたらできました。

というわけで、この記事ではHooksのみでClaudeCodeの会話履歴をObsidianメモとして取り込む方法を紹介します。

Hooksとは？

Hooksとは、ClaudeCodeのライフサイクルの特定のポイントでシェルコマンドやLLMプロンプトを実行させる機能です。

詳しくは公式ドキュメントを参照してください。

Hooksを使うと、以下のようなイベントでスクリプトを実行できます。

  • SessionStart: セッション開始時
  • Stop: ユーザーがメッセージを送信するたびに
  • SessionEnd: セッション終了時
  • PreToolUse: ツール実行前
  • Notification: 通知イベント

では、このHooksを使って具体的にどのように実装したかを説明します。

実装の詳細

全体の流れ

以下のタイミングでHooksを発火します。

  1. SessionStart → セッション開始時にObsidianメモ(実態はMarkdownファイル)を作成
  2. Stop（繰り返し） → ユーザーがメッセージを送信するたびに新規メッセージをメモに追記
  3. SessionEnd → セッション終了時に残りのメッセージを追記し、frontmatterを更新

この流れにより、セッション中の会話がリアルタイムにObsidianメモに保存されます。

実際に利用されたい方へ

以下を行ってください。

  • ~/.claude/settings.jsonへhooksの追記
  • ~/.claude/scripts/obsidian-save.pyの配置(スクリプト本文は下にあります)

settings.json の設定

Hooksの設定は ~/.claude/settings.json に記述します。以下が実際の設定内容です。

{
  "hooks": {
    "SessionStart": [
      {
        "matcher": "",
        "hooks": [
          {
            "type": "command",
            "command": "python3 ~/.claude/scripts/obsidian-save.py"
          }
        ]
      }
    ],
    "Stop": [
      {
        "matcher": "",
        "hooks": [
          {
            "type": "command",
            "command": "python3 ~/.claude/scripts/obsidian-save.py"
          }
        ]
      }
    ],
    "SessionEnd": [
      {
        "matcher": "",
        "hooks": [
          {
            "type": "command",
            "command": "python3 ~/.claude/scripts/obsidian-save.py"
          }
        ]
      }
    ]
  }
}

obsidian-save.py

Pythonスクリプト（~/.claude/scripts/obsidian-save.py）は、ClaudeCodeのセッションファイル（session.jsonl）を読み込み、Markdown形式に変換してObsidianに保存します。

主要な機能を説明します。

1. JSONLファイルの読み込み

ClaudeCodeのセッションファイルは、JSONL形式（1行1JSONオブジェクト）で保存されています。スクリプトはこのファイルを読み込み、各メッセージをパースします。

2. Markdownフォーマット

ユーザーとアシスタントのメッセージをMarkdown形式に変換します。具体的には、以下のようなフォーマットになります。

### 17:31:23 User

ここにユーザーのメッセージ

---

### 17:31:30 Claude

ここにClaudeの応答

3. ノイズ除去

ClaudeCodeのセッションファイルには、<system-reminder> などのシステムタグが含まれることがあります。これらは会話履歴としては不要なので、正規表現で除去します。

また、サブエージェントのメッセージ（isSidechain: true）や、メタメッセージ（isMeta: true）もスキップします。これにより、実際のユーザーとClaudeの会話のみが保存されます。

4. ファイル管理

出力されるMarkdownファイルは、以下のような命名規則で保存されます。

claudecode/YYYYMMDD/HHMMSS_セッションID8桁.md

例：claudecode/20260206/162928_336e0382.md

日付ごとにディレクトリが分かれるため、後から利用しやすいです。

5. Frontmatter生成

Markdownファイルの先頭には、以下のようなfrontmatterが生成されます。

---
session_id: 336e0382-xxxx-xxxx-xxxx-xxxxxxxxxxxx
project: /Users/k-matsuda/Documents/Obsidian
time_start: 2026-02-06T16:29:28+09:00
time_end: "2026-02-06T17:45:12+09:00"
changed_files:
  - "/Users/k-matsuda/Documents/Obsidian/Notes/2026-02-06-1645 ブログ原稿.md"
  - "/Users/k-matsuda/.claude/settings.json"
tags:
  - claude-code
---

session_id、project、time_start、time_end、changed_files などのメタデータが含まれます。changed_files には、セッション中に編集されたファイルのパスが記録されます。

6. 差分追記

スクリプトは、前回処理した行番号を状態ファイル（~/.claude/obsidian-hook-state.json）に記録します。次回実行時には、この行番号から新規行のみを読み込みます。

これにより、ファイル全体を毎回再処理する必要がなくなり、効率的に動作します。

スクリプトの内容

obsidian-save.py の全コード 
#!/usr/bin/env python3
"""Claude Code会話履歴をObsidianノートとして自動保存するフックスクリプト。

対応イベント: SessionStart, Stop, SessionEnd
環境変数からフック情報を取得し、セッションごとに1つのMarkdownファイルを管理する。
"""

import json
import os
import re
import sys
import traceback
from datetime import datetime, timezone, timedelta
from pathlib import Path

# ============================================
# 定数
# ============================================
JST = timezone(timedelta(hours=9))
OBSIDIAN_VAULT = Path.home() / "Documents" / "Obsidian"
OUTPUT_BASE = OBSIDIAN_VAULT / "claudecode"
STATE_FILE = Path.home() / ".claude" / "obsidian-hook-state.json"
LOG_FILE = Path.home() / ".claude" / "logs" / "obsidian-save.log"

# system-reminderなどのタグを除去する正規表現
SYSTEM_TAG_PATTERN = re.compile(
    r"<(?:system-reminder|command-name|local-command|task-notification)>.*?"
    r"</(?:system-reminder|command-name|local-command|task-notification)>",
    re.DOTALL,
)

# ============================================
# ユーティリティ
# ============================================

def log_error(msg: str) -> None:
    """エラーをログファイルに記録する。"""
    try:
        LOG_FILE.parent.mkdir(parents=True, exist_ok=True)
        with open(LOG_FILE, "a", encoding="utf-8") as f:
            ts = datetime.now(JST).strftime("%Y-%m-%d %H:%M:%S")
            f.write(f"[{ts}] {msg}\n")
    except Exception:
        pass


def load_state() -> dict:
    """状態ファイルを読み込む。"""
    if STATE_FILE.exists():
        try:
            return json.loads(STATE_FILE.read_text(encoding="utf-8"))
        except (json.JSONDecodeError, OSError):
            return {}
    return {}


def save_state(state: dict) -> None:
    """状態ファイルを保存する。"""
    STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
    STATE_FILE.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding="utf-8")


def read_hook_input() -> dict:
    """stdinからフック入力JSONを読み取る。"""
    try:
        data = sys.stdin.read()
        if data.strip():
            return json.loads(data)
    except (json.JSONDecodeError, OSError):
        pass
    return {}


def parse_timestamp(ts_str: str) -> datetime | None:
    """ISO形式のタイムスタンプをパースする。"""
    if not ts_str:
        return None
    try:
        # Python 3.11+のfromisoformatはタイムゾーン付きに対応
        return datetime.fromisoformat(ts_str)
    except (ValueError, TypeError):
        try:
            # フォールバック: Zサフィックス対応
            return datetime.fromisoformat(ts_str.replace("Z", "+00:00"))
        except (ValueError, TypeError):
            return None


def format_time(ts_str: str) -> str:
    """タイムスタンプからHH:MM:SS形式のJST時刻を返す。"""
    dt = parse_timestamp(ts_str)
    if dt:
        return dt.astimezone(JST).strftime("%H:%M:%S")
    return ""


# ============================================
# Markdownフォーマット
# ============================================

def clean_text(text: str) -> str:
    """system-reminderなどのタグを除去する。"""
    return SYSTEM_TAG_PATTERN.sub("", text).strip()




def should_skip_message(entry: dict, session_id: str) -> bool:
    """メッセージをスキップすべきか判定する。"""
    msg_type = entry.get("type", "")

    # user/assistant以外はスキップ
    if msg_type not in ("user", "assistant"):
        return True

    # メタメッセージをスキップ
    if entry.get("isMeta", False):
        return True

    # サブエージェントのメッセージをスキップ
    if entry.get("isSidechain", False):
        return True

    # セッションIDの不一致をスキップ
    entry_session = entry.get("sessionId", "")
    if entry_session and session_id and entry_session != session_id:
        return True

    return False


def format_entry(entry: dict) -> str | None:
    """JSONLの1エントリをMarkdownにフォーマットする。"""
    msg_type = entry.get("type", "")
    message = entry.get("message", {})
    content = message.get("content", "")
    timestamp = entry.get("timestamp", "")
    time_str = format_time(timestamp)

    if msg_type == "user":
        # contentが文字列の場合
        if isinstance(content, str):
            text = clean_text(content)
            if not text:
                return None
            # ノイズタグを含むメッセージをスキップ
            if re.search(r"<(?:local-command|command-name|system-reminder|task-notification)>", text):
                return None
            header = f"### {time_str} 🙋 User\n\n" if time_str else "### 🙋 User\n\n"
            return header + text + "\n"

        # contentがリストの場合、テキストブロックのみ抽出
        if isinstance(content, list):
            text_parts = []
            for block in content:
                if isinstance(block, dict) and block.get("type") == "text":
                    cleaned = clean_text(block.get("text", ""))
                    if cleaned and not re.search(
                        r"<(?:local-command|command-name|system-reminder|task-notification)>", cleaned
                    ):
                        text_parts.append(cleaned)

            if text_parts:
                header = f"### {time_str} 🙋 User\n\n" if time_str else "### 🙋 User\n\n"
                return header + "\n".join(text_parts) + "\n"
            return None

        return None

    elif msg_type == "assistant":
        if not isinstance(content, list):
            return None

        parts = []
        has_text = False

        for block in content:
            if not isinstance(block, dict):
                continue
            block_type = block.get("type", "")

            if block_type == "text":
                text = clean_text(block.get("text", ""))
                if text and not text.startswith("No response requested"):
                    if not has_text:
                        header = f"### {time_str} 🤖 Claude\n\n" if time_str else "### 🤖 Claude\n\n"
                        parts.append(header)
                        has_text = True
                    parts.append(text + "\n")

            # tool_use, thinking, server_tool_use等はスキップ

        return "\n".join(parts) if parts else None

    return None


# ============================================
# ファイル生成
# ============================================

def generate_frontmatter(session_id: str, cwd: str, start_time: datetime) -> str:
    """Markdownファイルのfrontmatterを生成する。"""
    return (
        "---\n"
        f"session_id: {session_id}\n"
        f"project: {cwd}\n"
        f"time_start: {start_time.astimezone(JST).isoformat()}\n"
        "time_end: \"\"\n"
        "changed_files: []\n"
        "tags:\n"
        "  - claude-code\n"
        "---\n\n"
    )


def generate_metadata_table(session_id: str, cwd: str, start_time: datetime) -> str:
    """メタデータテーブルを生成する。"""
    date_str = start_time.astimezone(JST).strftime("%Y-%m-%d %H:%M:%S")
    return (
        "| Key | Value |\n"
        "|---|---|\n"
        f"| Session | `{session_id[:8]}` |\n"
        f"| Project | `{cwd}` |\n"
        f"| Started | {date_str} |\n\n"
        "---\n\n"
        "## Conversation Log\n\n"
    )


def create_initial_file(session_id: str, cwd: str, start_time: datetime) -> Path:
    """初期Markdownファイルを作成する。"""
    date_dir = start_time.astimezone(JST).strftime("%Y%m%d")
    time_prefix = start_time.astimezone(JST).strftime("%H%M%S")
    short_id = session_id[:8]

    output_dir = OUTPUT_BASE / date_dir
    output_dir.mkdir(parents=True, exist_ok=True)

    filename = f"{time_prefix}_{short_id}.md"
    output_path = output_dir / filename

    frontmatter = generate_frontmatter(session_id, cwd, start_time)
    metadata = generate_metadata_table(session_id, cwd, start_time)

    output_path.write_text(frontmatter + metadata, encoding="utf-8")
    return output_path


# ============================================
# JSONL処理
# ============================================

def read_jsonl_lines(path: str, start_line: int = 0) -> list[tuple[int, dict]]:
    """JSONLファイルからstart_line以降の行を読み込む。"""
    results = []
    try:
        with open(path, "r", encoding="utf-8") as f:
            for i, line in enumerate(f):
                if i < start_line:
                    continue
                line = line.strip()
                if not line:
                    continue
                try:
                    results.append((i, json.loads(line)))
                except json.JSONDecodeError:
                    continue
    except (OSError, IOError):
        pass
    return results


def append_messages(output_path: Path, entries: list[tuple[int, dict]], session_id: str) -> None:
    """フォーマットしたメッセージをMarkdownファイルに追記する。"""
    parts = []
    for _line_num, entry in entries:
        if should_skip_message(entry, session_id):
            continue
        formatted = format_entry(entry)
        if formatted:
            parts.append(formatted)

    if parts:
        with open(output_path, "a", encoding="utf-8") as f:
            f.write("\n---\n\n".join(parts) + "\n\n---\n\n")


def extract_changed_files(entries: list[tuple[int, dict]]) -> list[str]:
    """tool_useからファイルパスを抽出する。"""
    files = set()
    for _line_num, entry in entries:
        if entry.get("type") != "assistant":
            continue
        message = entry.get("message", {})
        content = message.get("content", [])
        if not isinstance(content, list):
            continue
        for block in content:
            if not isinstance(block, dict) or block.get("type") != "tool_use":
                continue
            inp = block.get("input", {})
            if not isinstance(inp, dict):
                continue
            # 各ツールのファイルパスパラメータを抽出
            for key in ("file_path", "path", "filePath", "notebook_path"):
                val = inp.get(key, "")
                if val and isinstance(val, str) and "/" in val:
                    files.add(val)
    return sorted(files)


def update_frontmatter_end(output_path: Path, end_time: datetime, changed_files: list[str]) -> None:
    """frontmatterのtime_endとchanged_filesを更新する。"""
    try:
        text = output_path.read_text(encoding="utf-8")
    except OSError:
        return

    # time_endを更新
    end_str = end_time.astimezone(JST).isoformat()
    text = re.sub(r'time_end: ".*?"', f'time_end: "{end_str}"', text, count=1)

    # changed_filesを更新
    if changed_files:
        files_yaml = "\n".join(f"  - \"{f}\"" for f in changed_files[:50])  # 最大50ファイル
        text = re.sub(r"changed_files: \[.*?\]", f"changed_files:\n{files_yaml}", text, count=1)

    output_path.write_text(text, encoding="utf-8")


# ============================================
# イベントハンドラ
# ============================================

def ensure_initialized(session_id: str, transcript_path: str, cwd: str, state: dict) -> tuple[dict, Path]:
    """セッションが初期化されていなければ初期化する（resume対応）。"""
    if session_id in state:
        return state, Path(state[session_id]["output_path"])

    now = datetime.now(JST)

    # transcript_pathの最初の行からtimestampを取得して開始時刻を推定
    try:
        with open(transcript_path, "r", encoding="utf-8") as f:
            first_line = f.readline().strip()
            if first_line:
                first_entry = json.loads(first_line)
                ts = parse_timestamp(first_entry.get("timestamp", ""))
                if ts:
                    now = ts
    except (OSError, json.JSONDecodeError):
        pass

    output_path = create_initial_file(session_id, cwd, now)
    state[session_id] = {
        "output_path": str(output_path),
        "last_line": 0,
        "start_time": now.isoformat(),
    }
    save_state(state)
    return state, output_path


def handle_session_start(hook_input: dict) -> None:
    """SessionStartイベントの処理。"""
    session_id = hook_input.get("session_id", "")
    cwd = hook_input.get("cwd", "")
    if not session_id:
        return

    state = load_state()
    now = datetime.now(JST)

    output_path = create_initial_file(session_id, cwd, now)
    state[session_id] = {
        "output_path": str(output_path),
        "last_line": 0,
        "start_time": now.isoformat(),
    }
    save_state(state)
    log_error(f"SessionStart: {session_id[:8]} -> {output_path}")


def handle_stop(hook_input: dict) -> None:
    """Stopイベントの処理。新規行を追記する。"""
    session_id = hook_input.get("session_id", "")
    transcript_path = hook_input.get("transcript_path", "")
    cwd = hook_input.get("cwd", "")
    if not session_id or not transcript_path:
        return

    state = load_state()

    # SessionStartなしでStopが来た場合（resume等）は初期化
    state, output_path = ensure_initialized(session_id, transcript_path, cwd, state)

    last_line = state[session_id].get("last_line", 0)
    entries = read_jsonl_lines(transcript_path, last_line)

    if entries:
        append_messages(output_path, entries, session_id)
        new_last = entries[-1][0] + 1  # 次の開始行
        state[session_id]["last_line"] = new_last
        save_state(state)


def handle_session_end(hook_input: dict) -> None:
    """SessionEndイベントの処理。残りの行を追記し、frontmatterを更新する。"""
    session_id = hook_input.get("session_id", "")
    transcript_path = hook_input.get("transcript_path", "")
    cwd = hook_input.get("cwd", "")
    if not session_id or not transcript_path:
        return

    state = load_state()

    # SessionStartなしの場合に対応
    state, output_path = ensure_initialized(session_id, transcript_path, cwd, state)

    # 残りの行を追記
    last_line = state[session_id].get("last_line", 0)
    entries = read_jsonl_lines(transcript_path, last_line)

    if entries:
        append_messages(output_path, entries, session_id)
        new_last = entries[-1][0] + 1
        state[session_id]["last_line"] = new_last

    # 全行を読み込んでchanged_filesを抽出
    all_entries = read_jsonl_lines(transcript_path, 0)
    changed_files = extract_changed_files(all_entries)

    # frontmatter更新
    end_time = datetime.now(JST)
    update_frontmatter_end(output_path, end_time, changed_files)

    save_state(state)
    log_error(f"SessionEnd: {session_id[:8]} -> {output_path}")


# ============================================
# メイン
# ============================================

def main() -> None:
    hook_input = read_hook_input()
    event_name = hook_input.get("hook_event_name", "")

    if event_name == "SessionStart":
        handle_session_start(hook_input)
    elif event_name == "Stop":
        handle_stop(hook_input)
    elif event_name == "SessionEnd":
        handle_session_end(hook_input)
    else:
        log_error(f"Unknown event: {event_name}")


if __name__ == "__main__":
    try:
        main()
    except Exception:
        log_error(traceback.format_exc())
    # フック失敗がClaude Code本体に影響しないよう常にexit 0
    sys.exit(0)

100%claudecode産でございます。

動作確認

実際にClaudeCodeで会話を始めると、自動的にObsidianに以下のような形式でMarkdownファイルが生成されます。

出力されるMarkdownファイルの例

---
session_id: 336e0382-xxxx-xxxx-xxxx-xxxxxxxxxxxx
project: /Users/k-matsuda/Documents/Obsidian
time_start: 2026-02-06T16:29:28+09:00
time_end: "2026-02-06T17:45:12+09:00"
changed_files:
  - "/Users/k-matsuda/Documents/Obsidian/Notes/2026-02-06-1645 ブログ原稿.md"
  - "/Users/k-matsuda/.claude/settings.json"
tags:
  - claude-code
---

| Key | Value |
|---|---|
| Session | `336e0382` |
| Project | `/Users/k-matsuda/Documents/Obsidian` |
| Started | 2026-02-06 16:29:28 |

---

## Conversation Log

### 16:29:30 User

Pythonスクリプトを確認してほしい

---

### 16:29:35 Claude

スクリプトを確認します。

---

### 16:30:12 User

ありがとう。次にsettings.jsonを見てほしい

---

frontmatterにはセッション情報が記録され、メタデータテーブルにはプロジェクトと開始時刻が表示されます。会話ログは時系列順に並び、ユーザーとClaudeのメッセージが区別されています。

まとめ

Hooksを使うことで、ClaudeCodeの会話履歴をObsidianに取り込む仕組みをシンプルに実現できました。日報作成時に会話履歴を参照したり、後々の振り返りに活用したりできるので、ぜひ参考にしていただけると幸いです。

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?