はじめに
前回の記事では、PythonからNotion MCPへ接続し、ページの検索と取得を試しました。ただし、認証後にブラウザのアドレスバーからコールバックURLをコピーする必要があり、スクリプトを起動するたびに認証し直していました。
今回は認証時のコールバックをローカルで受け取り、認証情報を保存します。次回以降はその情報を再利用し、ブラウザでの認証を繰り返さずにNotion MCPへ接続できます。
試した環境
| 項目 | バージョン |
|---|---|
| Python | 3.13.2 |
mcp |
2.2.0 |
httpx2 |
2.13.1 |
pydantic |
2.13.5 |
接続先は https://mcp.notion.com/mcp、リダイレクト先は http://localhost:3000/callback です。
OAuth認証の手作業を減らす
コールバックURLを自動で受け取る
コールバックの受信には asyncio.start_server() を使い、ローカルで一時的に待ち受けます。
認証が必要になると、redirect_handler が先に待ち受けを開始し、その後ブラウザを開きます。Notionからのリダイレクトを受けたら、URLの code、state、iss を AuthorizationCodeResult に入れてSDKへ渡します。state と iss の照合や認証コードとトークンの交換はSDKに任せます。受信後は待ち受けを終了し、一定時間内に戻ってこなければタイムアウトにします。MCP Python SDKのOAuthガイド
ポート3000を別のプログラムが使っていると、待ち受けを開始できません。リダイレクト先のURLと待ち受け先も一致させます。
トークンとクライアント情報を保存する
前回の InMemoryTokenStorage を FileTokenStorage に置き換えました。トークンだけでなく、クライアント登録情報も保存します。登録情報を保存しないと、起動のたびにクライアント登録をやり直すためです。MCP Python SDKのOAuthガイド
Windowsでは認証情報をユーザーごとの AppData\Local\notion-mcp-python\auth.json に保存します。保存時は一時ファイルに書いてから置き換えます。
ただし、認証情報は暗号化せずに保存するため、取り扱いには注意が必要です。
実行して確かめる
初回の認証
仮想環境のPythonで実行し、開いたブラウザでNotionの認証を済ませます。
venv/Scripts/python.exe notion_mcp_client_part2.py
実際の出力は次のとおりです。
ブラウザで Notion の認証を完了してください。
Notion MCP に接続しました。利用可能なツール: 44 件
認証情報の保存先: C:\Users\<USER_NAME>\AppData\Local\notion-mcp-python\auth.json
2回目以降の起動
保存した認証情報が使える場合、SDKはそれを読み込みます。アクセストークンの更新が必要なら更新用トークンを使い、それも使えない場合はブラウザでの認証に戻ります。そのため、通常の再起動ではローカルの待ち受けもブラウザ操作も発生しません。
まとめ
認証時のコールバックをPythonで受け取り、トークンとクライアント情報をファイルへ保存するようにしました。これで、前回必要だったURLの貼り付けと、毎回のブラウザ認証を減らせます。次回はこの認証処理を使い、検索結果からページを選んで本文を取得するところを試します。
コード全文
"""Notion MCP の OAuth コールバック受信と認証情報の再利用を試す。"""
import asyncio
import json
import os
import tempfile
import webbrowser
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.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()
async def main() -> 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 = await session.list_tools()
print(f"Notion MCP に接続しました。利用可能なツール: {len(tools.tools)} 件")
print(f"認証情報の保存先: {storage.path}")
finally:
await callback.close()
if __name__ == "__main__":
asyncio.run(main())