手把手写一个 MCP Server：把 Notion 变成 AI 的知识库

上篇讲了 MCP 是什么、怎么装现成的。这篇我们做个更有意思的事——自己写一个 MCP Server，把 Notion 数据库变成 AI 能查的知识库。

完成后效果：你在 Claude Desktop（或 Cline、Open WebUI）里问「帮我查一下 Notion 里关于 Q3 计划的笔记」，AI 自动调用 MCP Server 从你的 Notion 数据库里查，然后回答你。

先看最终效果

配置好之后的工作流是这样的：

你：帮我查一下产品路线图里 Q3 计划做什么
                    ↓
AI 自动调用 → notion-search MCP Server → 查询 Notion API
                    ↓
返回结果：产品路线图数据库中有 3 条 Q3 相关的条目...
                    ↓
AI 总结后回答你

整个过程你只需要跟 AI 说一句话，中间的技术细节——API 调用、数据解析、格式转换——全部由 MCP Server 搞定。

准备工作

你需要： - Python 3.10+ - 一个 Notion 账号 + 集成 Token - 懂一点 Python（只用标准库，不需要高级技巧）

获取 Notion API Token

1. 打开 https://www.notion.so/my-integrations
2. 点击「新建集成」，取名「My MCP Server」
3. 选择你工作区
4. 复制 Internal Integration Token（以 ntn_ 或 secret_ 开头）
5. 在 Notion 中打开你要查询的页面，右上角「…」→ 添加连接 → 选择你的集成

第一步：项目结构

mcp-notion-server/
├── server.py       # MCP Server 主文件
├── notion.py       # Notion API 封装
├── requirements.txt
└── README.md

第二步：实现 Notion API 封装

# notion.py
import requests
from typing import list, dict, Any

class NotionClient:
    def __init__(self, token: str):
        self.headers = {
            "Authorization": f"Bearer {token}",
            "Notion-Version": "2022-06-28",
            "Content-Type": "application/json"
        }
        self.base_url = "https://api.notion.com/v1"

    def search_pages(self, query: str) -> list[dict[str, Any]]:
        """搜索 Notion 页面"""
        response = requests.post(
            f"{self.base_url}/search",
            headers=self.headers,
            json={
                "query": query,
                "sort": {"direction": "descending", "timestamp": "last_edited_time"}
            }
        )
        return response.json().get("results", [])

    def get_page_content(self, page_id: str) -> str:
        """获取页面内容"""
        response = requests.get(
            f"{self.base_url}/blocks/{page_id}/children",
            headers=self.headers
        )
        blocks = response.json().get("results", [])
        return self._blocks_to_text(blocks)

    def query_database(self, db_id: str, filter_dict: dict = None) -> list[dict]:
        """查询数据库"""
        payload = {}
        if filter_dict:
            payload["filter"] = filter_dict
        response = requests.post(
            f"{self.base_url}/databases/{db_id}/query",
            headers=self.headers,
            json=payload
        )
        return response.json().get("results", [])

    def _blocks_to_text(self, blocks: list) -> str:
        """把 Notion block 转成纯文本"""
        text_parts = []
        for block in blocks:
            block_type = block.get("type", "")
            if block_type in ["paragraph", "heading_1", "heading_2", "heading_3", "bulleted_list_item", "numbered_list_item"]:
                texts = block.get(block_type, {}).get("rich_text", [])
                text = "".join([t.get("plain_text", "") for t in texts])
                if text.strip():
                    text_parts.append(text)
            elif block_type == "code":
                texts = block.get("code", {}).get("rich_text", [])
                text = "".join([t.get("plain_text", "") for t in texts])
                if text.strip():
                    text_parts.append(f"```\n{text}\n```")
        return "\n".join(text_parts)

第三步：写 MCP Server

# server.py
import os
import json
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent, Resource, ResourceTemplate
from notion import NotionClient

# 初始化
NOTION_TOKEN = os.environ.get("NOTION_TOKEN", "")
notion = NotionClient(NOTION_TOKEN) if NOTION_TOKEN else None

app = Server("notion-mcp")

@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="search_notion",
            description="在 Notion 中搜索页面",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "搜索关键词"}
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="read_page",
            description="读取 Notion 页面内容",
            inputSchema={
                "type": "object",
                "properties": {
                    "page_id": {"type": "string", "description": "页面 ID（URL 中的那串 hash）"}
                },
                "required": ["page_id"]
            }
        ),
        Tool(
            name="query_database",
            description="查询 Notion 数据库",
            inputSchema={
                "type": "object",
                "properties": {
                    "database_id": {"type": "string", "description": "数据库 ID"},
                    "filter_text": {"type": "string", "description": "按标题筛选（可选）"}
                },
                "required": ["database_id"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    if not notion:
        return [TextContent(type="text", text="错误：未配置 NOTION_TOKEN")]

    if name == "search_notion":
        query = arguments["query"]
        results = notion.search_pages(query)
        simplified = []
        for r in results[:10]:
            props = r.get("properties", {})
            title = "无标题"
            for prop in props.values():
                if prop.get("type") == "title":
                    title_texts = prop.get("title", [])
                    title = "".join([t.get("plain_text", "") for t in title_texts])
                    break
            simplified.append({
                "id": r["id"],
                "title": title,
                "url": r.get("url", ""),
                "type": r.get("object", "")
            })
        return [TextContent(type="text", text=json.dumps(simplified, ensure_ascii=False, indent=2))]

    elif name == "read_page":
        page_id = arguments["page_id"]
        content = notion.get_page_content(page_id)
        return [TextContent(type="text", text=content)]

    elif name == "query_database":
        db_id = arguments["database_id"]
        filter_text = arguments.get("filter_text", "")
        filter_dict = None
        if filter_text:
            filter_dict = {
                "property": "标题",
                "title": {"contains": filter_text}
            }
        results = notion.query_database(db_id, filter_dict)
        items = []
        for r in results[:20]:
            props = r.get("properties", {})
            title = "无标题"
            for prop in props.values():
                if prop.get("type") == "title":
                    title_texts = prop.get("title", [])
                    title = "".join([t.get("plain_text", "") for t in title_texts])
                    break
            items.append({"id": r["id"], "title": title})
        return [TextContent(type="text", text=json.dumps(items, ensure_ascii=False, indent=2))]

    raise ValueError(f"未知工具: {name}")

if __name__ == "__main__":
    if not NOTION_TOKEN:
        print("请设置环境变量 NOTION_TOKEN")
        exit(1)
    import asyncio
    asyncio.run(stdio_server(app))

第四步：配置客户端使用

Claude Desktop

{
  "mcpServers": {
    "notion-server": {
      "command": "python",
      "args": ["C:\\path\\to\\mcp-notion-server\\server.py"],
      "env": {
        "NOTION_TOKEN": "ntn_你的token"
      }
    }
  }
}

Cline (VS Code 插件)

在 Cline 设置中，MCP Server 配置同样格式：

{
  "mcpServers": {
    "notion-server": {
      "command": "python",
      "args": ["C:/path/to/mcp-notion-server/server.py"],
      "env": {
        "NOTION_TOKEN": "ntn_你的token"
      }
    }
  }
}

第五步：测试

重启你的 AI 客户端，然后试着问：

「帮我查一下 Notion 里关于 AI agent 的笔记」

正常的话，你会看到这样的 [使用工具: search_notion] 提示，然后 AI 自动调用你的 MCP Server，搜索 Notion，然后把结果总结给你。

进阶：添加自动补全和资源模板

MCP 还支持 Resource 和 ResourceTemplate，让 AI 能像浏览网站一样浏览你的数据：

@app.list_resources()
async def list_resources() -> list[Resource]:
    # 列出常用的 Notion 数据库
    return [
        Resource(
            uri="notion://databases/xxx",
            name="产品路线图",
            mimeType="application/json"
        )
    ]

@app.resource_templates()
async def resource_templates() -> list[ResourceTemplate]:
    return [
        ResourceTemplate(
            uriTemplate="notion://pages/{page_id}",
            nameTemplate="Notion 页面 {page_id}",
            mimeType="text/markdown"
        )
    ]

安全性提醒

MCP Server 以你的身份运行，能访问你的数据。几点注意：

1. 永远不要把含真实 Token 的配置上传到 GitHub
2. 用 .env 文件管理 Token，加 .gitignore
3. 只在可信的 AI 客户端上启用自定义 MCP Server
4. 定期轮换 Notion Integration Token

总结

写一个自定义 MCP Server 并不难——核心就是三个步骤定义工具、实现逻辑、暴露服务。上面这个 Notion MCP Server 不到 200 行代码，但完成后你的 AI 助手直接「长出了手」，能主动去查你的知识库。

有了这个基础，你可以把任何 API 包装成 MCP Server——GitHub、Jira、飞书文档、内部管理系统……一个 MCP Server 大一统。

🔗 相关文章： 本地 LLM 推理方案横评：LM Studio vs Ollama vs vLLM