はじめに
最初の記事では、検索結果からページIDをコピーして入力し、本文を取得しました。今回は検索結果を番号付きで表示し、番号を選ぶだけで本文を取得できるようにします。認証処理は前回の記事を参考にしています。
試した環境
| 項目 | バージョン |
|---|---|
| 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 です。
認証情報は AppData\Local\notion-mcp-python\auth.json に保存します。保存済みの情報が使える場合は再利用し、認証が必要な場合はブラウザを開きます。保存とコールバック受信の仕組みは、前回の記事で説明しています。
検索から本文取得までをつなぐ
利用できる検索ツールを確認する
session.list_tools() でツール一覧を取得し、検索用の notion-search と本文取得用の notion-fetch を確認します。notion-search がない接続では、notion-ai-search を探します。
ツールの説明と入力項目を表示する --list-tools オプションも用意しました。
venv/Scripts/python.exe notion_mcp_client_part3.py --list-tools
一覧に notion-get-tool-access があれば、検索前に呼び出します。一部の接続で事前確認が求められるためです。Notion MCPで利用できるツール
キーワードで検索し、結果を読み取る
検索語を query に渡し、コンテンツ検索を指定する query_type="internal" で呼び出します。
search_result = await session.call_tool(
search_name, arguments={"query": query, "query_type": "internal"}
)
data = result_data(search_result)
result_data() は、structured_content の辞書を使うか、テキストのJSONを json.loads() で読み取ります。
検索結果からページを選ぶ
results からIDを持つページを取り出し、番号付きで表示します。
pages = [
item for item in data["results"]
if isinstance(item, dict) and item.get("type") == "page" and item.get("id")
]
for number, page in enumerate(pages, start=1):
print(f"{number}. {page.get('title') or '（タイトルなし）'}")
print(f" ID: {page['id']}")
if page.get("url"):
print(f" URL: {page['url']}")
選択した番号をリストの位置に対応させ、1 なら pages[0] を取得対象にします。不正な番号は再入力、空欄は終了としました。
入力待ちには asyncio.to_thread() を使い、MCPの通信を扱うイベントループを止めないようにしています。Pythonのasyncio.to_thread
選んだページの本文を取得する
選んだページのIDを notion-fetch に渡します。
fetched = result_data(
await session.call_tool("notion-fetch", arguments={"id": page["id"]})
)
取得結果の text をそのまま表示します。本文のほか、<page> や <properties> などのメタ情報も含まれます。truncated が真の場合は一部省略の表示を出しますが、省略部分の追加取得は行っていません。
実行して確かめる
venv/Scripts/python.exe notion_mcp_client_part3.py
Notion MCP に接続しました。利用可能なツール: 49 件
検索キーワードを入力してください: テスト
===== NOTION SEARCH (notion-search) =====
query: テスト
1. テストページ
ID: <PAGE_ID>
URL: https://app.notion.com/p/<PAGE_ID>?pvs=204
取得するページの番号（空欄で終了）: 1
===== NOTION FETCH: テストページ =====
タイトル: テストページ
URL: https://app.notion.com/p/<PAGE_ID>?pvs=204
===== PAGE CONTENT =====
Here is the result of "fetch" for the Page with URL https://app.notion.com/p/<PAGE_ID>?pvs=204 as of <FETCH_TIME>:
<page url="https://app.notion.com/p/<PAGE_ID>?pvs=204">
<ancestor-path></ancestor-path>
<properties>
{"title":"テストページ"}
</properties>
<iconMetadata>null</iconMetadata>
<content>
これはPythonからの接続を確認するテストページ
<empty-block/>
</content>
</page>
<content> の中に、確認用ページに書いた本文が返りました。
検索語と番号を引数で指定することもできます。番号はその実行で返った候補に対応するため、同じページを取得するとは限りません。
venv/Scripts/python.exe notion_mcp_client_part3.py --query テスト --select 1
まとめ
PythonからNotion MCPでキーワード検索を行い、検索結果の中から取得するページを指定できるようにしました。候補を番号付きで表示し、選択したページのIDを notion-fetch に渡すことで、検索と本文取得をつないでいます。
実行結果では、「テスト」で検索して「テストページ」を選び、ページに書いた本文が返ることを確認しました。検索結果に含まれるタイトルやIDと、ページを取得して得られる本文を、それぞれの処理で扱う流れを確かめられました。
認証には前回の記事で扱った保存とコールバック受信の方式を使っています。今回は読み取りの範囲ですが、検索で候補を探し、その中から対象を指定して内容を読む、という操作をPythonで実装できました。
コード全文
notion_mcp_client_part3.py の全文です。
"""Notion MCP で認証し、検索結果から番号で選んだページを取得する。"""
import argparse
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.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)}")
async def prompt(message: str) -> str:
# MCP の通信を動かしたまま、ターミナルからの入力を待つ。
return (await asyncio.to_thread(input, message)).strip()
async def select_page(pages: list[dict], selection: int | None) -> dict | None:
if selection is not None:
if not 1 <= selection <= len(pages):
raise RuntimeError(f"--select は 1〜{len(pages)} の範囲で指定してください。")
return pages[selection - 1]
while True:
value = await prompt("取得するページの番号（空欄で終了）: ")
if not value:
return None
if value.isdecimal() and 1 <= int(value) <= len(pages):
return pages[int(value) - 1]
print(f"1〜{len(pages)} の番号を入力してください。")
async def search_and_fetch(session: ClientSession, tools: dict, args) -> None:
search_name = next(
(name for name in ("notion-search", "notion-ai-search") if name in tools),
None,
)
if search_name is None or "notion-fetch" not in tools:
raise RuntimeError("検索ツールまたは notion-fetch が見つかりません。")
# 接続先によっては、検索前にアクセス情報の確認を求める。
if "notion-get-tool-access" in tools:
access = await session.call_tool("notion-get-tool-access", arguments={})
result_data(access)
query = args.query
if query is None:
query = await prompt("検索キーワードを入力してください: ")
if not query.strip():
print("検索キーワードが空欄のため終了します。")
return
print(f"\n===== NOTION SEARCH ({search_name}) =====")
print(f"query: {query}")
search_result = await session.call_tool(
search_name, arguments={"query": query, "query_type": "internal"}
)
data = result_data(search_result)
if not isinstance(data.get("results"), list):
raise RuntimeError(
"検索結果に results の配列がありません:\n"
+ json.dumps(data, ensure_ascii=False, indent=2)
)
pages = [
item for item in data["results"]
if isinstance(item, dict) and item.get("type") == "page" and item.get("id")
]
for notice in data.get("notices", []):
print("検索時のお知らせ:", notice)
if not pages:
print("取得できる Notion ページが見つかりませんでした。")
return
print(f"\n===== SEARCH RESULTS: {len(pages)} 件 =====")
for number, page in enumerate(pages, start=1):
print(f"{number}. {page.get('title') or '（タイトルなし）'}")
print(f" ID: {page['id']}")
if page.get("url"):
print(f" URL: {page['url']}")
page = await select_page(pages, args.select)
if page is None:
print("終了します。")
return
print(f"\n===== NOTION FETCH: {page.get('title') or page['id']} =====")
fetched = result_data(
await session.call_tool("notion-fetch", arguments={"id": page["id"]})
)
print("タイトル:", fetched.get("title", page.get("title", "")))
if fetched.get("url"):
print("URL:", fetched["url"])
print("\n===== PAGE CONTENT =====")
text = fetched.get("text")
if not isinstance(text, str):
raise RuntimeError(
"取得結果に text がありません:\n"
+ json.dumps(fetched, ensure_ascii=False, indent=2)
)
print(text)
if fetched.get("truncated"):
print("\n本文の一部が省略されています（truncated=true）。")
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",
"notion-get-tool-access", "notion-fetch",
):
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 search_and_fetch(session, tools, args)
finally:
await callback.close()
def parse_args():
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--query", help="検索キーワード（省略時は対話入力）")
parser.add_argument("--select", type=int, help="取得する候補の番号（省略時は対話入力）")
parser.add_argument("--list-tools", action="store_true", help="検索・取得ツールの仕様を表示して終了")
return parser.parse_args()
if __name__ == "__main__":
try:
asyncio.run(main(parse_args()))
except (KeyboardInterrupt, EOFError):
print("\n終了しました。")