1
2

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

NotionでAIを強化する | 第2章:タスクを自動化:AIによるページ操作

Posted at

はじめに

第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. ページ確認
    • 第1章で作成したNotionページ(例:MCP-Project)を使用。
    • テスト用にサブページやデータベースを追加(例:タスクデータベース)。
    • ページまたはデータベースのIDを記録(Notion UIのURLやAPIで取得可能)。
  2. APIトークンの確認
    • 第1章のトークンを再利用(Read Contentに加え、Write ContentInsert Content権限が必要)。
    • インテグレーションが対象ページにアクセス可能であることを確認。
  3. 環境変数
    第1章の.envファイル(NOTION_TOKENNOTION_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_TOKENNOTION_PAGE_IDNOTION_DATABASE_IDが設定済み。
  • APIトークンに読み書き権限がある。

サーバーのテスト

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

  1. サーバー起動

    python notion_task_server.py
    

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

  2. タスクページ作成のテスト
    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
    }
    
  3. データベースエントリ追加のテスト

    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に接続します:

  1. 設定ファイルの編集
    Claude Desktopの設定ファイル(例:claude_desktop_config.json)に以下を追加:

    {
      "mcp_servers": [
        {
          "name": "NotionTaskServer",
          "url": "http://localhost:8118",
          "auth": "none"
        }
      ]
    }
    
  2. 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に興味がある方は、ぜひお楽しみに!


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

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

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?