LoginSignup
1
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@BNR-Longin株式会社BinaryTech | 日本品質 × ベトナム開発、最高のバランスを。

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

Posted at

はじめに

第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テスト用のデータ。

インストールコマンド（必要に応じて）：

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または公式サイト）。
    • ローカルサーバーを公開： 
      ngrok http 8120
    • 生成されたURLを記録（例：https://abc123.ngrok.io）。
  4. 環境変数：
    第3章の.envファイルに以下を確認： 
    NOTION_TOKEN=your_token
NOTION_PAGE_ID=your_page_id
NOTION_DATABASE_ID=your_database_id

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

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

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起動：

    ngrok http 8120

    ngrok URL（例：https://abc123.ngrok.io）を記録。

  2. サーバー起動：

    python notion_realtime_server.py

    コンソールに「NotionリアルタイムMCPサーバーを起動中: http://localhost:8120」と「Webhookサーバーを起動中: http://localhost:8120」が表示。

  3. 更新検知のテスト：

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

  4. 最新更新取得のテスト：
    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))

    期待されるレスポンス：

    {
  "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. コメント追加のテスト：

    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）に以下を追加：

    {
  "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の未来に興味がある方は、ぜひお楽しみに！

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

1
1
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
1
1

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?