diff --git a/scripts/notion-mcp.py b/scripts/notion-mcp.py new file mode 100644 index 0000000..b222e53 --- /dev/null +++ b/scripts/notion-mcp.py @@ -0,0 +1,83 @@ +#!/usr/bin/env python3 +""" +notion-mcp-connect · 冰朔/铸渊/之之 通用的 Notion 连接工具 +铸渊 ICE-GL-ZY001 + Mavis · D166 · 2026-07-06 + +功能: + search 搜索页面/数据库 + get 获取页面详情 + blocks 列出页面的 blocks + create 创建子页面 + append <page_id> <text> 追加 paragraph block + check-todo <block_id> 勾选/取消 to_do block + check 检查 token 有效性 + query <db_id> 查询数据库 + +环境: NOTION_TOKEN=ntn_... 或命令行第二个参数 +API version: 2022-06-28 (默认) +""" +import json, os, sys, urllib.request, urllib.error + +NOTION_TOKEN = os.environ.get("NOTION_TOKEN", "") +NOTION_VERSION = os.environ.get("NOTION_VERSION", "2022-06-28") +NOTION_BASE_URL = os.environ.get("NOTION_BASE_URL", "https://api.notion.com") + +class NotionMCP: + def __init__(self, token): + if not token: + raise ValueError("需要 NOTION_TOKEN 环境变量或第二个参数") + self.token, self.base_url, self.version = token, NOTION_BASE_URL, NOTION_VERSION + + def _req(self, method, path, body=None): + url = f"{self.base_url}{path}" + data = json.dumps(body).encode() if body else None + req = urllib.request.Request(url, data=data, method=method, + headers={"Authorization": f"Bearer {self.token}", "Notion-Version": self.version, "Content-Type": "application/json"}) + try: + with urllib.request.urlopen(req, timeout=30) as r: + return json.loads(r.read().decode()) + except urllib.error.HTTPError as e: + return {"error": True, "status": e.code, "body": e.read().decode()[:500]} + + def check(self): return self._req("GET", "/v1/users/me") + def search(self, q, f=None, n=10): + b = {"query": q, "page_size": n} + if f: b["filter"] = {"value": f, "property": "object"} + return self._req("POST", "/v1/search", b) + def get_page(self, pid): return self._req("GET", f"/v1/pages/{pid}") + def get_blocks(self, pid): return self._req("GET", f"/v1/blocks/{pid}/children?page_size=100") + def create_page(self, parent, title, ptype="page_id"): + return self._req("POST", "/v1/pages", {"parent": {ptype: parent}, "properties": {"title": [{"type": "text", "text": {"content": title}}]}}) + def append_block(self, pid, text, btype="paragraph"): + return self._req("PATCH", f"/v1/blocks/{pid}/children", {"children": [{"object": "block", "type": btype, btype: {"rich_text": [{"type": "text", "text": {"content": text}}]}}]}) + def update_todo(self, bid, checked): + return self._req("PATCH", f"/v1/blocks/{bid}", {"to_do": {"checked": checked}}) + def query_db(self, did, f=None, n=100): + b = {"page_size": n} + if f: b["filter"] = f + return self._req("POST", f"/v1/databases/{did}/query", b) + + +def main(): + if len(sys.argv) < 2 or sys.argv[1] in ("-h", "--help"): + print(__doc__); return 0 + cmd = sys.argv[1] + token = NOTION_TOKEN + ao = 2 + if len(sys.argv) > 2 and sys.argv[2].startswith("ntn_"): + token, ao = sys.argv[2], 3 + m = NotionMCP(token) + methods = {"check": m.check, "search": lambda: m.search(sys.argv[ao]), + "get": lambda: m.get_page(sys.argv[ao]), + "blocks": lambda: m.get_blocks(sys.argv[ao]), + "create": lambda: m.create_page(sys.argv[ao], sys.argv[ao+1]), + "append": lambda: m.append_block(sys.argv[ao], sys.argv[ao+1]), + "check-todo": lambda: m.update_todo(sys.argv[ao], sys.argv[ao+1].lower()=="true" if len(sys.argv)>ao+1 else True), + "query": lambda: m.query_db(sys.argv[ao])} + if cmd not in methods: + print(f"未知命令: {cmd}"); print(__doc__); return 1 + print(json.dumps(methods[cmd](), ensure_ascii=False, indent=2)) + return 0 + +if __name__ == "__main__": + sys.exit(main())