はじめに

第1章では、Notionと**Model Context Protocol（MCP）**の基本を学び、Notionからページデータを取得するMCPサーバーを構築しました。これにより、AIがページのタイトルやメタデータを取得できるようになりました。今回は、この基盤を進化させ、Notionに新しいページを作成したり、既存のページを更新したりする機能を実装します。これにより、タスクを自動化するエージェントを構築します。

この第2章では、MCPサーバーを通じてAIがタスクページを作成したり、ページにプロパティを追加したり、ステータスを更新したりできるようにします。たとえば、ユーザーが「新しいタスクを追加して」と依頼すると、AIがNotionにタスクページを作成し、必要に応じて説明や期限を設定します。コード例とステップごとのガイドで、タスク自動化の可能性を体感しましょう。さあ、始めましょう！

タスク自動化エージェントとは？

タスク自動化エージェントは、Notionのページやデータベースを操作し、ノートやタスク管理を効率化するAIです。MCPサーバーを介して、以下のような機能を実現できます：

ページ作成：新しいタスクやノートを自動で追加。
ページ更新：ステータス、タグ、期限を変更。
タスク整理：優先順位やカテゴリに基づいてページを自動分類。

ユースケース

タスク管理：AIが新しいタスクページを自動で作成し、データベースにリンク。
ノート整理：AIがページにタグやプロパティを追加して整理。
チーム連携：AIがページにコメントを追加し、コラボレーションを促進。

開発環境の準備

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

Python 3.8以降、mcpライブラリ、requestsライブラリ、Claude Desktop：第1章と同じ。
python-dotenv：環境変数の管理（既にインストール済み）。
Notionページ：タスク自動化用のページ。

Notionのセットアップ

ページ確認：
- 第1章で作成したNotionページ（例：MCP-Project）を使用。
- テスト用にサブページやデータベースを追加（例：タスクデータベース）。
- ページまたはデータベースのIDを記録（Notion UIのURLやAPIで取得可能）。
APIトークンの確認：
- 第1章のトークンを再利用（Read Contentに加え、Write ContentとInsert Content権限が必要）。
- インテグレーションが対象ページにアクセス可能であることを確認。
環境変数：
第1章の.envファイル（NOTION_TOKEN、NOTION_PAGE_ID）を再利用。必要に応じてデータベースIDを追加：
```
NOTION_DATABASE_ID=your_database_id
```

コード例：タスク自動化用MCPサーバー

以下のMCPサーバーは、Notionからページデータを取得し、新しいページを作成したり、データベースにエントリを追加したりする機能を提供します。

from mcp import MCPServer
import os
from dotenv import load_dotenv
import requests

class NotionTaskServer(MCPServer):
    def __init__(self, host, port, token, page_id, database_id):
        super().__init__(host, port)
        self.token = token
        self.page_id = page_id
        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("get_page", self.get_page)
        self.register_tool("create_task_page", self.create_task_page)
        self.register_tool("add_database_entry", self.add_database_entry)

    def get_page(self, params):
        try:
            url = f"{self.base_url}/pages/{self.page_id}"
            response = requests.get(url, headers=self.headers)
            response.raise_for_status()
            page = response.json()
            page_info = {
                "id": page["id"],
                "title": page["properties"].get("title", {}).get("title", [{}])[0].get("plain_text", ""),
                "created_time": page["created_time"],
                "last_edited_time": page["last_edited_time"]
            }
            return {"status": "success", "page_info": page_info}
        except Exception as e:
            return {"status": "error", "message": str(e)}

    def create_task_page(self, params):
        try:
            title = params.get("title", "")
            description = params.get("description", "")
            if not title:
                return {"status": "error", "message": "ページタイトルが必要です"}
            
            url = f"{self.base_url}/pages"
            payload = {
                "parent": {"page_id": self.page_id},
                "properties": {
                    "title": {
                        "title": [{"text": {"content": title}}]
                    }
                },
                "children": [
                    {
                        "object": "block",
                        "type": "paragraph",
                        "paragraph": {
                            "rich_text": [{"text": {"content": description}}]
                        }
                    }
                ] if description else []
            }
            response = requests.post(url, headers=self.headers, json=payload)
            response.raise_for_status()
            page = response.json()
            return {"status": "success", "page_id": page["id"]}
        except Exception as e:
            return {"status": "error", "message": str(e)}

    def add_database_entry(self, params):
        try:
            title = params.get("title", "")
            status = params.get("status", "To Do")
            due_date = params.get("due_date", "")
            if not title:
                return {"status": "error", "message": "タスク名が必要です"}
            
            url = f"{self.base_url}/pages"
            payload = {
                "parent": {"database_id": self.database_id},
                "properties": {
                    "Name": {
                        "title": [{"text": {"content": title}}]
                    },
                    "Status": {
                        "select": {"name": status}
                    },
                    "Due Date": {
                        "date": {"start": due_date} if due_date else None
                    }
                }
            }
            response = requests.post(url, headers=self.headers, json=payload)
            response.raise_for_status()
            page = response.json()
            return {"status": "success", "entry_id": page["id"]}
        except Exception as e:
            return {"status": "error", "message": str(e)}

if __name__ == "__main__":
    load_dotenv()
    server = NotionTaskServer(
        host="localhost",
        port=8118,
        token=os.getenv("NOTION_TOKEN"),
        page_id=os.getenv("NOTION_PAGE_ID"),
        database_id=os.getenv("NOTION_DATABASE_ID")
    )
    print("NotionタスクMCPサーバーを起動中: http://localhost:8118")
    server.start()

コードの説明

get_page：第1章から再利用。ページのメタデータを取得。
create_task_page：指定した親ページの下に新しいタスクページを作成。タイトルと説明を指定可能。
add_database_entry：データベースに新しいエントリを追加。タスク名、ステータス、期限を指定。
register_resource/tool：ページ取得をリソース、ページ作成・データベース追加をツールとして登録。

前提条件

Notionページとデータベースが存在。
データベースにName（title）、Status（select）、Due Date（date）プロパティが設定済み。
.envファイルに正しいNOTION_TOKEN、NOTION_PAGE_ID、NOTION_DATABASE_IDが設定済み。
APIトークンに読み書き権限がある。

サーバーのテスト

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

サーバー起動：
```
python notion_task_server.py
```
コンソールに「NotionタスクMCPサーバーを起動中: http://localhost:8118」と表示。

タスクページ作成のテスト：
Pythonでリクエストを送信：

import requests
import json

url = "http://localhost:8118"
payload = {
    "jsonrpc": "2.0",
    "method": "create_task_page",
    "params": {
        "title": "新しいタスク",
        "description": "プロジェクト計画を策定"
    },
    "id": 1
}
response = requests.post(url, json=payload)
print(json.dumps(response.json(), indent=2, ensure_ascii=False))

期待されるレスポンス：

{
  "jsonrpc": "2.0",
  "result": {
    "status": "success",
    "page_id": "page124"
  },
  "id": 1
}

データベースエントリ追加のテスト：

payload = {
    "jsonrpc": "2.0",
    "method": "add_database_entry",
    "params": {
        "title": "コードレビュー",
        "status": "In Progress",
        "due_date": "2025-04-23"
    },
    "id": 2
}
response = requests.post(url, json=payload)
print(json.dumps(response.json(), indent=2, ensure_ascii=False))

期待されるレスポンス：

{
  "jsonrpc": "2.0",
  "result": {
    "status": "success",
    "entry_id": "entry125"
  },
  "id": 2
}

Claude Desktopとの接続

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

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

{
  "mcp_servers": [
    {
      "name": "NotionTaskServer",
      "url": "http://localhost:8118",
      "auth": "none"
    }
  ]
}

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

新しいタスクページ「プロジェクト計画」を作成してください。

レスポンス例：

「プロジェクト計画」ページを作成しました。

別のプロンプト：

データベースに「コードレビュー」タスクを追加してください。ステータスはIn Progress、期限は2025-04-23です。

レスポンス例：

データベースに「コードレビュー」タスクを追加しました。

実装のコツと注意点

エラーハンドリング：無効なページIDやデータベースプロパティを適切に処理。
レートリミティング：Notion APIの制限（通常3リクエスト/秒）に注意。
セキュリティ：本番環境では、auth: noneを避け、トークン認証を導入。
テスト：テスト用ページとデータベースを作成し、本番データに影響を与えない。
拡張性：大量のページを処理する場合、キャッシュ（例：Redis）を検討。

試してみよう：挑戦課題

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

ページにタグ（multi-selectプロパティ）を追加する機能。
データベースエントリのステータスを更新するツール。
作成したページにコメントを追加する機能。

まとめと次のステップ

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

次の第3章では、Notionのデータベースを活用してデータ分析エージェントを構築します。たとえば、AIがタスクの完了率や期限切れを分析し、プロジェクトのインサイトを提供します。データ分析AIに興味がある方は、ぜひお楽しみに！

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