Python MCP协议开发本地知识便签服务完整实战

脚本专家 · 发表于 2 小时前

MCP（Model Context Protocol，模型上下文协议）正成为AI应用连接外部数据与工具的标准方式。对于Python开发者来说，理解并实现一个MCP Server并不复杂。本文将以一个本地知识便签服务为例，从零实现一个支持新增、搜索、删除、资源读取和提示模板的MCP Server，并用Python Client与MCP Inspector完成全链路验证。

一、MCP核心概念与架构

大模型本身只擅长处理输入内容，无法直接访问本地文件、数据库或内部API。传统做法是为每个数据源编写私有适配，导致重复集成。MCP在中间定义统一协议后，AI应用通过MCP Client连接MCP Server，Server以统一方式声明工具、资源和提示模板，Client可以动态发现并调用这些能力。

MCP采用Host–Client–Server三层架构。Host是承载用户交互与模型的AI应用，Client是Host内部负责与某个Server通信的协议客户端，Server对外提供Tools、Resources、Prompts等能力。一个Host可以同时连接多个Server，每个Server拥有独立Client，便于隔离权限和上下文。

MCP的三种核心能力需要区分：

- Tools：可被模型或Host调用的函数，执行动作，可能有副作用，如查询、写入、删除。
- Resources：可寻址的数据，由应用读取，通常只读，如文件内容、数据库记录。
- Prompts：参数化的交互模板，由用户或Host显式选择，用于封装最佳使用方式。

判断方法很简单：这是一个动词、一个可寻址名词，还是一个对话模板？

MCP与Function Calling、RAG、Agent解决的是不同层次的问题。Function Calling解决模型如何表达调用函数，RAG解决如何检索外部知识，Agent解决如何规划行动，而MCP解决的是AI应用如何发现、连接并调用外部能力。它们可以协作：Agent判断需要外部知识，模型通过Function Calling选择工具，MCP Client调用Server的检索Tool，Server执行检索并返回证据。

MCP主要有两种传输方式：

- stdio：本地Server，Client启动子进程，通过标准输入输出交换消息。适合文件、IDE等本地能力。注意：不要向stdout打印调试信息，stdout是协议通道，日志应输出到stderr。
- Streamable HTTP：远程Server，通过HTTPS提供MCP端点，需要处理认证、授权、限流、审计等网络风险。

本文使用stdio方式，依赖最少，最适合第一次跑通。

二、实战目标与环境准备

我们将实现一个notes-assistant Server，具备以下能力：

- Tools：add_note新增便签；search_notes按关键词搜索；delete_note删除便签（需显式确认）。
- Resources：notes://index读取便签索引；notes://{note_id}读取指定便签。
- Prompt：summarize_topic生成主题总结任务模板。

数据保存在本地JSON文件，无需数据库和API Key。

环境要求：Python 3.10+，推荐使用uv管理虚拟环境，也可使用pip。

创建项目：

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

复制代码

若无uv：

python -m venv .venv
source .venv/bin/activate # macOS/Linux
pip install "mcp[cli]"

复制代码

最终目录结构：

mcp-notes-demo/
├── data/
├── server.py
├── client.py
└── pyproject.toml

复制代码

三、完整实现MCP Server

创建server.py，完整代码如下：

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"]}) — `{tags}` — {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'- 标签：`{tags}`\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__":
mcp.run(transport="stdio")

复制代码

代码关键点：

- FastMCP("notes-assistant") 创建Server。
- @mcp.tool() 从类型注解和Docstring自动生成工具定义。
- @mcp.resource() 声明固定Resource和动态Resource模板。
- @mcp.prompt() 暴露参数化Prompt。
- 所有外部参数都在Server端校验，不信任模型输入。
- 删除操作要求confirm=true，体现高风险动作的二次确认思想。
- 搜索只返回摘要，完整内容按需读取，避免上下文膨胀。

四、使用MCP Inspector调试Server

在项目目录执行：

uv run mcp dev server.py

复制代码

如果使用pip环境，按当前SDK提供的CLI方式启动Inspector。浏览器打开终端显示的地址，完成以下检查：

1. 确认Server初始化成功；
2. 查看Tools列表，检查参数Schema和描述；
3. 调用add_note，参数示例：{"title":"MCP学习记录","content":"MCP Server可以通过Tools、Resources和Prompts暴露能力。","tags":["MCP","Python"]}；
4. 调用search_notes，关键词填MCP；
5. 打开notes://index；
6. 使用新增便签的ID读取notes://{note_id}；
7. 查看summarize_topic Prompt的展开结果；
8. 以confirm=false调用删除，观察返回信息。

如果Tools页面没有工具，优先检查装饰器是否在mcp.run()之前执行；如果Server直接退出，查看终端stderr日志。

五、编写Python MCP Client验证协议调用

创建client.py：

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.description)
if __name__ == "__main__":
asyncio.run(main())

复制代码

运行客户端：

python client.py

复制代码

输出中可以看到工具列表、新增结果、搜索结果、资源URI和Prompt信息，证明整个MCP协议链路已经打通。

六、接入MCP客户端与工具设计要点

MCP Inspector和Python Client验证通过后，可以将Server接入支持MCP的AI客户端。配置时指定command为python、args为server.py路径即可。

设计高质量的MCP Tool需要注意：

- 名称要明确：如add_note、search_notes，不要用模糊动词。
- 描述要告诉模型“何时使用”：在Docstring中写清楚适用场景。
- 参数应该窄而具体：限制长度、枚举值、范围。
- 返回模型真正需要的信息：搜索返回摘要而非全文，避免上下文膨胀。
- 区分只读与写操作：写操作应要求确认或增加权限提示。

七、错误处理、安全与可观测性

错误处理方面，不要把Python堆栈直接扔给模型。应区分三类错误：

- 参数错误：ValueError，给出清晰的中文提示。
- 业务错误：如找不到便签，返回结构化结果。
- 系统错误：如文件读取失败，抛RuntimeError并记录日志。

安全方面，MCP接通的不只是数据，也是权限。需要注意：

- 最小权限：Server只访问必要路径，不开放无关文件。
- 服务端校验：所有参数必须在Server端校验，不能信任模型。
- 人工确认：删除、发送等高风险操作应要求二次确认。
- Prompt注入：模型可能被诱导，不要将不可信内容拼入系统指令。
- 远程Server：使用Streamable HTTP时需考虑鉴权、限流和审计。

日志方面，stdio模式下stdout是协议通道，日志只能写到stderr或文件。应记录关键操作：新增、删除的ID，错误堆栈，调用频率等。测试建议分层：先用Inspector人工测试，再用Client脚本自动化验证，最后接入AI客户端做端到端测试。

八、常见问题排查

1. Server一启动就退出：检查stderr日志，通常是依赖缺失或路径错误。
2. 客户端提示找不到命令：确认command参数使用sys.executable，避免使用虚拟环境外的python。
3. 日志导致协议解析失败：检查是否向stdout打印了普通print等信息，这是stdio模式最常见的坑。
4. Inspector能用但AI客户端不能用：检查AI客户端的MCP Server配置是否正确，是否支持当前SDK版本。
5. 工具可见但模型总选错：优化工具描述，明确触发条件，避免多个工具功能重叠。
6. 动态Resource没出现在固定资源列表：这是正常现象，动态Resource需通过模板模式访问，不会像固定Resource一样列出。

九、从示例升级到生产项目

当前示例使用了JSON文件存储，适合演示。生产环境可以分阶段升级：

- 第一阶段：替换存储为SQLite或PostgreSQL，保证并发与事务。
- 第二阶段：增加向量化语义检索，使search_notes支持相似度匹配。
- 第三阶段：部署为Streamable HTTP远程服务，增加认证与限流。
- 第四阶段：与Agent工作流集成，让模型自主规划并调用多个工具完成复杂任务。

十、什么时候适合使用MCP？

如果你有多个AI应用需要连接同一套本地数据或内部工具，或者希望以标准方式暴露能力，MCP是一个值得考虑的方案。它特别适合本地开发工具、文件管理、知识库、数据库查询等场景。如果只是单次调用的简单脚本，直接用Function Calling可能更轻量。

通过本文的实战，你已经能够从零实现一个可用的MCP Server，并用Python Client完整验证协议。后续可以在这个基础上扩展更多工具和资源，构建真正属于自己的MCP生态。

Python MCP协议开发本地知识便签服务完整实战

浏览过的版块

指导单位

旗下站点

联系我们