NotionでAIを強化する | 第3章：データベース分析：AIによるインサイト生成

## はじめに

第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`ファイルに以下を確認：
   ```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 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. **サーバー起動**：
   ```bash
   python notion_analysis_server.py
   ```
   コンソールに「Notion分析MCPサーバーを起動中: http://localhost:8119」と表示。

2. **タスク分析のテスト**：
   Pythonでリクエストを送信：
   ```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))
   ```
   期待されるレスポンス：
   ```json
   {
     "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`）に以下を追加：
   ```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に興味がある方は、ぜひお楽しみに！

---

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