Claude Code×Slack・Notion API連携【2026】社内の情報共有とナレッジ整理をAIエージェントで自動化

python -m venv .venv && source .venv/bin/activate
pip install slack_sdk notion-client python-dotenv

SLACK_BOT_TOKEN=xoxb-...
# 読んでよいチャンネルIDの許可リスト（カンマ区切り）
SLACK_ALLOWED_CHANNELS=C0123GENERAL,C0456FAQ
SLACK_REMIND_CHANNEL=C0789TEAM
NOTION_TOKEN=ntn_xxxxxxxxxxxx
NOTION_WEEKLY_DB=xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
NOTION_TASK_DB=xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
NOTION_FAQ_PARENT=xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx

office-agent/
├── CLAUDE.md                  # Claude Code が毎回読む手順書
├── .claude/
│   ├── settings.json          # permissions（allow / ask / deny）
│   └── skills/
│       └── weekly-digest/SKILL.md
├── .env                       # トークン類（deny＋.gitignore）
├── scripts/
│   ├── common.py              # クライアント生成・許可リストの確認
│   ├── read/
│   │   ├── slack_fetch.py     # チャンネル履歴 → data/*.json
│   │   ├── notion_overdue.py  # 期限切れタスク → data/overdue.json
│   │   └── notion_page_text.py# 議事録ページ → テキスト
│   └── write/
│       ├── notion_publish_md.py  # drafts/*.md → Notion（週報DB / FAQ）
│       ├── notion_create_tasks.py# drafts/todos.json → タスクDB
│       └── slack_remind.py       # data/overdue.json → Slack投稿
├── data/                      # 取得した生データ（.gitignore）
└── drafts/                    # Claude が書く下書き（人が確認する）

"""Slack / Notion クライアントの生成と、触ってよい範囲のチェックをまとめる"""
import os
import sys
from datetime import timedelta, timezone

from dotenv import load_dotenv
from notion_client import Client
from slack_sdk import WebClient
from slack_sdk.http_retry.builtin_handlers import RateLimitErrorRetryHandler

load_dotenv()
JST = timezone(timedelta(hours=9))
# この記事のコードは従来形式（database_id 指定）で書いているため、バージョンを固定する
NOTION_VERSION = "2022-06-28"


def slack() -> WebClient:
    client = WebClient(token=os.environ["SLACK_BOT_TOKEN"])
    # 429（レート制限）が返ったら Retry-After に従って自動で待って再試行する
    client.retry_handlers.append(RateLimitErrorRetryHandler(max_retry_count=3))
    return client


def notion() -> Client:
    return Client(auth=os.environ["NOTION_TOKEN"], notion_version=NOTION_VERSION)


def allowed_channels() -> set:
    return {c.strip() for c in os.environ.get("SLACK_ALLOWED_CHANNELS", "").split(",") if c.strip()}


def ensure_channel_allowed(client: WebClient, channel_id: str) -> dict:
    """許可リスト外・DM・プライベートチャンネルなら処理を止める"""
    if channel_id not in allowed_channels():
        sys.exit(f"{channel_id} は SLACK_ALLOWED_CHANNELS にありません")
    info = client.conversations_info(channel=channel_id)["channel"]
    if info.get("is_im") or info.get("is_mpim") or info.get("is_private"):
        sys.exit("DM・グループDM・プライベートチャンネルは対象外です")
    return info

"""使い方: python scripts/read/slack_fetch.py C0123GENERAL 7
指定チャンネルの直近N日分（スレッド返信を含む）を data/ に JSON で保存する。読み取り専用。"""
import json
import sys
from datetime import datetime, timedelta
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from common import JST, ensure_channel_allowed, slack  # noqa: E402

client = slack()
_names: dict = {}


def display_name(user_id: str) -> str:
    if not user_id:
        return "(unknown)"
    if user_id not in _names:
        profile = client.users_info(user=user_id)["user"]["profile"]
        _names[user_id] = profile.get("display_name") or profile.get("real_name") or user_id
    return _names[user_id]


def to_item(m: dict) -> dict:
    ts = datetime.fromtimestamp(float(m["ts"]), JST)
    return {"time": ts.strftime("%m/%d %H:%M"), "user": display_name(m.get("user", "")), "text": m.get("text", "")}


def fetch(channel_id: str, days: int) -> list:
    ensure_channel_allowed(client, channel_id)
    oldest = (datetime.now(JST) - timedelta(days=days)).timestamp()
    items, cursor = [], None
    while True:
        resp = client.conversations_history(channel=channel_id, oldest=str(oldest), limit=200, cursor=cursor)
        for m in resp["messages"]:
            if m.get("subtype"):  # 入退室・ボット投稿などは除外
                continue
            item = to_item(m)
            if m.get("reply_count"):
                thread = client.conversations_replies(channel=channel_id, ts=m["ts"], limit=200)
                item["replies"] = [to_item(r) for r in thread["messages"][1:]]
            items.append(item)
        cursor = resp.get("response_metadata", {}).get("next_cursor")
        if not cursor:
            break
    return list(reversed(items))  # 新しい順で返るので古い順に並べ直す


if __name__ == "__main__":
    channel, days = sys.argv[1], int(sys.argv[2]) if len(sys.argv) > 2 else 7
    data = fetch(channel, days)
    out = Path("data") / f"slack_{channel}_{datetime.now(JST):%Y%m%d}.json"
    out.parent.mkdir(exist_ok=True)
    out.write_text(json.dumps(data, ensure_ascii=False, indent=1), encoding="utf-8")
    print(f"{len(data)} messages -> {out}")

"""使い方: python scripts/read/notion_overdue.py
タスクDBから「期限が今日より前 かつ ステータスが完了以外」を抽出し data/overdue.json に保存する。"""
import json
import os
import sys
from datetime import datetime
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from common import JST, notion  # noqa: E402

client = notion()
today = datetime.now(JST).date().isoformat()
query_filter = {
    "and": [
        {"property": "期限", "date": {"before": today}},
        {"property": "ステータス", "status": {"does_not_equal": "完了"}},
    ]
}


def plain(rich: list) -> str:
    return "".join(t.get("plain_text", "") for t in rich)


tasks, cursor = [], None
while True:
    params = {"database_id": os.environ["NOTION_TASK_DB"], "filter": query_filter, "page_size": 100}
    if cursor:
        params["start_cursor"] = cursor
    # 従来形式のDBクエリ。新しいAPIバージョンでは data source 側の query に移っている（公式で確認）
    resp = client.databases.query(**params)
    for page in resp["results"]:
        props = page["properties"]
        tasks.append({
            "title": plain(props["名前"]["title"]),
            "due": props["期限"]["date"]["start"],
            "assignee_emails": [p.get("person", {}).get("email") for p in props["担当"]["people"]],
            "url": page["url"],
        })
    if not resp["has_more"]:
        break
    cursor = resp["next_cursor"]

Path("data").mkdir(exist_ok=True)
Path("data/overdue.json").write_text(json.dumps(tasks, ensure_ascii=False, indent=1), encoding="utf-8")
print(f"{len(tasks)} overdue tasks -> data/overdue.json")

"""使い方: python scripts/read/notion_page_text.py <page_id>
ページ本文をブロック単位でたどり、見出し・箇条書き・チェックボックスを含むテキストとして出力する。"""
import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from common import notion  # noqa: E402

client = notion()
PREFIX = {"heading_1": "# ", "heading_2": "## ", "heading_3": "### ", "bulleted_list_item": "- ", "numbered_list_item": "1. "}


def walk(block_id: str, depth: int = 0) -> None:
    cursor = None
    while True:
        params = {"block_id": block_id, "page_size": 100}
        if cursor:
            params["start_cursor"] = cursor
        resp = client.blocks.children.list(**params)
        for b in resp["results"]:
            body = b.get(b["type"], {})
            text = "".join(t["plain_text"] for t in body.get("rich_text", []))
            if b["type"] == "to_do":
                prefix = "[x] " if body.get("checked") else "[ ] "
            else:
                prefix = PREFIX.get(b["type"], "")
            if text:
                print("  " * depth + prefix + text)
            if b.get("has_children"):
                walk(b["id"], depth + 1)
        if not resp["has_more"]:
            break
        cursor = resp["next_cursor"]


if __name__ == "__main__":
    walk(sys.argv[1])

"""使い方:
  python scripts/write/notion_publish_md.py weekly drafts/weekly_20260926.md          # ドライラン
  python scripts/write/notion_publish_md.py weekly drafts/weekly_20260926.md --commit # 登録
  python scripts/write/notion_publish_md.py faq drafts/faq_20260926.md --commit
下書きの1行目（# で始まる行）をタイトルにし、残りをブロックに変換して登録する。"""
import argparse
import json
import os
import sys
from datetime import datetime, timedelta
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from common import JST, notion  # noqa: E402

MAX_BLOCKS = 100  # 1リクエストで渡せる子ブロック数の上限
MAX_TEXT = 2000   # 1つのテキスト要素の文字数上限


def to_blocks(lines: list) -> list:
    blocks = []
    for line in (ln.rstrip() for ln in lines):
        if not line:
            continue
        if line.startswith("## "):
            kind, text = "heading_2", line[3:]
        elif line.startswith("- "):
            kind, text = "bulleted_list_item", line[2:]
        else:
            kind, text = "paragraph", line
        blocks.append({"object": "block", "type": kind,
                       kind: {"rich_text": [{"type": "text", "text": {"content": text[:MAX_TEXT]}}]}})
    if len(blocks) > MAX_BLOCKS:
        sys.exit(f"ブロック数が {len(blocks)} あります。下書きを短くしてください（上限 {MAX_BLOCKS}）")
    return blocks


def build_request(target: str, title: str, blocks: list) -> dict:
    title_value = {"title": [{"text": {"content": title}}]}
    if target == "weekly":
        today = datetime.now(JST).date()
        return {
            "parent": {"database_id": os.environ["NOTION_WEEKLY_DB"]},
            "properties": {
                "名前": title_value,
                "対象期間": {"date": {"start": (today - timedelta(days=6)).isoformat(), "end": today.isoformat()}},
                "状態": {"select": {"name": "要確認"}},
            },
            "children": blocks,
        }
    # FAQ はページの子ページとして作る（親がページのときのプロパティは title だけ）
    return {"parent": {"page_id": os.environ["NOTION_FAQ_PARENT"]}, "properties": {"title": title_value}, "children": blocks}


if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("target", choices=["weekly", "faq"])
    ap.add_argument("draft")
    ap.add_argument("--commit", action="store_true", help="付けたときだけ実際に登録する")
    args = ap.parse_args()

    draft = Path(args.draft)
    if draft.parent.name != "drafts":
        sys.exit("drafts/ 配下のファイルだけ登録できます")
    lines = draft.read_text(encoding="utf-8").splitlines()
    title = lines[0].lstrip("# ").strip()
    body = build_request(args.target, title, to_blocks(lines[1:]))

    if not args.commit:
        print(json.dumps(body, ensure_ascii=False, indent=1)[:3000])
        print("\n(dry-run) 内容を確認し、問題なければ --commit を付けて再実行してください")
        sys.exit(0)
    page = notion().pages.create(**body)
    print("created:", page["url"])

"""使い方: python scripts/write/notion_create_tasks.py drafts/todos.json [--commit]
todos.json の形式: [{"title": "...", "due": "2026-10-03", "owner": "山田", "source_url": "https://..."}]"""
import argparse
import json
import os
import sys
from datetime import date
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from common import notion  # noqa: E402

MAX_TASKS = 15


def validate(todos: list) -> None:
    if not isinstance(todos, list) or not todos:
        sys.exit("ToDoが空です")
    if len(todos) > MAX_TASKS:
        sys.exit(f"{len(todos)}件は多すぎます（上限{MAX_TASKS}件）。議事録を分けてください")
    for i, t in enumerate(todos):
        if not t.get("title") or len(t["title"]) > 80:
            sys.exit(f"{i}件目: title が空か長すぎます")
        if t.get("due"):
            date.fromisoformat(t["due"])  # 形式が不正なら例外で止まる


def to_properties(t: dict) -> dict:
    props = {
        "名前": {"title": [{"text": {"content": t["title"]}}]},
        "ステータス": {"status": {"name": "未着手"}},
        # 担当は people 型に入れるにはユーザーIDが要るため、まずは仮の名前をテキストで入れて人が付け替える
        "担当（仮）": {"rich_text": [{"text": {"content": t.get("owner", "")}}]},
        "出典": {"url": t.get("source_url") or None},
    }
    if t.get("due"):
        props["期限"] = {"date": {"start": t["due"]}}
    return props


if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("path")
    ap.add_argument("--commit", action="store_true")
    args = ap.parse_args()
    todos = json.loads(Path(args.path).read_text(encoding="utf-8"))
    validate(todos)
    for t in todos:
        print(f"- {t['title']} / 期限 {t.get('due') or '未設定'} / 担当 {t.get('owner') or '未定'}")
    if not args.commit:
        print("\n(dry-run) 上記を起票します。問題なければ --commit を付けて再実行してください")
        sys.exit(0)
    client = notion()
    for t in todos:
        client.pages.create(parent={"database_id": os.environ["NOTION_TASK_DB"]}, properties=to_properties(t))
    print(f"{len(todos)} tasks created")

"""使い方: python scripts/write/slack_remind.py [--commit]
data/overdue.json を担当者ごとにまとめ、SLACK_REMIND_CHANNEL にメンション付きで1通だけ投稿する。"""
import argparse
import json
import os
import sys
from collections import defaultdict
from pathlib import Path

from slack_sdk.errors import SlackApiError

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
from common import slack  # noqa: E402

client = slack()


def mention(email: str) -> str:
    if not email:
        return "担当未設定"
    try:
        uid = client.users_lookupByEmail(email=email)["user"]["id"]
        return f"<@{uid}>"
    except SlackApiError:
        return email  # 見つからなければメンションせず文字で出す


if __name__ == "__main__":
    ap = argparse.ArgumentParser()
    ap.add_argument("--commit", action="store_true")
    args = ap.parse_args()
    tasks = json.loads(Path("data/overdue.json").read_text(encoding="utf-8"))
    if not tasks:
        sys.exit("期限切れタスクはありません")

    by_owner = defaultdict(list)
    for t in tasks:
        for email in (t["assignee_emails"] or [""]):
            by_owner[email].append(t)

    lines = [":spiral_calendar_pad: 期限を過ぎているタスクがあります（Notionのステータス更新もお願いします）"]
    for email, items in by_owner.items():
        lines.append(f"\n{mention(email)}")
        lines += [f"• <{t['url']}|{t['title']}>（期限 {t['due']}）" for t in items]
    text = "\n".join(lines)

    print(text)
    if not args.commit:
        print("\n(dry-run) 問題なければ --commit を付けて再実行してください")
        sys.exit(0)
    client.chat_postMessage(channel=os.environ["SLACK_REMIND_CHANNEL"], text=text)
    print("posted")

# CLAUDE.md（社内情報共有エージェント）

## 役割
- Slack / Notion のデータを読んで、要約・ToDo抽出・FAQ下書きを drafts/ に作る
- Slack への投稿・Notion への登録は、必ず人が drafts/ を確認したあとに write 系スクリプトで行う

## データの取り方（APIを直接叩かない。必ずスクリプト経由）
- Slack履歴: `python scripts/read/slack_fetch.py <channel_id> 7` → data/slack_*.json
- 期限切れタスク: `python scripts/read/notion_overdue.py` → data/overdue.json
- 議事録本文: `python scripts/read/notion_page_text.py <page_id>`
- 対象チャンネルは .env の許可リストにあるものだけ。DM・プライベートチャンネルは扱わない

## 下書きのルール
- 週報: drafts/weekly_YYYYMMDD.md。1行目は「# 週報 MM/DD〜MM/DD #チャンネル名」
  - 「## 決まったこと」「## 進行中」「## 困りごと・相談」「## 来週の予定」の4見出し
  - 各項目の末尾に元メッセージの日時（例: 09/24 14:05）を付け、根拠をたどれるようにする
  - 推測で補わない。決定か検討中か曖昧なものは「検討中」に入れる
- ToDo: drafts/todos.json（形式は scripts/write/notion_create_tasks.py の docstring 参照）
  - 議事録に担当・期限が書かれていないものは空欄にする。勝手に決めない
- FAQ: drafts/faq_YYYYMMDD.md。「## Q. 質問」「回答」の繰り返し。回答が割れている質問は載せず、一覧で報告する

## 書き出してはいけない情報
- 個人の評価・人事・健康・家庭の事情、給与や契約条件、顧客の個人情報
- 「社外秘」「未発表」と書かれた話題
- 見つけたら下書きから除き、除いたことだけを報告する（中身は書かない）

## write 系スクリプト
- まず --commit なしでドライランし、出力を人に見せて確認を求める
- 人が「OK」と言ったときだけ --commit を付けて再実行する

{
  "permissions": {
    "allow": [
      "Bash(python scripts/read/slack_fetch.py:*)",
      "Bash(python scripts/read/notion_overdue.py:*)",
      "Bash(python scripts/read/notion_page_text.py:*)",
      "Read(./data/**)",
      "Read(./drafts/**)",
      "Edit(./drafts/**)"
    ],
    "ask": [
      "Bash(python scripts/write/notion_publish_md.py:*)",
      "Bash(python scripts/write/notion_create_tasks.py:*)",
      "Bash(python scripts/write/slack_remind.py:*)"
    ],
    "deny": [
      "Read(./.env)",
      "Read(./.env.*)",
      "Edit(./scripts/**)",
      "Bash(curl:*)",
      "Bash(wget:*)",
      "Bash(env:*)",
      "Bash(printenv:*)",
      "Bash(python -c:*)"
    ]
  }
}

> #proj-renewal（C0123GENERAL）の今週分を週報にして、Notionの週報DBに登録して。

● Bash(python scripts/read/slack_fetch.py C0123GENERAL 7)
  ⎿ 86 messages -> data/slack_C0123GENERAL_20260926.json
● Read(data/slack_C0123GENERAL_20260926.json)
● Write(drafts/weekly_20260926.md)

  下書きを drafts/weekly_20260926.md に作成しました。
  - 決まったこと 3件 / 進行中 5件 / 困りごと 2件 / 来週の予定 4件
  - 09/23 の「〇〇さんの異動」に関するやり取りは、人事に関わるため除外しました
  - 「リリース日を10/15にする」は賛成が多いものの決定の発言がないため「検討中」に入れています
  内容を確認して、問題なければ登録します。

> 困りごとの2件目は解決済みなので消して。あとはOK。

● Edit(drafts/weekly_20260926.md)
● Bash(python scripts/write/notion_publish_md.py weekly drafts/weekly_20260926.md)
  ⎿ (dry-run) ...
● Bash(python scripts/write/notion_publish_md.py weekly drafts/weekly_20260926.md --commit)
                                             ← ask 設定のため承認を求められる
  ⎿ created: https://www.notion.so/xxxxxxxx

> 今日の定例の議事録（page_id: 1a2b...）からToDoを抜き出して、タスクDBに起票して。

● Bash(python scripts/read/notion_page_text.py 1a2b...)
● Write(drafts/todos.json)

  ToDoを6件抽出しました（drafts/todos.json）。
  | # | ToDo | 担当 | 期限 |
  | 1 | 新料金表のドラフト作成 | 佐藤 | 2026-10-02 |
  | 2 | 取引先A社への日程再調整の連絡 | （議事録に記載なし） | （記載なし） |
  ...
  2件目は担当・期限が議事録にないため空欄です。起票前に決めてください。

> 2件目は田中さん、期限は来週月曜で。

● Edit(drafts/todos.json)
● Bash(python scripts/write/notion_create_tasks.py drafts/todos.json --commit)  ← 承認
  ⎿ 6 tasks created

claude mcp add --transport http notion https://mcp.notion.com/mcp
# Claude Code 内で /mcp を開き、認証（ブラウザでNotionにログインして許可）と接続状況を確認

---
name: weekly-digest
description: Slackの指定チャンネルの1週間分を要約し、Notionの週報DBに「要確認」で登録する。週報、今週のまとめ、チャンネルの要約を頼まれたときに使う。
---

# 週報づくりの手順

1. 対象チャンネルIDを確認する（.env の SLACK_ALLOWED_CHANNELS にあるものだけ）
2. `python scripts/read/slack_fetch.py <channel_id> 7` を実行する
3. CLAUDE.md の「下書きのルール」に従い drafts/weekly_YYYYMMDD.md を書く
4. 除外した話題（件数と理由の種類だけ）と、判断に迷った項目を人に報告する
5. 人のOKが出たら notion_publish_md.py weekly をドライラン → --commit の順で実行する

#!/usr/bin/env bash
# 毎週金曜 17:00 に cron 等から実行。下書き作成までで止める
set -euo pipefail
cd "$(dirname "$0")/.."

claude -p "weekly-digest の手順1〜4を実行して、C0123GENERAL の週報の下書きを作って。登録はしないこと。" \
  --allowedTools "Bash(python scripts/read/slack_fetch.py:*)" "Read" "Edit(./drafts/**)" \
  --output-format json > "logs/weekly_$(date +%Y%m%d).json"

# 下書きができたことだけを担当者に知らせる（本文は送らない）
curl -s -X POST -H 'Content-type: application/json' \
  --data '{"text":"今週の週報の下書きができました。drafts/ を確認してください"}' \
  "$SLACK_NOTIFY_WEBHOOK_URL"