使用Python+MCP构建一个真正可用的本地知识便签服务

mkdir mcp-notes-demo
cd mcp-notes-demo
uv init
uv add "mcp[cli]"

python -m venv .venv

# windows powershell
.venv\scripts\activate.ps1

# macos / linux
source .venv/bin/activate

pip install "mcp[cli]"

from __future__ import annotations
import json
import logging
import sys
from datetime import datetime, timezone
from pathlib import path
from threading import lock
from uuid import uuid4
from mcp.server.fastmcp import fastmcp
# stdio 模式下 stdout 用于 mcp 协议，日志必须写到 stderr
logging.basicconfig(
    stream=sys.stderr,
    level=logging.info,
    format="%(asctime)s %(levelname)s %(message)s",
)
logger = logging.getlogger("notes-mcp")
mcp = fastmcp("notes-assistant")
base_dir = path(__file__).resolve().parent
data_dir = base_dir / "data"
notes_file = data_dir / "notes.json"
file_lock = lock()
def ensure_storage() -> none:
    data_dir.mkdir(parents=true, exist_ok=true)
    if not notes_file.exists():
        notes_file.write_text("[]", encoding="utf-8")
def load_notes() -> list[dict]:
    ensure_storage()
    with file_lock:
        try:
            content = notes_file.read_text(encoding="utf-8")
            data = json.loads(content)
        except (oserror, json.jsondecodeerror) as exc:
            logger.exception("failed to read notes")
            raise runtimeerror("便签存储暂时不可用") from exc
    if not isinstance(data, list):
        raise runtimeerror("便签文件格式错误")
    return data
def save_notes(notes: list[dict]) -> none:
    ensure_storage()
    # 先写临时文件，再替换正式文件，降低写到一半损坏数据的概率
    temp_file = notes_file.with_suffix(".tmp")
    payload = json.dumps(notes, ensure_ascii=false, indent=2)
    with file_lock:
        try:
            temp_file.write_text(payload, encoding="utf-8")
            temp_file.replace(notes_file)
        except oserror as exc:
            logger.exception("failed to save notes")
            raise runtimeerror("便签保存失败") from exc
@mcp.tool()
def add_note(title: str, content: str, tags: list[str] | none = none) -> dict:
    """新增一条本地便签。
    args:
        title: 便签标题，不能为空，最长 100 个字符。
        content: 便签正文，不能为空，最长 5000 个字符。
        tags: 可选标签列表，每条最多 20 个字符。
    """
    title = title.strip()
    content = content.strip()
    normalized_tags = [tag.strip() for tag in (tags or []) if tag.strip()]
    if not title or len(title) > 100:
        raise valueerror("title 长度必须在 1 到 100 之间")
    if not content or len(content) > 5000:
        raise valueerror("content 长度必须在 1 到 5000 之间")
    if len(normalized_tags) > 10:
        raise valueerror("tags 最多包含 10 项")
    if any(len(tag) > 20 for tag in normalized_tags):
        raise valueerror("每个 tag 最长 20 个字符")
    notes = load_notes()
    note = {
        "id": uuid4().hex[:12],
        "title": title,
        "content": content,
        "tags": normalized_tags,
        "created_at": datetime.now(timezone.utc).isoformat(),
    }
    notes.append(note)
    save_notes(notes)
    logger.info("created note id=%s", note["id"])
    return note
@mcp.tool()
def search_notes(keyword: str = "", limit: int = 10) -> list[dict]:
    """按标题、正文或标签搜索便签；keyword 为空时返回最近便签。"""
    keyword = keyword.strip().casefold()
    if not 1 <= limit <= 50:
        raise valueerror("limit 必须在 1 到 50 之间")
    notes = load_notes()
    notes.reverse()
    if keyword:
        notes = [
            note
            for note in notes
            if keyword
            in " ".join(
                [
                    note.get("title", ""),
                    note.get("content", ""),
                    *note.get("tags", []),
                ]
            ).casefold()
        ]
    # 搜索列表只返回摘要，完整内容通过 resource 获取
    return [
        {
            "id": note["id"],
            "title": note["title"],
            "preview": note["content"][:120],
            "tags": note.get("tags", []),
            "created_at": note["created_at"],
        }
        for note in notes[:limit]
    ]
@mcp.tool()
def delete_note(note_id: str, confirm: bool = false) -> dict:
    """删除指定便签。只有 confirm=true 时才会真正删除。"""
    note_id = note_id.strip()
    if not confirm:
        return {
            "deleted": false,
            "message": "这是写操作，请确认 note_id 后以 confirm=true 再次调用。",
        }
    notes = load_notes()
    remaining = [note for note in notes if note.get("id") != note_id]
    if len(remaining) == len(notes):
        return {"deleted": false, "message": "未找到对应便签"}
    save_notes(remaining)
    logger.info("deleted note id=%s", note_id)
    return {"deleted": true, "id": note_id}
@mcp.resource("notes://index")
def get_notes_index() -> str:
    """返回所有便签的 markdown 索引。"""
    notes = load_notes()
    if not notes:
        return "# 便签索引\n\n当前没有便签。"
    lines = ["# 便签索引", ""]
    for note in reversed(notes):
        tags = ", ".join(note.get("tags", [])) or "无标签"
        lines.append(
            f'- [{note["title"]}](notes://{note["id"]}) '
            f'— `python
便签
mcp
` — {note["created_at"]}'
        )
    return "\n".join(lines)
@mcp.resource("notes://{note_id}")
def get_note(note_id: str) -> str:
    """通过 notes://{note_id} 读取一条完整便签。"""
    note = next(
        (item for item in load_notes() if item.get("id") == note_id),
        none,
    )
    if note is none:
        raise valueerror(f"未找到便签：{note_id}")
    tags = ", ".join(note.get("tags", [])) or "无标签"
    return (
        f'# {note["title"]}\n\n'
        f'- id：`{note["id"]}`\n'
        f'- 标签：python
便签
mcp
\n'
        f'- 创建时间：{note["created_at"]}\n\n'
        f'{note["content"]}'
    )
@mcp.prompt()
def summarize_topic(topic: str, style: str = "要点式") -> str:
    """生成一个基于本地便签总结指定主题的提示模板。"""
    return f"""请总结本地便签中与“{topic}”有关的内容。
执行要求：
1. 先调用 search_notes 搜索“{topic}”；
2. 需要完整内容时读取对应的 notes://{{note_id}} resource；
3. 只根据便签中的内容总结，不要补充未出现的事实；
4. 如果没有相关便签，明确说明资料不足；
5. 使用“{style}”输出，并在结论后标注便签标题。
"""
if __name__ == "__main__":
    # stdio 是本地 mcp server 最常见的传输方式
    mcp.run(transport="stdio")

uv run mcp dev server.py

{
  "title": "mcp 学习记录",
  "content": "mcp server 可以通过 tools、resources 和 prompts 暴露能力。",
  "tags": ["mcp", "python"]
}

import asyncio
import sys
from pathlib import path
from mcp import clientsession, stdioserverparameters
from mcp.client.stdio import stdio_client
base_dir = path(__file__).resolve().parent
async def main() -> none:
    server = stdioserverparameters(
        command=sys.executable,
        args=[str(base_dir / "server.py")],
    )
    async with stdio_client(server) as (read_stream, write_stream):
        async with clientsession(read_stream, write_stream) as session:
            # 必须先初始化，初始化阶段会完成版本和能力协商
            await session.initialize()
            tools = await session.list_tools()
            print("可用工具：", [tool.name for tool in tools.tools])
            created = await session.call_tool(
                "add_note",
                arguments={
                    "title": "第一次调用 mcp tool",
                    "content": "这条便签由 python mcp client 创建。",
                    "tags": ["mcp", "实战"],
                },
            )
            print("新增结果：", created.content)
            searched = await session.call_tool(
                "search_notes",
                arguments={"keyword": "mcp", "limit": 5},
            )
            print("搜索结果：", searched.content)
            resources = await session.list_resources()
            print("固定资源：", [str(item.uri) for item in resources.resources])
            index = await session.read_resource("notes://index")
            print("便签索引：", index.contents[0].text)
            prompts = await session.list_prompts()
            print("可用 prompt：", [item.name for item in prompts.prompts])
            prompt = await session.get_prompt(
                "summarize_topic",
                arguments={"topic": "mcp", "style": "表格"},
            )
            print("prompt：", prompt.messages[0].content)
if __name__ == "__main__":
    asyncio.run(main())

uv run python client.py

python client.py

client 启动 server 子进程
  → initialize 能力协商
  → list_tools 能力发现
  → call_tool 工具调用
  → read_resource 资源读取
  → get_prompt 获取模板

{
  "mcpservers": {
    "notes-assistant": {
      "command": "uv",
      "args": [
        "--directory",
        "c:/projects/mcp-notes-demo",
        "run",
        "python",
        "server.py"
      ]
    }
  }
}

search_notes
get_order_status
create_support_ticket

handle
execute
process_data

def execute(payload: dict) -> dict:
    ...

def search_notes(keyword: str, limit: int = 10) -> list[dict]:
    ...

get_note     # 只读
update_note  # 写入
delete_note  # 高风险写入

{
  "ok": false,
  "error": {
    "code": "note_not_found",
    "message": "未找到指定便签",
    "retryable": false
  }
}

print("server started")

import logging
import sys
logging.basicconfig(stream=sys.stderr, level=logging.info)

get-command uv
get-command python

which uv
which python