📔
Claude Code Hooksで作るSlack→Notion自動サマライズ

ほんまでっか？
2025/08/29に公開
こんにちは！

株式会社カンリーでSREをやっている本間です。ほんまですよ！
今回は、Slack→Notionデータベースへの自動サマライズを、Claude Code Hooksを使って作ってみたので紹介します。
これを作ろうと思ったモチベーションは、インシデントが発生した際のインシデント台帳やポストモーテムなどのドキュメント作成の手間を減らすためです。弊社ではこれらのドキュメントをNotionデータベースに蓄積しており、原因・対応履歴の情報をフォーマットに合わせて埋めていく必要があります。
しかし、システムのアラートはSlackに通知され、調査の状況報告や恒久対応の共有もSlackで行われるため、Slackに情報が溜まってしまいがちです。これらフロー情報をストックしていくためにAIを活用してみました。今回の例はインシデントですが、それ以外の情報ストックにも活かせそうです…！
（Claude Codeを使わずに、AIエージェントやBotの形態の方が便利そうですが、まずは手頃なところから始めてみました）

 実行結果イメージしやすいよう先に実行結果からです。
まずはSlackで、インシデント対応を模した投稿をしてみました



上記をサマライズし、Notionデータベースに登録された内容が以下になります

本文にも、Slackで会話した内容がまとまっています。

今回スレッドには余計な情報がなかったので精度が良かった可能性はありますが、テキストからNotionデータベースの列にうまく当てはめられています！
Claude Code実行の様子

途中バリデーションにひっかかっていますが、自動で修正してくれています。

 構成構成は以下のようになっています。

ポイント
hooksはUserPromptSubmitを使い、プロンプト実施タイミングでpython実行
関係ないプロンプトの時まで実行しないよう条件分岐（後ほど解説）
Slackはスレッドのテキスト取得するだけなのでSlack Apiを使用

 Claude Code Hooksの設定まずはClaude Code Hooksの設定です。
~/.claude/settings.json
  (省略)
  "hooks": {
    "UserPromptSubmit": [
      {
        "matcher": "",
        "hooks": [
          {
            "type": "command",
            "command": "python3 /<your_path>/user_prompt_router.py"
          }
        ]
      }
    ]
  }
UserPromptSubmitは、プロンプト実行前に実行されるhooksです
matcherは空です。

これは/hooksでUserPromptSubmitを選択したところこのように生成されたので、そのままにしています。
以下公式ドキュメントを見る限り、記述しても効かなそうです

https://docs.anthropic.com/ja/docs/claude-code/hooks#構造
👉 hooksが動かない時に確認したいポイント
settings.jsonが正しいか

/hooksでまずブロックを追加して、後からcommandなど追記がいいと思います

実行ファイルのパス、権限
まずはシンプルなコマンドで試す

 >> test.logなどでログを残す

 Notion MCPの設定割愛しますmm

 Slack APIの準備こちらは重要なポイントのみ書きます。

https://api.slack.com/apps から[Create New App]
OAuth & Permissions で以下を付与してWorkspaceにインストール
channels:history（Publicチャンネルの履歴）
groups:history（Privateチャンネルの履歴、Botが参加している必要あり）
channels:read / groups:read（チャンネル情報参照）

Botを該当チャンネルに招待（Privateで取得したい場合は必須）
Bot Token (xoxb-***) を取得

 Pythonコード（Slack API呼び出し）Pythonのコードは2つあります。

1つめがHooksから呼ばれるコード、2つめがそこから呼ばれるSlack API呼び出しのコードです。

 1つめ：Hooksから呼ばれるコードコード全文user_prompt_router.py
#!/usr/bin/env python3
import sys, json, re, subprocess, shlex

data = json.load(sys.stdin)
prompt = data.get("prompt","")

# 特定プロンプトの場合のみ実行するよう分岐
m = re.match(r"^\s*Notionデータベースにまとめて\s+notion:(https?://\S+)\s+slack:(https?://\S+)", prompt)
if m:
    open("/Users/honmayuki/develop/tools/slack-fetch-thread/router_executed.log", "a").write(f"{__import__('datetime').datetime.now()}: user_prompt_router.py executed\n")
    notion_url, slack_url = m.group(1), m.group(2)
    # 任意のコマンドを実行（例: スレッド取得→Notion更新ツール）
    cmd = f'python3 /<your_path>/slack_fetch_thread.py {shlex.quote(slack_url)}'
    try:
        result = subprocess.run(cmd, shell=True, check=True, capture_output=True, text=True)
    except subprocess.CalledProcessError as e:
        # 詳細なエラー情報を記録
        error_log = f"前処理に失敗しました: {e}\nstdout: {e.stdout}\nstderr: {e.stderr}\n"
        open("/<your_path>/router_error.log", "a").write(f"{__import__('datetime').datetime.now()}: {error_log}")
        sys.stderr.write(error_log)
        sys.exit(2)

    # Claudeに追加コンテキストを渡したい場合だけ stdout に出力（exit 0）
    # 何も渡したくなければ print を消す
    print(f"スレッド要約を実行しました。Notion: {notion_url} / Slack: {slack_url}")
    sys.exit(0)

# どのパターンにも該当しない→何もしない（実行もしない）
sys.exit(0)
標準入力からプロンプトを取得することができます
data = json.load(sys.stdin)
prompt = data.get("prompt","")
このフォーマットに一致する場合のみ、次のコードを実行します。これにより、hooksのmatcherのような動作をさせることができます。
# 特定プロンプトの場合のみ実行するよう分岐
m = re.match(r"^\s*Notionデータベースにまとめて\s+notion:(https?://\S+)\s+slack:(https?://\S+)", prompt)
if m:
そして、サマライズを依頼する時はこのフォーマットに合わせてプロンプトします

Notionデータベースにまとめて notion:https://<NotionデータベースのURL> slack:https://<SlackスレッドのURL>
※NotionデータベースのURLは、下記の場所から取得できます


 2つめ：Slack API呼び出しコード全文slack_fetch_thread.py
#!/usr/bin/env python3
"""
Slack Thread Fetcher for Notion MCP Integration

このスクリプトはSlack APIを使用してスレッドの内容を取得し、
Notion MCPで使用可能な形式でデータを整形します。

使用方法:
1. 環境変数 SLACK_BOT_TOKEN を設定
2. スクリプトを実行: python slack_fetch_thread.py <thread_url>
3. 出力されたデータをNotion MCPに渡す

例:
python slack_fetch_thread.py "https://app.slack.com/client/T123ABC456/C123ABC456/thread/T123ABC456-1234567890.123456"
"""

import os
import re
import sys
import json
import requests
from datetime import datetime
from typing import List, Dict, Optional, Any
from urllib.parse import urlparse, parse_qs


class SlackThreadFetcher:
    def __init__(self, bot_token: str):
        """
        Slack Thread Fetcherを初期化
        
        Args:
            bot_token: Slack Bot Token (xoxb-で始まる)
        """
        self.bot_token = bot_token
        self.headers = {
            'Authorization': f'Bearer {bot_token}',
            'Content-Type': 'application/json'
        }
        self.base_url = 'https://slack.com/api'
    
    def parse_slack_url(self, url: str) -> Dict[str, str]:
        """
        Slack URLからチャンネルIDとタイムスタンプを抽出
        
        Args:
            url: Slack thread URL
            
        Returns:
            Dict containing channel_id and thread_ts
        """
        # パターン1: https://app.slack.com/client/WORKSPACE_ID/CHANNEL_ID/thread/WORKSPACE_ID-TIMESTAMP
        pattern1 = r'https://app\.slack\.com/client/([^/]+)/([^/]+)/thread/[^-]+-(.+)'
        
        # パターン2: https://workspace.slack.com/archives/CHANNEL_ID/pTIMESTAMP
        pattern2 = r'https://[^/]+\.slack\.com/archives/([^/]+)/p(\d+)'
        
        match = re.match(pattern1, url)
        if match:
            _, channel_id, timestamp = match.groups()
            # タイムスタンプを正規化（ドットを追加）
            if '.' not in timestamp:
                timestamp = f"{timestamp[:10]}.{timestamp[10:]}"
            return {'channel_id': channel_id, 'thread_ts': timestamp}
        
        match = re.match(pattern2, url)
        if match:
            channel_id, timestamp = match.groups()
            # pプレフィックスを削除してタイムスタンプを正規化
            timestamp = f"{timestamp[:10]}.{timestamp[10:]}"
            return {'channel_id': channel_id, 'thread_ts': timestamp}
        
        raise ValueError(f"不正なSlack URL形式です: {url}")
    
    def get_thread_messages(self, channel_id: str, thread_ts: str) -> List[Dict[str, Any]]:
        """
        スレッドのメッセージを取得
        
        Args:
            channel_id: Slack channel ID
            thread_ts: Thread timestamp
            
        Returns:
            List of message objects
        """
        url = f"{self.base_url}/conversations.replies"
        params = {
            'channel': channel_id,
            'ts': thread_ts,
            'limit': 200  # 最大200メッセージまで取得
        }
        
        response = requests.get(url, headers=self.headers, params=params)
        response.raise_for_status()
        
        data = response.json()
        if not data.get('ok'):
            raise Exception(f"Slack API Error: {data.get('error', 'Unknown error')}")
        
        return data.get('messages', [])
    
    def get_user_info(self, user_id: str) -> Dict[str, str]:
        """
        ユーザー情報を取得
        
        Args:
            user_id: Slack user ID
            
        Returns:
            User information dict
        """
        url = f"{self.base_url}/users.info"
        params = {'user': user_id}
        
        response = requests.get(url, headers=self.headers, params=params)
        response.raise_for_status()
        
        data = response.json()
        if not data.get('ok'):
            return {'name': f'Unknown User ({user_id})', 'real_name': 'Unknown User'}
        
        user = data.get('user', {})
        return {
            'name': user.get('name', 'Unknown'),
            'real_name': user.get('real_name', user.get('name', 'Unknown')),
            'display_name': user.get('profile', {}).get('display_name', ''),
            'email': user.get('profile', {}).get('email', '')
        }
    
    def format_message_for_notion(self, message: Dict[str, Any], user_cache: Dict[str, Dict[str, str]]) -> Dict[str, Any]:
        """
        メッセージをNotion用にフォーマット
        
        Args:
            message: Slack message object
            user_cache: User information cache
            
        Returns:
            Formatted message for Notion
        """
        user_id = message.get('user', 'Unknown')
        
        # ユーザー情報をキャッシュから取得、なければAPIから取得
        if user_id not in user_cache and user_id != 'Unknown':
            try:
                user_cache[user_id] = self.get_user_info(user_id)
            except:
                user_cache[user_id] = {'name': f'Unknown ({user_id})', 'real_name': 'Unknown'}
        
        user_info = user_cache.get(user_id, {'name': 'Unknown', 'real_name': 'Unknown'})
        
        # タイムスタンプを日時に変換
        timestamp = float(message.get('ts', 0))
        formatted_time = datetime.fromtimestamp(timestamp).strftime('%Y-%m-%d %H:%M:%S')
        
        # メッセージテキストを取得（スレッドの場合は改行で区切る）
        text = message.get('text', '')
        
        # ファイルがある場合は追加
        files = []
        if 'files' in message:
            for file in message['files']:
                files.append({
                    'name': file.get('name', 'Unknown file'),
                    'url': file.get('url_private', ''),
                    'type': file.get('mimetype', 'unknown')
                })
        
        return {
            'user_name': user_info['real_name'],
            'user_id': user_id,
            'timestamp': formatted_time,
            'text': text,
            'files': files,
            'is_thread_reply': 'thread_ts' in message and message.get('thread_ts') != message.get('ts')
        }
    
    def fetch_thread_data(self, thread_url: str) -> Dict[str, Any]:
        """
        スレッドのデータを取得してフォーマット
        
        Args:
            thread_url: Slack thread URL
            
        Returns:
            Formatted thread data for Notion
        """
        # URLを解析
        parsed = self.parse_slack_url(thread_url)
        channel_id = parsed['channel_id']
        thread_ts = parsed['thread_ts']
        
        # メッセージを取得
        messages = self.get_thread_messages(channel_id, thread_ts)
        
        if not messages:
            raise Exception("スレッドのメッセージが見つかりませんでした")
        
        # ユーザー情報のキャッシュ
        user_cache = {}
        
        # メッセージをフォーマット
        formatted_messages = []
        for message in messages:
            formatted_message = self.format_message_for_notion(message, user_cache)
            formatted_messages.append(formatted_message)
        
        # スレッドの概要を作成
        thread_summary = {
            'thread_url': thread_url,
            'channel_id': channel_id,
            'thread_ts': thread_ts,
            'total_messages': len(formatted_messages),
            'participants': list(set([msg['user_name'] for msg in formatted_messages])),
            'start_time': formatted_messages[0]['timestamp'] if formatted_messages else None,
            'end_time': formatted_messages[-1]['timestamp'] if formatted_messages else None,
            'messages': formatted_messages
        }
        
        return thread_summary


def main():
    """メイン関数"""
    if len(sys.argv) != 2:
        print("使用方法: python slack_fetch_thread.py <thread_url>")
        sys.exit(1)
    
    thread_url = sys.argv[1]
    
    # 環境変数からBotトークンを取得
    bot_token = os.getenv('SLACK_BOT_TOKEN')
    if not bot_token:
        print("エラー: SLACK_BOT_TOKEN環境変数が設定されていません")
        sys.exit(1)
    
    try:
        fetcher = SlackThreadFetcher(bot_token)
        thread_data = fetcher.fetch_thread_data(thread_url)
        
        # JSON形式で出力（Notion MCPで使用可能）
        print("=== Slack Thread Data (JSON) ===")
        print(json.dumps(thread_data, ensure_ascii=False, indent=2))
        
        # 人間が読みやすい形式でも出力
        print("\n=== Human Readable Summary ===")
        print(f"スレッドURL: {thread_data['thread_url']}")
        print(f"メッセージ数: {thread_data['total_messages']}")
        print(f"参加者: {', '.join(thread_data['participants'])}")
        print(f"開始時間: {thread_data['start_time']}")
        print(f"終了時間: {thread_data['end_time']}")
        
        print("\n=== Messages ===")
        for i, message in enumerate(thread_data['messages'], 1):
            reply_indicator = "└─ " if message['is_thread_reply'] else ""
            print(f"{reply_indicator}[{i}] {message['user_name']} ({message['timestamp']})")
            print(f"    {message['text'][:100]}{'...' if len(message['text']) > 100 else ''}")
            if message['files']:
                print(f"    📎 添付ファイル: {len(message['files'])}個")
            print()
        
    except Exception as e:
        print(f"エラー: {e}")
        sys.exit(1)


if __name__ == "__main__":
    main()
※注意点

Slackのスレッドに大量の投稿や長文が含まれていることは考慮されていません。
実行前に環境変数のセットが必要です。Slack APIの手順4でコピーしたトークンをセットします
export SLACK_BOT_TOKEN=xoxb-***
2つのURLパターンを想定しています
        # パターン1: https://app.slack.com/client/WORKSPACE_ID/CHANNEL_ID/thread/WORKSPACE_ID-TIMESTAMP
        pattern1 = r'https://app\.slack\.com/client/([^/]+)/([^/]+)/thread/[^-]+-(.+)'
        
        # パターン2: https://workspace.slack.com/archives/CHANNEL_ID/pTIMESTAMP
        pattern2 = r'https://[^/]+\.slack\.com/archives/([^/]+)/p(\d+)'
        
        match = re.match(pattern1, url)
Notion MCP統合を意識した構造化データになっているそうです！（claude-4-sonnetさんより）
        # スレッドの概要を作成
        thread_summary = {
            'thread_url': thread_url,
            'channel_id': channel_id,
            'thread_ts': thread_ts,
            'total_messages': len(formatted_messages),
            'participants': list(set([msg['user_name'] for msg in formatted_messages])),
            'start_time': formatted_messages[0]['timestamp'] if formatted_messages else None,
            'end_time': formatted_messages[-1]['timestamp'] if formatted_messages else None,
            'messages': formatted_messages
        }

 まとめ今回作成したものは個人利用を前提とした仕組みになっていますが、Slackチャンネルに参加している誰もが使えるような仕組みも今後は検討していきたいです（元々そっちを記事にしようと考えていたのですが、思ったより時間がかかってしまい、この形になりました…🥺）
また、Claude Codeのhooksは、使えたら便利そうだと思いつつ活用ができていませんでしたが、今回の実装で理解が進んだ点はよかったと思います！

便利そうと思ったらぜひ、真似してみてください！
カンリーテックブログPublication
株式会社カンリーは「店舗経営を支える、世界的なインフラを創る」をミッションに、店舗アカウントの一括管理・分析SaaS「カンリー店舗集客」の開発・提供、他複数のサービスを提供しております。技術系以外のnoteはこちらから note.com/canly
実行結果

構成

Claude Code Hooksの設定

Notion MCPの設定

Slack APIの準備

Pythonコード（Slack API呼び出し）

1つめ：Hooksから呼ばれるコード

2つめ：Slack API呼び出し

まとめ

Discussion
