PythonからNotion MCPでページを作成・更新してみた

はじめに

前回は、Notion MCPの検索結果からページを選び、本文を取得しました。今回はPythonから新しいページを作成し、そのページのタイトルと本文を更新します。

作成時に返ったページIDを更新処理に渡し、最後に同じページを取得して変更を確認します。認証には第2回で扱った保存とコールバック受信の仕組みを使っています。

試した環境

項目	バージョン
Python	3.13.2
mcp	2.2.0
httpx2	2.13.1
pydantic	2.13.5

WindowsのGit Bashで、仮想環境内のPythonを使いました。接続先は https://mcp.notion.com/mcp です。スクリプトは notion_mcp_client_part4.py として保存しています。

認証情報は %LOCALAPPDATA%\notion-mcp-python\auth.json に保存します。保存済みの情報が使える場合は再利用し、認証情報がない場合はブラウザを開いて新たに認証します。認証結果を受け取るコールバックのURLは http://localhost:3000/callback です。

ページを作成・更新する

今回のスクリプトは、1回の実行でページの新規作成、作成結果の取得、タイトルと本文の更新、更新結果の取得を順に行います。作成時に返ったページIDを使い、その実行で作成したページを更新します。

今回使うツールは次のとおりです。

ツール	用途
notion-create-pages	ページを作成する
notion-update-page	タイトルや本文を更新する
notion-fetch	作成・更新したページを取得する

接続後に session.list_tools() で一覧を取得し、この3つがそろっているか確認しています。ツールの説明と入力項目は、次のコマンドで表示します。このオプションではページを作成しません。

python notion_mcp_client_part4.py --list-tools

また、本文の作成前に session.read_resource() で notion://docs/enhanced-markdown-spec を取得しています。Notionの本文に使うMarkdownの仕様を確認するためのリソースです。

ページを作成する

notion-create-pages の pages に、タイトルと本文を指定します。今回は1件だけ作成し、タイトルを properties の title、本文を content に入れています。

INITIAL_CONTENT = "## 作成時の本文\n\nPythonからNotion MCPで作成したページです。"

title = args.title or "Python MCP 第4回テスト " + datetime.now().strftime("%Y%m%d-%H%M%S")
arguments = {
    "pages": [{"properties": {"title": title}, "content": INITIAL_CONTENT}],
}

親ページを指定しない場合は、ワークスペース直下のプライベートページになります。今回はこの方法で試しました。Notion MCPで利用できるツール

スクリプトでは --parent-page も用意しています。指定した場合は、親ページを取得してから parent にそのIDを設定します。対象はデータベースではない通常のページです。

if args.parent_page:
    arguments["parent"] = {"page_id": args.parent_page}

作成結果は call_notion() を通して辞書に変換し、pages から作成したページを取り出します。

created = await call_notion(session, tools, "notion-create-pages", arguments)
pages = created.get("pages")
if not isinstance(pages, list) or len(pages) != 1 or not isinstance(pages[0], dict):
    raise RuntimeError("作成結果からページを特定できません:\n" + json.dumps(created, ensure_ascii=False))
page = pages[0]
identifier = page_id(page.get("id") or page.get("url") or "")

identifier が以降の取得・更新に使うページIDです。作成直後に notion-fetch で本文を取得し、作成時の文章が含まれているか確認しています。

作成したページを更新する

タイトルと本文は、どちらも notion-update-page で更新します。変更する内容に応じて command を変えています。

タイトルは update_properties で更新し、作成時のタイトルの末尾に「（更新済み）」を付けます。

updated_title = title + "（更新済み）"
await call_notion(session, tools, "notion-update-page", {
    "page_id": identifier,
    "command": "update_properties",
    "properties": {"title": updated_title},
})

本文は replace_content で全体を置き換えます。新しい本文は new_str に指定します。

UPDATED_CONTENT = (
    "## 更新後の本文\n\nPythonからNotion MCPで本文を更新しました。\n\n"
    "- ページの作成\n- タイトルの更新\n- 本文の更新\n- 更新結果の取得"
)

await call_notion(session, tools, "notion-update-page", {
    "page_id": identifier,
    "command": "replace_content",
    "new_str": UPDATED_CONTENT,
})

call_notion() は session.call_tool() を呼び出す共通関数です。ツールが返した構造化データ、またはテキスト中のJSONを辞書として読み取ります。接続先の入力仕様が更新引数を data にまとめる形式なら、その形式に合わせて渡しています。

更新後はもう一度 notion-fetch を呼び、タイトルと本文を表示します。タイトルが更新後の値と一致すること、新しい本文の各行が含まれること、作成時の文章が残っていないことを確認しています。

実行結果

スクリプトを実行し、ページの作成からタイトル・本文の更新まで確認しました。

python notion_mcp_client_part4.py
Notion MCP に接続しました。利用可能なツール: 65 件

===== CREATE PAGE =====
作成先: ワークスペース直下のプライベートページ
タイトル: Python MCP 第4回テスト <RUN_TIME>
## 作成時の本文

PythonからNotion MCPで作成したページです。
作成結果: {
  "id": "<PAGE_ID>",
  "url": "https://app.notion.com/p/<PAGE_ID>?pvs=204",
  "properties": {
    "title": "Python MCP 第4回テスト <RUN_TIME>"
  }
}
ページID: <PAGE_ID>
以降の更新対象は、今回作成したページです。

===== CREATED PAGE =====
タイトル: Python MCP 第4回テスト <RUN_TIME>
URL: https://app.notion.com/p/<PAGE_ID>?pvs=204
Here is the result of "fetch" for the Page with URL https://app.notion.com/p/<PAGE_ID> as of <FETCH_TIME_1>:
<page url="https://app.notion.com/p/<PAGE_ID>">
<ancestor-path></ancestor-path>
<properties>
{"title":"Python MCP 第4回テスト <RUN_TIME>"}
</properties>
<iconMetadata>null</iconMetadata>
<content>
## 作成時の本文
PythonからNotion MCPで作成したページです。
</content>
</page>

===== UPDATE TITLE =====
更新後のタイトル: Python MCP 第4回テスト <RUN_TIME>（更新済み）

===== UPDATE CONTENT =====

===== UPDATED PAGE =====
タイトル: Python MCP 第4回テスト <RUN_TIME>（更新済み）
URL: https://app.notion.com/p/<PAGE_ID>?pvs=204
Here is the result of "fetch" for the Page with URL https://app.notion.com/p/<PAGE_ID> as of <FETCH_TIME_2>:
<page url="https://app.notion.com/p/<PAGE_ID>">
<ancestor-path></ancestor-path>
<properties>
{"title":"Python MCP 第4回テスト <RUN_TIME>（更新済み）"}
</properties>
<iconMetadata>null</iconMetadata>
<content>
## 更新後の本文
PythonからNotion MCPで本文を更新しました。
- ページの作成
- タイトルの更新
- 本文の更新
- 更新結果の取得
</content>
</page>

タイトルと本文の更新を確認しました。
再実行すると、新しいテストページを1件作成します。

作成直後の取得結果には「作成時の本文」と、作成時に指定した文章が返りました。更新後はタイトルの末尾に「（更新済み）」が付き、<content> の中身も「更新後の本文」と箇条書きに変わっています。同じページIDで取得しているため、作成したページに変更が反映されたことを確認できました。

注意点

• replace_content は本文全体を置き換える操作です。今回は、この実行で新しく作ったサンプルページを対象にしています。スクリプトには既存ページのIDを更新対象として指定する引数を設けていません。--parent-page は作成先の指定に使います。

• 再実行すると、別のテストページを1件作成します。途中で更新に失敗しても、作成済みのページは残ります。作成結果のIDとURLを先に表示するのは、その場合にも対象を確認できるようにするためです。

• 接続先が allow_async を受け付ける場合、作成処理では False を指定しています。直後に同じページを取得・更新するためです。更新処理などで非同期タスクが返った場合は、共通関数内で notion-get-async-task に問い合わせて完了を待つ実装にしています。今回の出力には、非同期タスクの状態は表示していません。

まとめ

PythonからNotion MCPでページを作成し、返されたページIDを使ってタイトルと本文を更新しました。タイトルは update_properties、本文全体の置き換えは replace_content で行っています。

実行結果では、作成時と更新後のページをそれぞれ取得し、タイトルと本文の変更を確認しました。前回の検索・取得に加え、今回は新しいページを作り、そのページを更新するところまで試せました。

コード全文

notion_mcp_client_part4.py の全文です。

"""Notion MCP で認証し、ページを作成し、そのページのタイトルと本文を更新する。"""

import argparse
import asyncio
import json
import os
import tempfile
import webbrowser
from datetime import datetime
from uuid import UUID
from pathlib import Path
from urllib.parse import parse_qs, urlsplit

import httpx2
from pydantic import AnyUrl

from mcp import ClientSession
from mcp.client.auth import AuthorizationCodeResult, OAuthClientProvider
from mcp.client.streamable_http import streamable_http_client
from mcp.types import PaginatedRequestParams
from mcp.shared.auth import (
    OAuthClientInformationFull,
    OAuthClientMetadata,
    OAuthToken,
)


MCP_URL = "https://mcp.notion.com/mcp"
REDIRECT_URI = "http://localhost:3000/callback"
CALLBACK_TIMEOUT = 180


def auth_file_path() -> Path:
    # Windows ではユーザーごとの LocalAppData に認証情報を保存する。
    base = os.environ.get("LOCALAPPDATA")
    if base:
        return Path(base) / "notion-mcp-python" / "auth.json"
    return Path.home() / ".config" / "notion-mcp-python" / "auth.json"


class FileTokenStorage:
    """SDK の TokenStorage に必要な4メソッドをファイル保存で実装する。"""

    def __init__(self, path: Path) -> None:
        self.path = path
        if path.exists():
            try:
                self.data = json.loads(path.read_text(encoding="utf-8"))
                if not isinstance(self.data, dict):
                    raise ValueError("JSON の最上位がオブジェクトではありません")
            except (OSError, ValueError) as exc:
                raise RuntimeError(f"認証情報を読み込めません: {path}") from exc
        else:
            self.data = {}

    def _save(self) -> None:
        self.path.parent.mkdir(parents=True, exist_ok=True)
        temporary_path = None
        try:
            with tempfile.NamedTemporaryFile(
                mode="w",
                encoding="utf-8",
                dir=self.path.parent,
                prefix="auth-",
                suffix=".tmp",
                delete=False,
            ) as temporary:
                temporary_path = Path(temporary.name)
                json.dump(self.data, temporary, ensure_ascii=False, indent=2)
                temporary.write("\n")
            if os.name != "nt":
                temporary_path.chmod(0o600)
            os.replace(temporary_path, self.path)
        finally:
            if temporary_path is not None:
                temporary_path.unlink(missing_ok=True)

    async def get_tokens(self) -> OAuthToken | None:
        value = self.data.get("tokens")
        return OAuthToken.model_validate(value) if value is not None else None

    async def set_tokens(self, tokens: OAuthToken) -> None:
        self.data["tokens"] = tokens.model_dump(mode="json")
        self._save()

    async def get_client_info(self) -> OAuthClientInformationFull | None:
        value = self.data.get("client_info")
        return (
            OAuthClientInformationFull.model_validate(value)
            if value is not None
            else None
        )

    async def set_client_info(self, client_info: OAuthClientInformationFull) -> None:
        self.data["client_info"] = client_info.model_dump(mode="json")
        self._save()


class LoopbackCallback:
    """認証が必要になったときだけ localhost でコールバックを受け取る。"""

    def __init__(self) -> None:
        self.server: asyncio.AbstractServer | None = None
        self.result: asyncio.Future[AuthorizationCodeResult] | None = None

    async def redirect(self, auth_url: str) -> None:
        if self.server is not None:
            raise RuntimeError("コールバックの待ち受けが既に開始されています")

        self.result = asyncio.get_running_loop().create_future()
        try:
            self.server = await asyncio.start_server(
                self._handle_request, host="localhost", port=3000
            )
        except OSError as exc:
            raise RuntimeError("localhost:3000 で待ち受けできません") from exc

        print("ブラウザで Notion の認証を完了してください。")
        if not webbrowser.open(auth_url):
            print(f"ブラウザが開かない場合は次のURLを開いてください:\n{auth_url}")

    async def callback(self) -> AuthorizationCodeResult:
        if self.result is None:
            raise RuntimeError("コールバックの待ち受けが開始されていません")
        try:
            return await asyncio.wait_for(self.result, timeout=CALLBACK_TIMEOUT)
        except TimeoutError as exc:
            raise RuntimeError("認証の待ち受けがタイムアウトしました") from exc
        finally:
            await self.close()

    async def close(self) -> None:
        if self.server is not None:
            self.server.close()
            await self.server.wait_closed()
            self.server = None

    async def _handle_request(
        self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter
    ) -> None:
        status = "404 Not Found"
        message = "ページが見つかりません。"
        try:
            request = await asyncio.wait_for(reader.readuntil(b"\r\n\r\n"), timeout=5)
            request_line = request.split(b"\r\n", 1)[0].decode("ascii")
            method, target, _ = request_line.split(" ", 2)
            parsed = urlsplit(target)
            if method == "GET" and parsed.path == "/callback":
                params = parse_qs(parsed.query)
                if "error" in params:
                    status = "400 Bad Request"
                    message = "Notion の認証が完了しませんでした。ターミナルを確認してください。"
                    if self.result is not None and not self.result.done():
                        self.result.set_exception(
                            RuntimeError(f"認証エラー: {params['error'][0]}")
                        )
                elif params.get("code"):
                    status = "200 OK"
                    message = "認証を受け取りました。このタブを閉じてください。"
                    if self.result is not None and not self.result.done():
                        self.result.set_result(
                            AuthorizationCodeResult(
                                code=params["code"][0],
                                state=params.get("state", [None])[0],
                                iss=params.get("iss", [None])[0],
                            )
                        )
                else:
                    status = "400 Bad Request"
                    message = "認証コードが見つかりません。"
        except (UnicodeError, ValueError, asyncio.IncompleteReadError, asyncio.LimitOverrunError, TimeoutError):
            status = "400 Bad Request"
            message = "リクエストを読み取れませんでした。"

        body = message.encode("utf-8")
        response = (
            f"HTTP/1.1 {status}\r\n"
            "Content-Type: text/plain; charset=utf-8\r\n"
            "Cache-Control: no-store\r\n"
            "Connection: close\r\n"
            f"Content-Length: {len(body)}\r\n\r\n"
        ).encode("ascii") + body
        try:
            writer.write(response)
            await writer.drain()
        finally:
            writer.close()
            await writer.wait_closed()


def result_text(result) -> str:
    return "\n".join(
        content.text for content in result.content if content.type == "text"
    )


def result_data(result) -> dict:
    """構造化データ、またはテキストに含まれる JSON を読み取る。"""
    if result.is_error:
        raise RuntimeError(f"Notion のツールがエラーを返しました:\n{result_text(result)}")
    if isinstance(result.structured_content, dict):
        return result.structured_content
    for content in result.content:
        if content.type != "text":
            continue
        try:
            data = json.loads(content.text)
        except ValueError:
            continue
        if isinstance(data, dict):
            return data
    raise RuntimeError(f"結果を JSON として読み取れませんでした:\n{result_text(result)}")


INITIAL_CONTENT = "## 作成時の本文\n\nPythonからNotion MCPで作成したページです。"
UPDATED_CONTENT = (
    "## 更新後の本文\n\nPythonからNotion MCPで本文を更新しました。\n\n"
    "- ページの作成\n- タイトルの更新\n- 本文の更新\n- 更新結果の取得"
)
WRITE_TOOLS = ("notion-create-pages", "notion-update-page", "notion-fetch")


def page_id(value: str) -> str:
    """ページID、またはNotionのページURLをUUIDにそろえる。"""
    value = value.strip()
    parsed = urlsplit(value)
    if parsed.scheme:
        host = parsed.hostname or ""
        if parsed.scheme != "https" or not (
            host in ("notion.so", "notion.site", "notion.com")
            or host.endswith((".notion.so", ".notion.site", ".notion.com"))
        ):
            raise argparse.ArgumentTypeError("NotionのページURLを指定してください。")
        value = parsed.path.rstrip("/").rsplit("/", 1)[-1]
        # 通常のURLはタイトル末尾に32桁のIDが付く。/p/<UUID>にも対応する。
        try:
            return str(UUID(value))
        except ValueError:
            value = value[-32:]
    try:
        return str(UUID(value))
    except ValueError as exc:
        raise argparse.ArgumentTypeError("正しいページIDまたはページURLを指定してください。") from exc


async def call_notion(session: ClientSession, tools: dict, name: str, arguments: dict) -> dict:
    # 接続先のスキーマが data ラッパーを要求する場合は、それに合わせる。
    schema = tools[name].input_schema
    if "data" in schema.get("properties", {}) and "command" in arguments:
        arguments = {"data": arguments}
    data = result_data(await session.call_tool(name, arguments=arguments))
    # 更新は同期指定でも、サーバーの待ち時間を超えると非同期になる場合がある。
    async with asyncio.timeout(300):
        while data.get("object") == "async_task":
            if data.get("status") == "succeeded":
                result = data.get("result")
                if not isinstance(result, dict):
                    raise RuntimeError("非同期処理の結果が辞書ではありません。")
                data = result
                continue
            if data.get("status") in ("failed", "cancelled"):
                raise RuntimeError("非同期処理が失敗しました:\n" + json.dumps(data, ensure_ascii=False))
            if "notion-get-async-task" not in tools or not data.get("id"):
                raise RuntimeError("非同期処理を追跡できません:\n" + json.dumps(data, ensure_ascii=False))
            delay = data.get("poll_after_seconds", 2)
            if not isinstance(delay, (int, float)) or delay < 0:
                raise RuntimeError("非同期処理の待ち時間が不正です。")
            await asyncio.sleep(max(1, delay))
            data = result_data(await session.call_tool(
                "notion-get-async-task", arguments={"task_id": data["id"]}
            ))
    for warning in data.get("warnings", []):
        print("ツールからのお知らせ:", warning)
    return data


async def fetch_page(session: ClientSession, tools: dict, identifier: str, heading: str) -> dict:
    print(f"\n===== {heading} =====")
    fetched = await call_notion(session, tools, "notion-fetch", {"id": identifier})
    text = fetched.get("text")
    if not isinstance(text, str) or fetched.get("truncated"):
        raise RuntimeError("ページ本文を完全に取得できなかったため処理を終了します。")
    print("タイトル:", fetched.get("title", ""))
    print("URL:", fetched.get("url", ""))
    print(text)
    return fetched


async def create_and_update(session: ClientSession, tools: dict, args) -> None:
    missing = [name for name in WRITE_TOOLS if name not in tools]
    if missing:
        raise RuntimeError("必要なツールが見つかりません: " + ", ".join(missing))

    if "notion-get-tool-access" in tools:
        await call_notion(session, tools, "notion-get-tool-access", {})
    # 作成・更新に使うNotion Markdownの仕様を接続先から取得する。
    spec = await session.read_resource("notion://docs/enhanced-markdown-spec")
    if not any(getattr(item, "text", "") for item in spec.contents):
        raise RuntimeError("Notion Markdownの仕様を取得できませんでした。")

    if args.parent_page:
        await fetch_page(session, tools, args.parent_page, "PARENT PAGE")
    title = args.title or "Python MCP 第4回テスト " + datetime.now().strftime("%Y%m%d-%H%M%S")
    updated_title = title + "（更新済み）"
    arguments = {
        "pages": [{"properties": {"title": title}, "content": INITIAL_CONTENT}],
    }
    if args.parent_page:
        arguments["parent"] = {"page_id": args.parent_page}
    # 直後に取得・更新するため同期実行を指定。古い接続先には送らない。
    if "allow_async" in tools["notion-create-pages"].input_schema.get("properties", {}):
        arguments["allow_async"] = False

    print("\n===== CREATE PAGE =====")
    print("作成先:", args.parent_page or "ワークスペース直下のプライベートページ")
    print("タイトル:", title)
    print(INITIAL_CONTENT)
    created = await call_notion(session, tools, "notion-create-pages", arguments)
    pages = created.get("pages")
    if not isinstance(pages, list) or len(pages) != 1 or not isinstance(pages[0], dict):
        raise RuntimeError("作成結果からページを特定できません:\n" + json.dumps(created, ensure_ascii=False))
    page = pages[0]
    # URLやIDが返ったら、後続処理の失敗時にも確認できるよう先に表示する。
    print("作成結果:", json.dumps(page, ensure_ascii=False, indent=2))
    identifier = page_id(page.get("id") or page.get("url") or "")
    print("ページID:", identifier)
    print("以降の更新対象は、今回作成したページです。")
    before = await fetch_page(session, tools, identifier, "CREATED PAGE")
    if "PythonからNotion MCPで作成したページです。" not in before["text"]:
        raise RuntimeError("作成時の本文を確認できませんでした。表示されたページを確認してください。")

    print("\n===== UPDATE TITLE =====")
    await call_notion(session, tools, "notion-update-page", {
        "page_id": identifier,
        "command": "update_properties",
        "properties": {"title": updated_title},
    })
    print("更新後のタイトル:", updated_title)

    print("\n===== UPDATE CONTENT =====")
    # 本文全体の置換は、この実行で作ったサンプルページだけに行う。
    await call_notion(session, tools, "notion-update-page", {
        "page_id": identifier,
        "command": "replace_content",
        "new_str": UPDATED_CONTENT,
    })
    after = await fetch_page(session, tools, identifier, "UPDATED PAGE")
    if after.get("title") != updated_title or any(
        line not in after["text"] for line in UPDATED_CONTENT.splitlines() if line
    ) or "PythonからNotion MCPで作成したページです。" in after["text"]:
        raise RuntimeError("更新結果が予定した内容と一致しません。表示されたページを確認してください。")
    print("\nタイトルと本文の更新を確認しました。")
    print("再実行すると、新しいテストページを1件作成します。")


async def main(args) -> None:
    storage = FileTokenStorage(auth_file_path())
    callback = LoopbackCallback()
    oauth = OAuthClientProvider(
        server_url=MCP_URL,
        client_metadata=OAuthClientMetadata(
            client_name="Python Notion MCP Client",
            redirect_uris=[AnyUrl(REDIRECT_URI)],
            grant_types=["authorization_code", "refresh_token"],
            response_types=["code"],
            token_endpoint_auth_method="none",
        ),
        storage=storage,
        redirect_handler=callback.redirect,
        callback_handler=callback.callback,
    )
    try:
        async with httpx2.AsyncClient(
            auth=oauth, timeout=httpx2.Timeout(30.0, read=300.0)
        ) as http_client:
            async with streamable_http_client(MCP_URL, http_client=http_client) as (read, write):
                async with ClientSession(read, write) as session:
                    await session.initialize()
                    tools = {}
                    cursor = None
                    while True:
                        listed = await session.list_tools(
                            params=PaginatedRequestParams(cursor=cursor) if cursor else None
                        )
                        tools.update((tool.name, tool) for tool in listed.tools)
                        cursor = listed.next_cursor
                        if not cursor:
                            break
                    print(f"Notion MCP に接続しました。利用可能なツール: {len(tools)} 件")
                    if args.list_tools:
                        for name in (*WRITE_TOOLS, "notion-get-tool-access", "notion-get-async-task"):
                            if name in tools:
                                tool = tools[name]
                                print(f"\n===== {name} =====")
                                print(tool.description)
                                print(json.dumps(tool.input_schema, ensure_ascii=False, indent=2))
                        return
                    await create_and_update(session, tools, args)
    finally:
        await callback.close()


def parse_args():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--parent-page", type=page_id, help="作成先の通常ページのIDまたはURL（省略時はプライベートページ）")
    parser.add_argument("--title", help="作成するページのタイトル（省略時は日時付きのテスト名）")
    parser.add_argument("--list-tools", action="store_true", help="作成・更新・取得ツールの仕様を表示して終了（ページは作成しない）")
    args = parser.parse_args()
    if args.title is not None:
        args.title = args.title.strip()
        if not args.title:
            parser.error("--title に空文字は指定できません。")
    return args


if __name__ == "__main__":
    try:
        asyncio.run(main(parse_args()))
    except (KeyboardInterrupt, EOFError):
        print("\n終了しました。")

参考資料

• Notion MCPで利用できるツール
• MCP Python SDK