0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

PythonからNotion MCPに接続してページの検索・取得を試してみた

0
Posted at

はじめに

Notion MCPにPythonから接続できるか試してみました。やってみたのは、ブラウザでのOAuth認証、使えるツールの確認、キーワード検索、ページ本文の取得です。MCPのPython SDKを使って、一つのスクリプトにまとめています。

Notion MCPはNotionが提供するリモートのMCPサーバーです。接続先は https://mcp.notion.com/mcp で、利用時にはブラウザでNotionアカウントの認証とアクセス許可を行います。

試した環境

スクリプトを書いたときの環境は次のとおりです。

項目 バージョン
Python 3.13.2
mcp 2.2.0
httpx2 2.13.1
pydantic 2.13.5
python -m venv venv

venv/Scripts/python.exe -m pip install "mcp==2.2.0" "httpx2==2.13.1" "pydantic==2.13.5"

# macOS/Linuxの場合
# venv/bin/python -m pip install "mcp==2.2.0" "httpx2==2.13.1" "pydantic==2.13.5"

Notion側では、取得したいページを閲覧できるアカウントを使います。検索対象になるのも、そのアカウントからアクセスできる内容です。

書いたコード

notion_mcp_client.py として、以下のコードを書きました。まず検索結果を表示し、取得したいページのIDを手入力する作りです。

import asyncio
import webbrowser
from urllib.parse import parse_qs, urlparse

import httpx2
from pydantic import AnyUrl

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


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


class InMemoryTokenStorage(TokenStorage):
    def __init__(self):
        self.tokens: OAuthToken | None = None
        self.client_info: OAuthClientInformationFull | None = None

    async def get_tokens(self) -> OAuthToken | None:
        return self.tokens

    async def set_tokens(self, tokens: OAuthToken) -> None:
        self.tokens = tokens

    async def get_client_info(self) -> OAuthClientInformationFull | None:
        return self.client_info

    async def set_client_info(
        self, client_info: OAuthClientInformationFull
    ) -> None:
        self.client_info = client_info


async def handle_redirect(auth_url: str) -> None:
    print("ブラウザでNotion認証を開始します:", auth_url)
    webbrowser.open(auth_url)


async def handle_callback() -> AuthorizationCodeResult:
    print("認証後、アドレスバーのcallback URLをそのまま貼り付けてください。")
    callback_url = input("> ").strip()
    params = parse_qs(urlparse(callback_url).query)

    if "code" not in params:
        raise RuntimeError("Callback URLにcodeがありません。")

    return AuthorizationCodeResult(
        code=params["code"][0],
        state=params.get("state", [None])[0],
        iss=params.get("iss", [None])[0],
    )


def print_tool_result(result) -> None:
    for content in result.content:
        print(content.text if hasattr(content, "text") else content)


async def main() -> None:
    query = input("検索キーワードを入力してください: ").strip()

    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=InMemoryTokenStorage(),
        redirect_handler=handle_redirect,
        callback_handler=handle_callback,
    )

    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()

                print("===== AVAILABLE TOOLS =====")
                tools = await session.list_tools()
                for tool in tools.tools:
                    print(tool.name)

                print("===== NOTION SEARCH =====")
                print(f"query: {query}")
                search_result = await session.call_tool(
                    "notion-search",
                    arguments={"query": query, "query_type": "internal"},
                )
                print_tool_result(search_result)

                print("===== NOTION FETCH =====")
                page_id = input("取得したいPage IDを入力してください: ").strip()
                if not page_id:
                    raise RuntimeError("Page IDが入力されていません。")

                print("===== PAGE CONTENT =====")
                page_result = await session.call_tool(
                    "notion-fetch", arguments={"id": page_id}
                )
                print_tool_result(page_result)


if __name__ == "__main__":
    asyncio.run(main())

実行の流れ

Pythonでスクリプトを実行します。

python notion_mcp_client.py

最初に検索キーワードを入力するようにしました。続いてブラウザでNotionの認証画面が開き、ログインとアクセス許可へ進みます。ブラウザが開かないときのために、認証URLもターミナルへ表示しています。

00dbcd49-228e-4cac-a80f-77b48655ec38-1.png

00dbcd49-228e-4cac-a80f-77b48655ec38-2.png

認証後の戻り先には http://localhost:3000/callback を指定しました。ローカルのWebサーバーは立てていないので、ブラウザには接続エラーが表示される場合があります。そこで、アドレスバーに残った ?code=...&state=... 付きのURL全体をコピーし、ターミナルへ貼り付ける方式にしました。

00dbcd49-228e-4cac-a80f-77b48655ec38-3.png

正常に接続できたら、検索結果が表示されるので、取得したいページのIDを入力して notion-fetch の結果を表示する流れにしました。

取得結果

確認用に作った「テスト」というページを、キーワード「テスト」で検索しました。検索結果にページが1件返り、そのIDを入力すると本文の「Notion MCP の確認用ページ」まで取得できました。

以下は実際の出力から検索・取得に関係する部分を抜粋したものです。

===== NOTION SEARCH =====
query: テスト
{
  "results": [
    {
      "id": "<PAGE_ID>",
      "title": "テスト",
      "type": "page",
      "highlight": "**テスト**"
    }
  ],
  "type": "workspace_search"
}

検索結果に含まれるIDを入力しました。

===== NOTION FETCH =====
取得したいPage IDを入力してください: <PAGE_ID>

===== PAGE CONTENT =====
{
  "metadata": {"type": "page"},
  "title": "テスト",
  "url": "<PAGE_URL>",
  "text": "<page url=\"<PAGE_URL>\">\n<properties>\n{\"title\":\"テスト\"}\n</properties>\n<content>\nNotion MCP の確認用ページ\n<empty-block/>\n</content>\n</page>"
}

notion-search ではページのタイトルとIDが返り、notion-fetch では本文が <content> の中に含まれていました。

実装で確認したこと

OAuthの情報はメモリに保存した

認証URLの生成やトークン取得には OAuthClientProvider を使いました。token_endpoint_auth_method="none" は、クライアントシークレットなしの公開クライアントとして登録するための指定です。

保存先は、試すだけなら扱いやすいメモリ上にしました。InMemoryTokenStorage がトークンとクライアント登録情報を保持します。ただし、スクリプトを終了すると消えるため、次回起動時には再認証が必要です。

HTTP接続の上にMCPセッションを作った

接続まわりは、httpx2.AsyncClient にOAuthプロバイダーを渡し、そのHTTPクライアントを streamable_http_client へ渡す形にしました。そこから ClientSession を作って initialize() すると、Notionのツールを呼び出せます。

検索結果からページを選んで取得した

最初に list_tools() で使えるツールを表示するようにしました。検索には notion-search を使い、query_type="internal" でコンテンツ検索を指定しています。結果を見てページIDを入力し、そのIDを notion-fetch に渡す形です。今回は読み取りだけを試しています。

試した範囲での注意点

  • 今回は検索結果を整形・解析せず、そのまま表示するところまでにしました。ページIDも手動で入力します。
  • Notion MCPのツール構成は変わることがあります。実行時の一覧に notion-search がない場合、このコードの検索部分はそのままでは動きません。公式ドキュメントでは、検索前に notion-get-tool-access で利用可能な検索ツールを確認する手順が案内されています。
  • コールバックURLに含まれる state や iss は、取得できた値をSDKへ渡す形にしました。
  • ページを取得できないときは、入力したIDと、認証したNotionアカウントの閲覧権限を確認する必要があります。
  • 今回は手作業でコールバックURLを貼り付け、トークンもメモリだけに保存しています。継続して使うなら、この部分は作り直したいところです。

まとめ

PythonからNotion MCPへ接続する小さなスクリプトを書き、認証からページ取得までの流れを試しました。OAuthやMCPの通信はSDKに任せられるので、コード上ではツールを呼び出す部分に集中できました。一方、コールバックURLの貼り付けとページIDの選択は手作業です。まず一連の流れを確かめるには、このくらいの作りが分かりやすかったです。

参考資料

0
0
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
0
0

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?