使用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