PythonからNotion MCPでCSVデータを取り込み、既存レコードを更新してみる

はじめに

前回は、Notion MCPを使ってデータベースにレコードを1件追加し、プロパティを更新してから条件付きで取得しました。今回はCSVの各行を同じデータベースに反映します。CSVに含まれるIDを使って既存レコードを探し、見つかれば更新し、なければ追加する流れを試します。

更新版のCSVを実行したときに、既存レコードが更新されるかも確認します。検索と書き込みは別の操作なので、同時実行や検索結果の反映待ちまで含めて重複を防ぐ仕組みではありません。今回の対象は、スクリプトを一つずつ実行する検証用データベースです。

CSVとデータベースの対応

第5回で使った「Notion MCP 検証用データベース」に、「外部ID」というテキストのプロパティを追加して使います。CSVの external_id と対応させ、レコードを探すためのキーにします。「名前」は人が読むタイトルとして扱い、同じ名前でも別のレコードを区別できるようにします。

CSVの列	Notionのプロパティ	扱い方
external_id	外部ID	既存レコードの検索に使う
name	名前	ページのタイトルにする
status	状態	データベースにある選択肢を指定する
due_date	期限	日付がある行だけ設定する

初回の取り込みには、次の part6_sample_initial.csv を使います。due_date が空欄の行では、追加時に期限を設定せず、更新時は既存の期限を変えません。

external_id,name,status,due_date
task-001,原稿を確認する,未着手,2026-12-10
task-002,図を作成する,進行中,

更新時に使う part6_sample_updated.csv は次の内容です。

external_id,name,status,due_date
task-001,原稿を確認する,進行中,2026-12-10
task-002,図を作成する,進行中,

初回のCSVから変えたのは、task-001 の status を「未着手」から「進行中」にした箇所だけです。

Notion側の「外部ID」には一意制約がありません。取り込み前にCSV内のIDの重複を調べ、Notion側で同じIDのレコードが複数見つかった場合も処理を止める方針です。

データの取得方法

既存レコードの取得には notion-query-data-sources を使います。このツールにはviewモードとSQLモードがあります。

viewモード

指定したビューに表示されるレコードを取得します。取得した結果から、クライアント側で「外部ID」が一致するレコードを探します。今回のように全レコードを照合する場合は、絞り込みのないビューを使い、結果が複数ページに分かれたら最後まで取得します。

SQLモード

WHERE 句で「外部ID」が一致するレコードだけを取得できます。ただし、Notionの資料では、SQLモードにはワークスペースのプランに応じた利用枠があり、viewモードにはこのツール固有の利用枠がないと説明されています。Notion MCPで利用できるツール

今回はviewモードを使って動作を確認しました。

PythonからCSVの各行を反映する

認証処理は第5回と同じ方式を第6回のスクリプト内に含めています。データベースとデータソースを取得し、「外部ID」「名前」「状態」「期限」のプロパティを確認してからCSVを処理します。CSVはUTF-8で読み込み、必須列、空のIDや名前、日付の形式、状態の選択肢を確認します。

外部IDで既存レコードを探す

対象のビューは絞り込みを外し、「外部ID」プロパティを表示しておきます。スクリプトの DATABASE_ID、DATABASE_TITLE、VIEW_URL は、対象のデータベースとビューに合わせて設定します。

スクリプトはビューに絞り込みがないことを確認し、複数ページに分かれた結果を最後まで読み取ります。その中から、CSVの external_id とNotionの「外部ID」が一致するレコードをPythonで探します。一致するものが0件なら新規作成、1件ならそのページIDを使って更新します。複数件見つかった場合は、どのレコードを更新するか決められないため停止します。

新しいレコードを追加する

該当するレコードがない行は、notion-create-pages でデータソースに追加します。作成時に返ったページIDを記録し、「外部ID」「名前」「状態」と、CSVに日付がある場合の「期限」を設定します。

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

該当するレコードが1件ある行は、notion-update-page の update_properties で値を更新します。今回はCSVに値があるプロパティを更新し、期限の空欄は既存値を残す扱いにします。CSVから期限を削除した場合にNotion側の期限も消す処理は、今回の対象に含めません。

実行して確かめる

CSVのIDと一致するレコードがない状態で、初回のCSVを取り込みました。

python notion_mcp_client_part6.py part6_sample_initial.csv
Notion MCP に接続しました。利用可能なツール: 68 件
データベース: Notion MCP 検証用データベース
データソース: collection://<DATA_SOURCE_ID>
照合するビュー: view://<VIEW_ID>
task-001: 作成 ページID=<PAGE_ID_1>
task-002: 作成 ページID=<PAGE_ID_2>
結果: 作成 2件、更新 0件、変更なし 0件

次に、上の更新版CSVを実行しました。

python notion_mcp_client_part6.py part6_sample_updated.csv
Notion MCP に接続しました。利用可能なツール: 68 件
データベース: Notion MCP 検証用データベース
データソース: collection://<DATA_SOURCE_ID>
照合するビュー: view://<VIEW_ID>
task-001: 更新 ページID=<PAGE_ID_1> 項目=状態
task-002: 変更なし ページID=<PAGE_ID_2>
結果: 作成 0件、更新 1件、変更なし 1件

初回は2件とも作成され、更新版では task-001 の「状態」だけが更新されました。両方の実行で各行のページIDが一致しているため、更新版で新しいページを作らず、初回に作成したページを更新したことが分かります。

試す際の注意点

「外部ID」に一致するレコードを検索してから作成するまでの間に、別の処理が同じIDでレコードを作成すると、重複する可能性があります。また、作成直後に検索結果へ反映されるまで時間がかかる場合、すぐに再実行すると同じことが起こり得ます。

まとめ

CSVの2行をデータベースに追加し、更新版のCSVでは1行だけ既存レコードの「状態」を変更しました。どちらの実行でも同じページIDを使っていることから、作成したレコードを更新できたことを確認しました。

参考資料

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

コード全文

以下は notion_mcp_client_part6.py の全文です。掲載コードの DATABASE_ID と VIEW_URL は伏せています。実行時は、対象のデータベースIDとビューURLを設定し、DATABASE_TITLE も対象の名前に合わせます。

"""CSV の各行を Notion MCP の検証用データベースに作成・更新する。"""

import argparse
import asyncio
import csv
import json
import os
import re
import tempfile
import webbrowser
from dataclasses import dataclass
from datetime import date
from pathlib import Path
from urllib.parse import parse_qs, urlsplit
from uuid import UUID

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


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


COLUMNS = ("external_id", "name", "status", "due_date")
VIEW_URL = "view://<VIEW_ID>"
REQUIRED_TOOLS = (
    "notion-fetch",
    "notion-create-pages",
    "notion-update-page",
    "notion-query-data-sources",
)


@dataclass(frozen=True)
class CsvRecord:
    line: int
    external_id: str
    name: str
    status: str
    due_date: str


def read_csv(path: Path) -> list[CsvRecord]:
    records = []
    seen = set()
    with path.open("r", encoding="utf-8-sig", newline="") as file:
        reader = csv.DictReader(file)
        if reader.fieldnames is None or tuple(reader.fieldnames) != COLUMNS:
            raise ValueError("CSV の列は external_id,name,status,due_date の順に指定してください。")
        for line, row in enumerate(reader, start=2):
            if None in row or any(value is None for value in row.values()):
                raise ValueError(f"{line}行目: 列の数が一致しません。")
            values = {key: value.strip() for key, value in row.items()}
            external_id = values["external_id"]
            if not external_id or not values["name"] or not values["status"]:
                raise ValueError(f"{line}行目: external_id、name、status は必須です。")
            if external_id in seen:
                raise ValueError(f"{line}行目: external_id が重複しています: {external_id}")
            seen.add(external_id)
            due_date = values["due_date"]
            if due_date:
                if not re.fullmatch(r"\d{4}-\d{2}-\d{2}", due_date):
                    raise ValueError(f"{line}行目: due_date は YYYY-MM-DD 形式で指定してください。")
                try:
                    date.fromisoformat(due_date)
                except ValueError as exc:
                    raise ValueError(f"{line}行目: due_date が正しい日付ではありません。") from exc
            records.append(CsvRecord(line, external_id, values["name"], values["status"], due_date))
    if not records:
        raise ValueError("CSV にデータ行がありません。")
    return records


def validate_schema(source_text: str, records: list[CsvRecord]) -> None:
    schema = data_source_schema(source_text)
    if schema.get("外部ID", {}).get("type") != "text":
        raise RuntimeError("テキスト型の「外部ID」プロパティが必要です。")
    options = {item["name"] for item in schema["状態"]["options"]}
    for record in records:
        if record.status not in options:
            raise ValueError(f"{record.line}行目: 状態の選択肢に {record.status} がありません。")


async def query_record(session, tools, view_url: str, external_id: str) -> list[dict]:
    matches = []
    cursor = None
    while True:
        arguments = {"data": {"mode": "view", "view_url": view_url, "page_size": 100}}
        if cursor:
            arguments["data"]["start_cursor"] = cursor
        result = await call_notion(session, tools, "notion-query-data-sources", arguments)
        rows = result.get("results")
        if not isinstance(rows, list) or any(not isinstance(row, dict) for row in rows):
            raise RuntimeError("ビューの取得結果を読み取れません: " + json.dumps(result, ensure_ascii=False))
        for row in rows:
            if "外部ID" not in row:
                raise RuntimeError("ビューの取得結果に「外部ID」がありません。")
            if row["外部ID"] == external_id:
                matches.append(row)
        if not result.get("has_more"):
            return matches
        cursor = result.get("next_cursor")
        if not isinstance(cursor, str) or not cursor:
            raise RuntimeError("ビューの続きの取得位置がありません。")


def changed_properties(record: CsvRecord, current: dict) -> dict:
    properties = {}
    if current.get("名前") != record.name:
        properties["名前"] = record.name
    if current.get("状態") != record.status:
        properties["状態"] = record.status
    if record.due_date and current.get("date:期限:start") != record.due_date:
        properties["date:期限:start"] = record.due_date
        properties["date:期限:is_datetime"] = 0
    return properties


async def wait_for_record(session, tools, view_url: str, record: CsvRecord, page_id: str) -> None:
    for delay in (2, 4, 8, 15):
        await asyncio.sleep(delay)
        rows = await query_record(session, tools, view_url, record.external_id)
        if len(rows) > 1:
            raise RuntimeError(f"外部ID {record.external_id} に複数のレコードがあります。")
        if rows and notion_id(rows[0].get("url", "")) == page_id:
            if not changed_properties(record, rows[0]):
                return
    raise RuntimeError(f"検索結果への反映を確認できません。外部ID: {record.external_id}、ページID: {page_id}")


async def run_import(session: ClientSession, tools: dict, records: list[CsvRecord], dry_run: bool) -> None:
    missing = [name for name in REQUIRED_TOOLS if name not in tools]
    if missing:
        raise RuntimeError("必要なツールが見つかりません: " + ", ".join(missing))
    database = await call_notion(session, tools, "notion-fetch", {"id": DATABASE_ID})
    if database.get("title") != DATABASE_TITLE or database.get("truncated"):
        raise RuntimeError("検証用データベースを確認できません。")
    database_text = database.get("text")
    if not isinstance(database_text, str):
        raise RuntimeError("データベースの取得結果が不正です。")
    source_url = data_source_url(database_text)
    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("データソースを完全に取得できませんでした。")
    validate_schema(source_text, records)
    source_id = source_url.removeprefix("collection://")
    view = await call_notion(session, tools, "notion-fetch", {"id": VIEW_URL})
    view_text = view.get("text")
    if not isinstance(view_text, str) or source_url not in view_text or view.get("truncated"):
        raise RuntimeError("検証用データベースのビューを確認できません。")
    if re.search(r'"(?:filter|filters)"\s*:', view_text):
        raise RuntimeError("ビューに絞り込みがあります。全レコードを表示するビューが必要です。")
    print("データベース:", DATABASE_TITLE)
    print("データソース:", source_url)
    print("照合するビュー:", VIEW_URL)

    # 書き込み前に CSV の全 ID を照合し、Notion 側の重複を検出する。
    existing = {}
    for record in records:
        rows = await query_record(session, tools, VIEW_URL, record.external_id)
        if len(rows) > 1:
            raise RuntimeError(f"外部ID {record.external_id} に複数のレコードがあります。")
        existing[record.external_id] = rows[0] if rows else None

    counts = {"作成": 0, "更新": 0, "変更なし": 0}
    for record in records:
        current = existing[record.external_id]
        if current is None:
            if dry_run:
                print(f"{record.external_id}: 作成予定")
                continue
            properties = {"外部ID": record.external_id, "名前": record.name, "状態": record.status}
            if record.due_date:
                properties["date:期限:start"] = record.due_date
                properties["date:期限:is_datetime"] = 0
            arguments = {"parent": {"data_source_id": source_id}, "pages": [{"properties": properties}]}
            if "allow_async" in tools["notion-create-pages"].input_schema.get("properties", {}):
                arguments["allow_async"] = False
            result = await call_notion(session, tools, "notion-create-pages", arguments)
            pages = result.get("pages")
            if not isinstance(pages, list) or len(pages) != 1 or not isinstance(pages[0], dict):
                raise RuntimeError("作成結果からページを特定できません: " + json.dumps(result, ensure_ascii=False))
            page_id = notion_id(pages[0].get("id") or pages[0].get("url") or "")
            print(f"{record.external_id}: 作成 ページID={page_id}", flush=True)
            counts["作成"] += 1
            await wait_for_record(session, tools, VIEW_URL, record, page_id)
            continue

        page_id = notion_id(current.get("url", ""))
        properties = changed_properties(record, current)
        if not properties:
            print(f"{record.external_id}: 変更なし ページID={page_id}")
            counts["変更なし"] += 1
            continue
        if dry_run:
            print(f"{record.external_id}: 更新予定 ページID={page_id} 項目={','.join(properties)}")
            continue
        await call_notion(session, tools, "notion-update-page", {
            "page_id": page_id,
            "command": "update_properties",
            "properties": properties,
        })
        print(f"{record.external_id}: 更新 ページID={page_id} 項目={','.join(properties)}", flush=True)
        counts["更新"] += 1
        await wait_for_record(session, tools, VIEW_URL, record, page_id)
    if not dry_run:
        print("結果:", "、".join(f"{key} {value}件" for key, value in counts.items()))


async def main(records: list[CsvRecord], dry_run: bool) -> None:
    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=FileTokenStorage(auth_file_path()),
        redirect_handler=callback.redirect,
        callback_handler=callback.callback,
    )
    try:
        async with httpx2.AsyncClient(auth=oauth, timeout=httpx2.Timeout(30.0, read=300.0)) as client:
            async with streamable_http_client(MCP_URL, http_client=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)} 件")
                    await run_import(session, tools, records, dry_run)
    finally:
        await callback.close()


def parse_args():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("csv_file", type=Path, help="取り込む CSV ファイル")
    parser.add_argument("--dry-run", action="store_true", help="照合結果だけ表示し、書き込まない")
    return parser.parse_args()


if __name__ == "__main__":
    try:
        args = parse_args()
        asyncio.run(main(read_csv(args.csv_file), args.dry_run))
    except (OSError, ValueError, RuntimeError) as exc:
        raise SystemExit(str(exc)) from exc
    except (KeyboardInterrupt, EOFError):
        print("\n終了しました。")