NotionでAIを強化する | 第4章：リアルタイム管理AIの強化：Webhook連携

## はじめに

第3章では、**Notion**のデータベースを活用してデータ分析エージェントを構築し、完了率や期限切れタスクを分析しました。これにより、AIがプロジェクトのインサイトを提供し、プロセス改善を支援できるようになりました。今回は、NotionのWebhook機能を利用して、データベースやページの更新をリアルタイムで検知する**リアルタイム管理AI**を構築します。

この第4章では、MCPサーバーにWebhookを統合し、Notionデータベースのエントリ追加やステータス変更を即座に捕捉します。AIはこれを利用して、変更内容をコメントとして記録したり、重要な更新をチームに通知したりできます。コード例とステップごとのガイドで、リアルタイム管理AIの構築を体験しましょう。さあ、始めましょう！

## リアルタイム管理AIとは？

リアルタイム管理AIは、Notionデータベースやページの変更を監視し、プロジェクトの動的な管理を支援するエージェントです。MCPサーバーとNotionのWebhookを組み合わせることで、以下のような機能を実現できます：
- **変更検知**：データベースエントリの追加、ステータス変更、期限更新をリアルタイムで捕捉。
- **コメント生成**：変更内容を要約し、Notionにコメントとして投稿。
- **通知**：重要な変更をチームに通知（例：Slackやメール）。

### ユースケース
- プロジェクト管理：タスクが`Done`に変更された際にコメントを追加。
- チームコラボレーション：期限変更を即座にチームに通知。
- プロセス監視：意図しない変更（例：期限の削除）を検出。

## 開発環境の準備

第3章の環境を基に、以下の追加準備を行います：
- **Python 3.8以降**、**mcpライブラリ**、**requestsライブラリ**、**Claude Desktop**：これまでと同じ。
- **python-dotenv**：環境変数の管理（既にインストール済み）。
- **ngrok**：ローカルサーバーを公開し、Webhookを受信。
- **Notionデータベース**：Webhookテスト用のデータ。

インストールコマンド（必要に応じて）：
```bash
pip install requests python-dotenv
```

### Notionのセットアップ
1. **データベース準備**：
   - 第3章のNotionデータベース（例：`タスクデータベース`）を使用。
   - プロパティを確認（例：`Name`（title）、`Status`（select）、`Due Date`（date））。
   - テスト用にエントリを追加・更新（例：ステータスを`To Do`から`In Progress`に変更）。
2. **Webhookの設定**：
   - NotionはネイティブWebhookを提供しないため、インテグレーション経由で変更を検知します（例：定期ポーリングや外部サービスとの連携）。
   - 代替として、Notion APIの`database/query`エンドポイントを使用し、変更をシミュレート（本番ではAutomationツールやサードパーティWebhookを推奨）。
   - Webhookシミュレーション用にngrokでローカルサーバーを公開。
3. **ngrokの設定**：
   - ngrokをインストール（`brew install ngrok`または公式サイト）。
   - ローカルサーバーを公開：
     ```bash
     ngrok http 8120
     ```
   - 生成されたURLを記録（例：`https://abc123.ngrok.io`）。
4. **環境変数**：
   第3章の`.env`ファイルに以下を確認：
   ```bash
   NOTION_TOKEN=your_token
   NOTION_PAGE_ID=your_page_id
   NOTION_DATABASE_ID=your_database_id
   ```

## コード例：リアルタイム管理用MCPサーバー

以下のMCPサーバーは、Notionデータベースの変更を検知（シミュレーション）し、コメントを追加する機能を提供します。

```python
from mcp import MCPServer
import os
from dotenv import load_dotenv
import requests
from http.server import BaseHTTPRequestHandler, HTTPServer
import json
import threading
import time

class NotionRealtimeServer(MCPServer):
    def __init__(self, host, port, token, database_id):
        super().__init__(host, port)
        self.token = token
        self.database_id = database_id
        self.base_url = "https://api.notion.com/v1"
        self.headers = {
            "Authorization": f"Bearer {token}",
            "Notion-Version": "2022-06-28",
            "Content-Type": "application/json"
        }
        self.latest_update = None
        self.last_checked = None
        self.register_resource("get_latest_update", self.get_latest_update)
        self.register_tool("add_comment", self.add_comment)
        self.start_webhook_server()

    def check_for_updates(self):
        try:
            url = f"{self.base_url}/databases/{self.database_id}/query"
            response = requests.post(url, headers=self.headers, json={})
            response.raise_for_status()
            entries = response.json()["results"]
            for entry in entries:
                last_edited = entry["last_edited_time"]
                if not self.last_checked or last_edited > self.last_checked:
                    self.latest_update = {
                        "action": {
                            "data": {
                                "entry": {
                                    "id": entry["id"],
                                    "name": entry["properties"].get("Name", {}).get("title", [{}])[0].get("plain_text", ""),
                                    "status": entry["properties"].get("Status", {}).get("select", {}).get("name", "")
                                }
                            },
                            "type": "updateEntry",
                            "date": last_edited
                        }
                    }
                    self.last_checked = last_edited
            return {"status": "success"}
        except Exception as e:
            return {"status": "error", "message": str(e)}

    def get_latest_update(self, params):
        try:
            self.check_for_updates()
            if self.latest_update:
                action = self.latest_update["action"]
                update_info = {
                    "entry_id": action["data"]["entry"]["id"],
                    "entry_name": action["data"]["entry"]["name"],
                    "action_type": action["type"],
                    "status": action["data"]["entry"]["status"],
                    "date": action["date"]
                }
                return {"status": "success", "update_info": update_info}
            return {"status": "success", "update_info": None, "message": "更新なし"}
        except Exception as e:
            return {"status": "error", "message": str(e)}

    def add_comment(self, params):
        try:
            entry_id = params.get("entry_id", "")
            message = params.get("message", "")
            if not entry_id or not message:
                return {"status": "error", "message": "エントリIDとコメントが必要です"}
            
            url = f"{self.base_url}/comments"
            payload = {
                "parent": {"page_id": entry_id},
                "rich_text": [{"text": {"content": message}}]
            }
            response = requests.post(url, headers=self.headers, json=payload)
            response.raise_for_status()
            comment = response.json()
            return {"status": "success", "comment_id": comment["id"]}
        except Exception as e:
            return {"status": "error", "message": str(e)}

    def start_webhook_server(self):
        class WebhookHandler(BaseHTTPRequestHandler):
            def do_POST(self):
                content_length = int(self.headers["Content-Length"])
                post_data = self.rfile.read(content_length)
                self.server.parent.latest_update = json.loads(post_data.decode("utf-8"))
                self.send_response(200)
                self.end_headers()
                self.wfile.write(b"Webhook received")

        server = HTTPServer(("localhost", 8120), WebhookHandler)
        server.parent = self
        threading.Thread(target=server.serve_forever, daemon=True).start()
        print("Webhookサーバーを起動中: http://localhost:8120")

if __name__ == "__main__":
    load_dotenv()
    server = NotionRealtimeServer(
        host="localhost",
        port=8120,
        token=os.getenv("NOTION_TOKEN"),
        database_id=os.getenv("NOTION_DATABASE_ID")
    )
    print("NotionリアルタイムMCPサーバーを起動中: http://localhost:8120")
    server.start()
```

### コードの説明
- **check_for_updates**：データベースをポーリングし、最新の変更を検知（NotionのWebhookがないためシミュレーション）。
- **get_latest_update**：最新のデータベース更新情報を取得（エントリID、名前、ステータス、更新日）。
- **add_comment**：指定したエントリにコメントを追加。
- **start_webhook_server**：ローカルWebhookサーバーを起動（外部ツールからのWebhook受信を想定）。

### 前提条件
- Notionデータベースが存在し、エントリが追加・更新されている。
- `.env`ファイルに正しい`NOTION_TOKEN`と`NOTION_DATABASE_ID`が設定済み。
- APIトークンに読み書きおよびコメント権限がある。
- ngrokが設定済み（本番ではAutomationツール推奨）。

## サーバーのテスト

サーバーが正しく動作するか確認します：

1. **ngrok起動**：
   ```bash
   ngrok http 8120
   ```
   ngrok URL（例：`https://abc123.ngrok.io`）を記録。

2. **サーバー起動**：
   ```bash
   python notion_realtime_server.py
   ```
   コンソールに「NotionリアルタイムMCPサーバーを起動中: http://localhost:8120」と「Webhookサーバーを起動中: http://localhost:8120」が表示。

3. **更新検知のテスト**：
   - NotionのUIでデータベースエントリのステータスを変更（例：`To Do`から`In Progress`）。
   - サーバーが変更を検知（ポーリングによる）。

4. **最新更新取得のテスト**：
   Pythonでリクエストを送信：
   ```python
   import requests
   import json

   url = "http://localhost:8120"
   payload = {
       "jsonrpc": "2.0",
       "method": "get_latest_update",
       "params": {},
       "id": 1
   }
   response = requests.post(url, json=payload)
   print(json.dumps(response.json(), indent=2, ensure_ascii=False))
   ```
   期待されるレスポンス：
   ```json
   {
     "jsonrpc": "2.0",
     "result": {
       "status": "success",
       "update_info": {
         "entry_id": "entry123",
         "entry_name": "コードレビュー",
         "action_type": "updateEntry",
         "status": "In Progress",
         "date": "2025-04-22T10:00:00.000Z"
       }
     },
     "id": 1
   }
   ```

5. **コメント追加のテスト**：
   ```python
   payload = {
       "jsonrpc": "2.0",
       "method": "add_comment",
       "params": {
         "entry_id": "entry123",
         "message": "ステータスがIn Progressに変更されました"
       },
       "id": 2
   }
   response = requests.post(url, json=payload)
   print(json.dumps(response.json(), indent=2, ensure_ascii=False))
   ```

## Claude Desktopとの接続

サーバーをClaude Desktopに接続します：

1. **設定ファイルの編集**：
   Claude Desktopの設定ファイル（例：`claude_desktop_config.json`）に以下を追加：
   ```json
   {
     "mcp_servers": [
       {
         "name": "NotionRealtimeServer",
         "url": "http://localhost:8120",
         "auth": "none"
       }
     ]
   }
   ```

2. **Claudeでテスト**：
   Claude Desktopを起動し、プロンプトを入力：
   ```
   最新のデータベース更新を教えてください。
   ```
   レスポンス例：
   ```
   最新のデータベース更新：
   - エントリ：コードレビュー
   - アクション：ステータスがIn Progressに変更
   - 日時：2025-04-22 10:00
   ```

   別のプロンプト：
   ```
   エントリ「entry123」にステータス変更のコメントを追加してください。
   ```
   レスポンス例：
   ```
   エントリ「コードレビュー」に「ステータスがIn Progressに変更されました」をコメントしました。
   ```

## 実装のコツと注意点

- **Webhookの制限**：NotionはネイティブWebhookをサポートしないため、ポーリングやサードパーティ（例：Zapier、Make）を検討。本番環境ではAutomationツールを活用。
- **レートリミティング**：Notion APIの制限（通常3リクエスト/秒）に注意。
- **セキュリティ**：本番環境では、Webhookリクエストの検証を強化（例：トークン認証）。
- **テスト**：テスト用データベースを作成し、本番データに影響を与えない。
- **拡張性**：大量の更新を処理する場合、キャッシュやキュー（例：Redis）を検討。

## 試してみよう：挑戦課題
以下の機能を追加して、エージェントを強化してみてください：
- 特定プロパティ（例：`Status`）の変更のみを検知するフィルター。
- 更新データをSlackに通知するツール。
- 期限変更を検知し、自動でコメントを追加する機能。

## まとめと次のステップ

この第4章では、Notionのデータベース更新を検知するリアルタイム管理AIを構築しました。変更をリアルタイムで捕捉し、AIがコメントや通知を生成できるようになりました。

次の第5章では、MCPサーバーの**最適化とコミュニティへの貢献**に焦点を当てます。サーバーのパフォーマンス向上、セキュリティ強化、そしてNotion用MCPサーバーをオープンソースとして共有する方法を学びます。コミュニティAIの未来に興味がある方は、ぜひお楽しみに！

---

役に立ったと思ったら、「いいね」や「ストック」をしていただけると嬉しいです！次の章でまたお会いしましょう！