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を強化する | 第3章：データベース分析：AIによるインサイト生成

Posted at

はじめに

第2章では、Notionにページとデータベースエントリを追加するMCPサーバーを構築し、タスク自動化エージェントを実現しました。これにより、AIがタスクページを作成したり、データベースにエントリを追加したりできるようになり、ノート管理の効率が向上しました。今回は、この基盤を活用して、Notionのデータベースを解析するデータ分析エージェントを構築します。

この第3章では、Notionデータベースのタスクデータを分析し、完了率、期限切れタスク、優先順位別の分布などのインサイトを生成します。たとえば、AIが「To Doステータスのタスクが多すぎる」と指摘したり、期限切れのタスクを特定したりできます。コード例とステップごとのガイドで、データ分析AIの構築を体験しましょう。さあ、始めましょう！

データ分析エージェントとは？

データ分析エージェントは、Notionのデータベースを解析し、タスクやプロジェクトの進捗に関するインサイトを提供するAIです。MCPサーバーを介して、以下のような機能を実現できます：

  • 完了率分析：データベース全体のタスク完了率を計算。
  • 期限分析：期限切れや近日中のタスクを特定。
  • 分布分析：ステータスや優先順位ごとのタスク分布を評価。

ユースケース

  • プロジェクト管理：AIが遅延リスクを警告し、優先順位を提案。
  • チームパフォーマンス：タスク完了率や効率を分析。
  • プロセス改善：ボトルネックを特定し、ワークフローを最適化。

開発環境の準備

第2章の環境を基に、以下の準備を行います：

  • Python 3.8以降、mcpライブラリ、requestsライブラリ、Claude Desktop：これまでと同じ。
  • python-dotenv：環境変数の管理（既にインストール済み）。
  • Notionデータベース：分析用のデータを含むデータベース。

Notionのセットアップ

  1. データベース準備：
    • 第2章のNotionデータベース（例：タスクデータベース）を使用。
    • プロパティを確認（例：Name（title）、Status（select）、Due Date（date）、Priority（select））。
    • データ追加：
      • 複数のタスクを追加（例：ステータスTo Do、In Progress、Done）。
      • 期限（例：2025-04-22）や優先順位（例：High、Medium、Low）を設定。
    • データベースIDを記録（Notion UIのURLやAPIで取得）。
  2. APIトークンの確認：
    • 第2章のトークンを再利用（Read Content権限が必要）。
    • インテグレーションがデータベースにアクセス可能であることを確認。
  3. 環境変数：
    第2章の.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 datetime import datetime, timezone
import statistics

class NotionAnalysisServer(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.register_resource("analyze_tasks", self.analyze_tasks)

    def get_database_entries(self):
        try:
            url = f"{self.base_url}/databases/{self.database_id}/query"
            response = requests.post(url, headers=self.headers, json={})
            response.raise_for_status()
            return response.json()["results"]
        except Exception as e:
            return {"status": "error", "message": str(e)}

    def analyze_tasks(self, params):
        try:
            entries = self.get_database_entries()
            if isinstance(entries, dict) and "status" in entries:
                return entries

            total_tasks = len(entries)
            status_counts = {"To Do": 0, "In Progress": 0, "Done": 0}
            overdue_tasks = []
            priority_counts = {"High": 0, "Medium": 0, "Low": 0}
            completion_times = []

            now = datetime.now(timezone.utc)
            for entry in entries:
                properties = entry["properties"]
                status = properties.get("Status", {}).get("select", {}).get("name", "")
                due_date = properties.get("Due Date", {}).get("date", {}).get("start", "")
                priority = properties.get("Priority", {}).get("select", {}).get("name", "")
                created_time = entry["created_time"]
                last_edited_time = entry["last_edited_time"]

                # ステータス分布
                if status in status_counts:
                    status_counts[status] += 1

                # 期限切れタスク
                if due_date and status != "Done":
                    due = datetime.fromisoformat(due_date.replace("Z", "+00:00"))
                    if due < now:
                        overdue_tasks.append({
                            "id": entry["id"],
                            "title": properties.get("Name", {}).get("title", [{}])[0].get("plain_text", ""),
                            "due_date": due_date
                        })

                # 優先順位分布
                if priority in priority_counts:
                    priority_counts[priority] += 1

                # 完了時間（Doneタスクの場合）
                if status == "Done":
                    created = datetime.fromisoformat(created_time.replace("Z", "+00:00"))
                    completed = datetime.fromisoformat(last_edited_time.replace("Z", "+00:00"))
                    completion_times.append((completed - created).total_seconds() / 3600)

            completion_rate = (status_counts["Done"] / total_tasks * 100) if total_tasks > 0 else 0
            avg_completion_time = statistics.mean(completion_times) if completion_times else 0

            return {
                "status": "success",
                "analysis": {
                    "completion_rate": f"{completion_rate:.2f}%",
                    "avg_completion_time_hours": f"{avg_completion_time:.2f}",
                    "status_counts": status_counts,
                    "overdue_tasks": overdue_tasks,
                    "priority_counts": priority_counts
                }
            }
        except Exception as e:
            return {"status": "error", "message": str(e)}

if __name__ == "__main__":
    load_dotenv()
    server = NotionAnalysisServer(
        host="localhost",
        port=8119,
        token=os.getenv("NOTION_TOKEN"),
        database_id=os.getenv("NOTION_DATABASE_ID")
    )
    print("Notion分析MCPサーバーを起動中: http://localhost:8119")
    server.start()

コードの説明

  • get_database_entries：データベースの全エントリを取得。
  • analyze_tasks：データベースエントリを解析し、以下のインサイトを生成：
    • 完了率：Doneステータスのタスク数を全体で割る。
    • 平均完了時間：Doneタスクの作成から完了までの時間を計算。
    • ステータス分布：各ステータス（To Do、In Progress、Done）のタスク数。
    • 期限切れタスク：期限が過ぎた未完了タスクを特定。
    • 優先順位分布：各優先順位（High、Medium、Low）のタスク数。
  • register_resource：タスク分析をリソースとして登録。

前提条件

  • NotionデータベースにName（title）、Status（select）、Due Date（date）、Priority（select）プロパティが存在。
  • データベースに複数のエントリがあり、ステータス、期限、優先順位が設定済み。
  • .envファイルに正しいNOTION_TOKENとNOTION_DATABASE_IDが設定済み。

サーバーのテスト

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

  1. サーバー起動：

    python notion_analysis_server.py

    コンソールに「Notion分析MCPサーバーを起動中: http://localhost:8119」と表示。

  2. タスク分析のテスト：
    Pythonでリクエストを送信：

    import requests
import json

url = "http://localhost:8119"
payload = {
    "jsonrpc": "2.0",
    "method": "analyze_tasks",
    "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",
    "analysis": {
      "completion_rate": "33.33%",
      "avg_completion_time_hours": "48.00",
      "status_counts": {
        "To Do": 2,
        "In Progress": 1,
        "Done": 1
      },
      "overdue_tasks": [
        {
          "id": "entry123",
          "title": "コードレビュー",
          "due_date": "2025-04-20"
        }
      ],
      "priority_counts": {
        "High": 1,
        "Medium": 2,
        "Low": 1
      }
    }
  },
  "id": 1
}

Claude Desktopとの接続

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

  1. 設定ファイルの編集：
    Claude Desktopの設定ファイル（例：claude_desktop_config.json）に以下を追加：

    {
  "mcp_servers": [
    {
      "name": "NotionAnalysisServer",
      "url": "http://localhost:8119",
      "auth": "none"
    }
  ]
}

  2. Claudeでテスト：
    Claude Desktopを起動し、プロンプトを入力：

    タスクデータベースの進捗を分析してください。

    レスポンス例：

    タスクデータベース分析：
- 完了率：33.33%
- 平均完了時間：48.00時間
- ステータス：To Do（2）、In Progress（1）、Done（1）
- 期限切れタスク：コードレビュー（期限：2025-04-20）
- 優先順位：High（1）、Medium（2）、Low（1）

実装のコツと注意点

  • データ品質：プロパティ（ステータス、期限、優先順位）が欠けている場合、分析結果が制限される。
  • レートリミティング：Notion APIの制限（通常3リクエスト/秒）に注意。
  • セキュリティ：本番環境では、auth: noneを避け、トークン認証を導入。
  • パフォーマンス：大量のエントリを処理する場合、キャッシュ（例：Redis）を検討。
  • テスト：テスト用データベースを作成し、本番データに影響を与えない。

試してみよう：挑戦課題

以下の機能を追加して、エージェントを強化してみてください：

  • 特定ステータス（例：In Progress）のタスクだけを分析するフィルター。
  • 期限切れタスクに自動でコメントを追加するツール。
  • 分析結果をCSVファイルにエクスポートする機能。

まとめと次のステップ

この第3章では、Notionのデータベースを活用してデータ分析エージェントを構築しました。完了率や期限切れタスクを分析することで、AIがプロジェクトのインサイトを提供し、プロセス改善を支援できるようになりました。

次の第4章では、NotionのWebhookを活用してリアルタイム管理AIを構築します。たとえば、AIがデータベースの更新やページ変更をリアルタイムで検知し、通知やコメントを生成します。リアルタイム管理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?