PythonからNotion MCPでデータベースのレコードを追加・更新・検索してみた

はじめに

前回はPythonからNotion MCPを使い、通常ページを作成してタイトルと本文を更新しました。今回は既存のデータベースにレコードを追加し、そのプロパティを更新してから、条件を指定して取得します。認証処理は前回と同じものをスクリプト内に含めています。

通常ページとデータベース内のページ

Notionのデータベースは、共通のプロパティを持つページをまとめて扱います。表の1行に見えるレコードもNotionページです。

比較する点	通常ページ	データベース内のページ
主に扱う内容	本文の文章やブロック	本文に加え、データベースで定義されたプロパティ
一覧や絞り込み	ページを探して取得する	プロパティを条件にレコードを取得する

Notion MCPでデータベースを取得すると、含まれるデータソースのURLが分かります。データソースを取得するとプロパティ構成を確認でき、レコードの条件付き取得にもそのURLを使います。Notion MCPで利用できるツール

Pythonからデータベースを操作する

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

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

認証情報は %LOCALAPPDATA%\notion-mcp-python\auth.json に保存します。保存済みの情報があれば再利用し、ない場合はブラウザで認証して http://localhost:3000/callback で結果を受け取ります。今回掲載する実行結果では、保存済みの認証情報を使いました。

対象は「Notion MCP 検証用データベース」です。あらかじめ次のプロパティを用意しました。

プロパティ	型	今回の使い方
名前	タイトル	追加したレコードを識別する
状態	セレクト	「未着手」で追加し、「進行中」に更新する
期限	日付	構成を確認する。今回の実行では値を指定しない

スクリプトは session.list_tools() でツール一覧を取得し、検索・取得・追加・更新・条件付き取得に必要なツールがあることを確認します。--list-tools を付けると、使うツールの説明と入力項目を表示して終了します。この場合はレコードを作成しません。

python notion_mcp_client_part5.py --list-tools

データベースを探してプロパティを確認する

notion-search でデータベース名を検索し、検索結果に指定したデータベースIDが含まれることを確認します。同名の別データベースを選ばないためです。続いて notion-fetch でデータベースとデータソースを取得します。

found = await call_notion(session, tools, search_name, {
    "query": DATABASE_TITLE,
    "query_type": "internal",
})
results = found.get("results")
if not isinstance(results, list) or not any(is_target_database(item) for item in results):
    raise RuntimeError("検索結果に検証用データベースが見つかりません。")

database = await call_notion(session, tools, "notion-fetch", {"id": DATABASE_ID})
database_text = database.get("text")
if not isinstance(database_text, str) or database.get("truncated"):
    raise RuntimeError("データベースを完全に取得できませんでした。")
source_url = data_source_url(database_text)
source_id = source_url.removeprefix("collection://")
source = await call_notion(session, tools, "notion-fetch", {"id": source_url})
source_text = source.get("text")
if not isinstance(source_text, str) or source.get("truncated"):
    raise RuntimeError("データソースを完全に取得できませんでした。")
schema = data_source_schema(source_text)

データベースの取得結果には collection://... 形式のデータソースURLが含まれます。今回のデータベースにはデータソースが1つあり、スクリプトはそのスキーマから「名前」「状態」「期限」の型と、「状態」の選択肢を確認してから書き込みます。

レコードを追加する

notion-create-pages の parent にデータソースIDを指定します。今回は「名前」に実行時刻を含む値を、「状態」に「未着手」を入れて1件作成しました。--due-date を指定した場合だけ「期限」にも値を入れます。

title = args.title or "Python MCP 検証 " + datetime.now().strftime("%Y%m%d-%H%M%S")
properties = {"名前": title, "状態": "未着手"}
if args.due_date:
    properties["date:期限:start"] = args.due_date
    properties["date:期限:is_datetime"] = 0
create_args = {"parent": {"data_source_id": source_id}, "pages": [{"properties": properties}]}

created = await call_notion(session, tools, "notion-create-pages", create_args)
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 = notion_id(page.get("id") or page.get("url") or "")

作成時に返ったページIDを、その後の更新と取得結果の確認に使います。直後に更新するため、接続先が対応していれば allow_async=False で作成結果を待ちます。スクリプトはIDを更新前に表示するため、途中で処理が止まっても作成済みのレコードを特定できます。

レコードのプロパティを更新する

notion-update-page の update_properties で、作成したレコードの「状態」を「進行中」に変えます。第4回のタイトル更新と同じツールですが、今回はデータベースで定義したプロパティを更新しています。

await call_notion(session, tools, "notion-update-page", {
    "page_id": identifier,
    "command": "update_properties",
    "properties": {"状態": "進行中"},
})

条件を指定してレコードを取得する

notion-query-data-sources にSQLを渡し、「状態」が「進行中」で「名前」が今回作成した値と一致するレコードを取得します。値はSQL文に埋め込まず、? と params で渡しています。

query = f'SELECT url, "名前", "状態", "date:期限:start" FROM "{source_url}" WHERE "状態" = ? AND "名前" = ?'
queried = await call_notion(session, tools, "notion-query-data-sources", {
    "data": {
        "mode": "sql",
        "data_source_urls": [source_url],
        "query": query,
        "params": ["進行中", title],
    }
})

取得結果のURLからページIDを取り出し、作成時に返ったIDと一致するか確認します。更新直後は条件付き取得に反映されるまで時間がかかる場合があるため、スクリプトは間隔を空けて最大3回取得します。作成後に処理が止まった場合は、表示されたページIDを --record-id に渡すと、新しいレコードを作らずに同じレコードの検証を再開できます。

SQLでのデータソース取得には、ワークスペースのプランによる利用枠があります。今回の接続では実行できました。Notion MCPで利用できるツール

実行結果

スクリプトを実行し、レコードの追加、状態の更新、条件付き取得まで確認しました。

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

===== SEARCH DATABASE =====
データベース: Notion MCP 検証用データベース
ID: <DATABASE_ID>
データソース: collection://<DATA_SOURCE_ID>
プロパティ: 名前 (title), 期限 (date), 状態 (select)

===== CREATE RECORD =====
作成結果: {
  "id": "<PAGE_ID>",
  "url": "https://app.notion.com/p/<PAGE_ID>?pvs=204",
  "properties": {
    "名前": "Python MCP 検証 20261025-084649",
    "状態": "未着手"
  }
}
ページID: <PAGE_ID>

===== UPDATE RECORD =====
状態を「進行中」に更新しました。

===== QUERY RECORD =====
取得結果 (1回目): [
  {
    "url": "https://app.notion.com/<PAGE_ID>",
    "名前": "Python MCP 検証 20261025-084649",
    "状態": "進行中",
    "date:期限:start": null
  }
]
レコードの更新と条件付き取得を確認しました。
再実行すると、新しい検証用レコードを1件作成します。

作成結果では状態が「未着手」で、条件付き取得の結果では同じレコードの状態が「進行中」になっています。取得結果のURLには、作成時に返されたページIDが含まれています。期限は今回指定していないため、date:期限:start は null です。Notionの画面でも、データベースにレコードが追加されたことを確認しました。

まとめ

通常ページと同じ notion-create-pages と notion-update-page を使い、データベース内のページを追加・更新しました。データベースではデータソースのプロパティ構成を確認し、そのプロパティの値を条件にレコードを取得できます。実行結果では、作成時に「未着手」だったレコードが「進行中」になり、同じページIDのレコードを条件付き取得で確認できました。

コード全文

notion_mcp_client_part5.py の全文です。掲載コードの DATABASE_ID は <DATABASE_ID> としています。実行時は対象データベースのIDを設定し、DATABASE_TITLE も対象の名前に合わせています。

"""Notion MCP で認証し、検証用データベースのレコードを追加・更新・検索する。"""

import argparse
import asyncio
import json
import os
import re
import tempfile
import webbrowser
from datetime import date, 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
DATABASE_ID = "<DATABASE_ID>"
DATABASE_TITLE = "Notion MCP 検証用データベース"


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)}")


REQUIRED_TOOLS = (
    "notion-fetch",
    "notion-create-pages",
    "notion-update-page",
    "notion-query-data-sources",
)


def notion_id(value: str) -> str:
    """NotionのIDまたは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 ValueError("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 ValueError("正しいNotion 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


def data_source_url(database_text: str) -> str:
    """データベース取得結果に含まれる最初のデータソースURLを取り出す。"""
    match = re.search(r'<data-source url="\{?\{?(collection://[0-9a-f-]+)\}?\}?"', database_text)
    if match is None:
        raise RuntimeError("データベースの取得結果からデータソースを特定できません。")
    return match.group(1)


def data_source_schema(source_text: str) -> dict:
    """データソースのスキーマから今回使うプロパティを確認する。"""
    match = re.search(r"<data-source-state>\s*(\{.*?\})\s*</data-source-state>", source_text, re.S)
    if match is None:
        raise RuntimeError("データソースのプロパティ構成を読み取れません。")
    state = json.loads(match.group(1))
    schema = state.get("schema")
    if not isinstance(schema, dict):
        raise RuntimeError("データソースのスキーマが不正です。")
    expected = {"名前": "title", "状態": "select", "期限": "date"}
    for name, property_type in expected.items():
        prop = schema.get(name)
        if not isinstance(prop, dict) or prop.get("type") != property_type:
            raise RuntimeError(f"プロパティ {name} ({property_type}) が見つかりません。")
    options = schema["状態"].get("options", [])
    names = {option.get("name") for option in options if isinstance(option, dict)}
    if not {"未着手", "進行中"}.issubset(names):
        raise RuntimeError("状態プロパティに未着手・進行中の選択肢がありません。")
    return schema


def is_target_database(item: object) -> bool:
    if not isinstance(item, dict):
        return False
    identifier = item.get("id") or item.get("url")
    if not isinstance(identifier, str):
        return False
    try:
        return notion_id(identifier) == DATABASE_ID
    except ValueError:
        return False


async def run_database_demo(session: ClientSession, tools: dict, args) -> None:
    search_name = next(
        (name for name in ("notion-search", "notion-ai-search") if name in tools),
        None,
    )
    missing = [name for name in REQUIRED_TOOLS if name not in tools]
    if search_name is None:
        missing.insert(0, "notion-search / notion-ai-search")
    if missing:
        raise RuntimeError("必要なツールが見つかりません: " + ", ".join(missing))

    if "notion-get-tool-access" in tools:
        access = await call_notion(session, tools, "notion-get-tool-access", {})
        query_access = access.get("current_tool_access", {}).get("query_data_sources", {})
        if query_access.get("status") in ("plan_required", "full_version_required", "not_enabled"):
            raise RuntimeError("この接続ではデータソースの条件付き取得を利用できません。")

    print("\n===== SEARCH DATABASE =====")
    found = await call_notion(session, tools, search_name, {
        "query": DATABASE_TITLE,
        "query_type": "internal",
    })
    results = found.get("results")
    if not isinstance(results, list) or not any(is_target_database(item) for item in results):
        raise RuntimeError("検索結果に検証用データベースが見つかりません。")
    print("データベース:", DATABASE_TITLE)
    print("ID:", DATABASE_ID)

    database = await call_notion(session, tools, "notion-fetch", {"id": DATABASE_ID})
    database_text = database.get("text")
    if not isinstance(database_text, str) or database.get("truncated"):
        raise RuntimeError("データベースを完全に取得できませんでした。")
    source_url = data_source_url(database_text)
    source_id = source_url.removeprefix("collection://")
    source = await call_notion(session, tools, "notion-fetch", {"id": source_url})
    source_text = source.get("text")
    if not isinstance(source_text, str) or source.get("truncated"):
        raise RuntimeError("データソースを完全に取得できませんでした。")
    schema = data_source_schema(source_text)
    print("データソース:", source_url)
    print("プロパティ:", ", ".join(f"{name} ({value['type']})" for name, value in schema.items()))

    if args.record_id:
        identifier = args.record_id
        fetched = await call_notion(session, tools, "notion-fetch", {"id": identifier})
        fetched_text = fetched.get("text")
        if not isinstance(fetched_text, str) or fetched.get("truncated"):
            raise RuntimeError("既存レコードを完全に取得できませんでした。")
        if f'<parent-data-source url="{source_url}"' not in fetched_text:
            raise RuntimeError("指定したレコードは検証用データベースに属していません。")
        title = fetched.get("title")
        if not isinstance(title, str) or not title:
            raise RuntimeError("既存レコードの名前を取得できませんでした。")
        match = re.search(r"<properties>\s*(\{.*?\})\s*</properties>", fetched_text, re.S)
        current_properties = json.loads(match.group(1)) if match else {}
        current_status = current_properties.get("状態")
        print("\n===== EXISTING RECORD =====")
        print("名前:", title)
        print("ページID:", identifier)
    else:
        title = args.title or "Python MCP 検証 " + datetime.now().strftime("%Y%m%d-%H%M%S")
        properties = {"名前": title, "状態": "未着手"}
        if args.due_date:
            properties["date:期限:start"] = args.due_date
            properties["date:期限:is_datetime"] = 0
        create_args = {"parent": {"data_source_id": source_id}, "pages": [{"properties": properties}]}
        # 後続の更新で作成結果が必要なため同期実行する。
        if "allow_async" in tools["notion-create-pages"].input_schema.get("properties", {}):
            create_args["allow_async"] = False

        print("\n===== CREATE RECORD =====")
        created = await call_notion(session, tools, "notion-create-pages", create_args)
        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 = notion_id(page.get("id") or page.get("url") or "")
        current_status = "未着手"
        print("作成結果:", json.dumps(page, ensure_ascii=False, indent=2))
        print("ページID:", identifier)

    print("\n===== UPDATE RECORD =====")
    if current_status == "進行中":
        print("状態は既に「進行中」です。")
    else:
        await call_notion(session, tools, "notion-update-page", {
            "page_id": identifier,
            "command": "update_properties",
            "properties": {"状態": "進行中"},
        })
        print("状態を「進行中」に更新しました。")

    print("\n===== QUERY RECORD =====")
    query = f'SELECT url, "名前", "状態", "date:期限:start" FROM "{source_url}" WHERE "状態" = ? AND "名前" = ?'
    delays = (0, 3, 8) if current_status == "進行中" else (3, 8, 15)
    for attempt, delay in enumerate(delays, start=1):
        # 更新直後はデータソースの検索結果へ反映されるまで時間がかかる場合がある。
        await asyncio.sleep(delay)
        queried = await call_notion(session, tools, "notion-query-data-sources", {
            "data": {
                "mode": "sql",
                "data_source_urls": [source_url],
                "query": query,
                "params": ["進行中", title],
            }
        })
        rows = queried.get("results")
        if not isinstance(rows, list):
            raise RuntimeError("取得結果にレコードの一覧がありません:\n" + json.dumps(queried, ensure_ascii=False))
        print(f"取得結果 ({attempt}回目):", json.dumps(rows, ensure_ascii=False, indent=2))
        if any(
            isinstance(row, dict)
            and isinstance(row.get("url"), str)
            and notion_id(row["url"]) == identifier
            and row.get("状態") == "進行中"
            for row in rows
        ):
            print("レコードの更新と条件付き取得を確認しました。")
            if not args.record_id:
                print("再実行すると、新しい検証用レコードを1件作成します。")
            return
        if attempt < len(delays):
            print("検索結果への反映を待って再取得します。")
    raise RuntimeError("条件付き取得でレコードを確認できませんでした。作成したページIDを確認してください: " + identifier)


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 ("notion-search", "notion-ai-search", *REQUIRED_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 run_database_demo(session, tools, args)
    finally:
        await callback.close()


def parse_args():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--title", help="作成するレコードの名前（省略時は日時付きの検証名）")
    parser.add_argument("--due-date", help="期限（YYYY-MM-DD、省略時は設定しない）")
    parser.add_argument("--record-id", help="作成済みのレコードID。指定すると新規作成せず検証を再開する")
    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 に空文字は指定できません。")
    if args.due_date:
        try:
            if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", args.due_date):
                raise ValueError
            date.fromisoformat(args.due_date)
        except ValueError:
            parser.error("--due-date は YYYY-MM-DD 形式で指定してください。")
    if args.record_id:
        try:
            args.record_id = notion_id(args.record_id)
        except ValueError as exc:
            parser.error(str(exc))
        if args.title or args.due_date:
            parser.error("--record-id は --title、--due-date と同時に指定できません。")
    return args


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

参考資料

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