はじめに

第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のセットアップ

データベース準備：
- 第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で取得）。
APIトークンの確認：
- 第2章のトークンを再利用（Read Content権限が必要）。
- インテグレーションがデータベースにアクセス可能であることを確認。

環境変数：
第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が設定済み。

サーバーのテスト

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

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

タスク分析のテスト：
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に接続します：

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

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

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

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