跳到主要内容

MCP 快速入门

本教程将带你从零开始构建一个完整的 MCP 服务器，包含工具、资源和提示三大核心功能。通过这个实战案例，你将掌握 MCP 开发的核心技能。

目标​

我们将构建一个「笔记管理」MCP 服务器，提供以下功能：

  • 工具：创建笔记、搜索笔记、删除笔记
  • 资源：读取笔记内容、列出所有笔记
  • 提示：笔记摘要模板、笔记审查模板

环境准备​

安装依赖​

# 创建项目目录
mkdir note-mcp-server
cd note-mcp-server

# 创建虚拟环境
python -m venv .venv
source .venv/bin/activate  # Windows: .venv\Scripts\activate

# 安装 MCP SDK
pip install mcp

验证安装​

import mcp
print(mcp.__version__)  # 应输出已安装的版本号

构建服务器​

第一步：基础结构​

创建 server.py 文件，搭建服务器基础结构：

import json
import os
from datetime import datetime
from typing import Any
from pathlib import Path
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import (
    Tool,
    TextContent,
    Resource,
    Prompt,
    PromptArgument
)

# 创建服务器实例
app = Server("note-server")

# 配置笔记存储目录
NOTES_DIR = Path.home() / ".notes"
NOTES_DIR.mkdir(exist_ok=True)

def get_note_path(note_id: str) -> Path:
    """获取笔记文件路径"""
    return NOTES_DIR / f"{note_id}.json"

def load_note(note_id: str) -> dict | None:
    """加载笔记内容"""
    path = get_note_path(note_id)
    if path.exists():
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    return None

def save_note(note_id: str, content: dict):
    """保存笔记"""
    path = get_note_path(note_id)
    with open(path, "w", encoding="utf-8") as f:
        json.dump(content, f, ensure_ascii=False, indent=2)

def list_all_notes() -> list[dict]:
    """列出所有笔记"""
    notes = []
    for path in NOTES_DIR.glob("*.json"):
        with open(path, "r", encoding="utf-8") as f:
            note = json.load(f)
            note["id"] = path.stem
            notes.append(note)
    return sorted(notes, key=lambda x: x.get("created", ""), reverse=True)

这段代码完成了基础设置：

  • 创建了 MCP 服务器实例
  • 设置了笔记存储目录
  • 定义了笔记的加载、保存和列表函数

第二步：定义工具​

工具是 AI 模型可以调用的函数。我们定义三个核心工具：

@app.list_tools()
async def list_tools() -> list[Tool]:
    """返回服务器提供的所有工具"""
    return [
        Tool(
            name="create_note",
            description="创建一条新笔记。笔记将保存到本地存储中。",
            inputSchema={
                "type": "object",
                "properties": {
                    "title": {
                        "type": "string",
                        "description": "笔记标题，简要概括笔记内容"
                    },
                    "content": {
                        "type": "string",
                        "description": "笔记正文内容"
                    },
                    "tags": {
                        "type": "array",
                        "items": {"type": "string"},
                        "description": "标签列表，用于分类",
                        "default": []
                    }
                },
                "required": ["title", "content"]
            }
        ),
        Tool(
            name="search_notes",
            description="搜索笔记。可以在标题、内容或标签中搜索关键词。",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {
                        "type": "string",
                        "description": "搜索关键词"
                    },
                    "search_in": {
                        "type": "string",
                        "enum": ["title", "content", "tags", "all"],
                        "description": "搜索范围",
                        "default": "all"
                    }
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="delete_note",
            description="删除指定的笔记。此操作不可撤销。",
            inputSchema={
                "type": "object",
                "properties": {
                    "note_id": {
                        "type": "string",
                        "description": "要删除的笔记 ID"
                    }
                },
                "required": ["note_id"]
            }
        )
    ]

第三步：实现工具逻辑​

@app.call_tool()
async def call_tool(name: str, arguments: dict | None) -> list[TextContent]:
    """处理工具调用"""
    if arguments is None:
        arguments = {}
    
    if name == "create_note":
        return await handle_create_note(arguments)
    elif name == "search_notes":
        return await handle_search_notes(arguments)
    elif name == "delete_note":
        return await handle_delete_note(arguments)
    
    raise ValueError(f"Unknown tool: {name}")


async def handle_create_note(args: dict) -> list[TextContent]:
    """处理创建笔记"""
    title = args["title"]
    content = args["content"]
    tags = args.get("tags", [])
    
    # 生成唯一 ID
    note_id = datetime.now().strftime("%Y%m%d%H%M%S")
    
    # 构建笔记对象
    note = {
        "title": title,
        "content": content,
        "tags": tags,
        "created": datetime.now().isoformat(),
        "modified": datetime.now().isoformat()
    }
    
    # 保存笔记
    save_note(note_id, note)
    
    return [TextContent(
        type="text",
        text=f"笔记创建成功！\n\nID: {note_id}\n标题: {title}\n标签: {', '.join(tags) if tags else '无'}"
    )]


async def handle_search_notes(args: dict) -> list[TextContent]:
    """处理搜索笔记"""
    query = args["query"].lower()
    search_in = args.get("search_in", "all")
    
    all_notes = list_all_notes()
    results = []
    
    for note in all_notes:
        matched = False
        
        if search_in in ["title", "all"]:
            if query in note.get("title", "").lower():
                matched = True
        
        if search_in in ["content", "all"] and not matched:
            if query in note.get("content", "").lower():
                matched = True
        
        if search_in in ["tags", "all"] and not matched:
            tags = note.get("tags", [])
            if any(query in tag.lower() for tag in tags):
                matched = True
        
        if matched:
            results.append(note)
    
    if not results:
        return [TextContent(
            type="text",
            text=f"未找到包含「{query}」的笔记。"
        )]
    
    output = f"找到 {len(results)} 条笔记：\n\n"
    for note in results:
        output += f"- [{note['id']}] {note['title']}\n"
        if note.get("tags"):
            output += f"  标签: {', '.join(note['tags'])}\n"
    
    return [TextContent(type="text", text=output)]


async def handle_delete_note(args: dict) -> list[TextContent]:
    """处理删除笔记"""
    note_id = args["note_id"]
    
    # 检查笔记是否存在
    note = load_note(note_id)
    if not note:
        return [TextContent(
            type="text",
            text=f"笔记 {note_id} 不存在。"
        )]
    
    # 删除笔记
    get_note_path(note_id).unlink()
    
    return [TextContent(
        type="text",
        text=f"笔记「{note['title']}」已删除。"
    )]

第四步：定义资源​

资源用于提供上下文数据，让 AI 可以读取笔记内容：

@app.list_resources()
async def list_resources() -> list[Resource]:
    """返回所有可用的资源"""
    notes = list_all_notes()
    
    resources = [
        Resource(
            uri="notes://list",
            name="所有笔记列表",
            description="包含所有笔记的概览信息",
            mimeType="application/json"
        )
    ]
    
    # 为每条笔记创建资源
    for note in notes:
        resources.append(Resource(
            uri=f"notes://{note['id']}",
            name=note["title"],
            description=f"笔记: {note['title']}",
            mimeType="application/json"
        ))
    
    return resources


@app.read_resource()
async def read_resource(uri: str) -> list[TextContent]:
    """读取资源内容"""
    if uri == "notes://list":
        notes = list_all_notes()
        return [TextContent(
            type="text",
            text=json.dumps(notes, ensure_ascii=False, indent=2)
        )]
    
    if uri.startswith("notes://"):
        note_id = uri[8:]  # 移除 "notes://" 前缀
        note = load_note(note_id)
        
        if note:
            return [TextContent(
                type="text",
                text=json.dumps(note, ensure_ascii=False, indent=2)
            )]
        
        return [TextContent(
            type="text",
            text=f"笔记 {note_id} 不存在"
        )]
    
    raise ValueError(f"Unknown resource: {uri}")

第五步：定义提示模板​

提示模板是预定义的工作流，帮助用户完成特定任务：

@app.list_prompts()
async def list_prompts() -> list[Prompt]:
    """返回所有可用的提示模板"""
    return [
        Prompt(
            name="summarize_notes",
            description="总结所有笔记的内容，生成摘要报告",
            arguments=[
                PromptArgument(
                    name="tags",
                    description="只总结包含这些标签的笔记，多个标签用逗号分隔",
                    required=False
                )
            ]
        ),
        Prompt(
            name="review_note",
            description="审查指定笔记，检查内容质量和提供建议",
            arguments=[
                PromptArgument(
                    name="note_id",
                    description="要审查的笔记 ID",
                    required=True
                )
            ]
        )
    ]


@app.get_prompt()
async def get_prompt(name: str, arguments: dict | None) -> list[TextContent]:
    """获取提示模板内容"""
    if arguments is None:
        arguments = {}
    
    if name == "summarize_notes":
        tags_filter = arguments.get("tags", "")
        
        # 获取笔记列表
        notes = list_all_notes()
        
        # 如果有标签过滤
        if tags_filter:
            filter_tags = [t.strip().lower() for t in tags_filter.split(",")]
            notes = [n for n in notes if any(
                tag.lower() in filter_tags 
                for tag in n.get("tags", [])
            )]
        
        prompt = f"""请对以下 {len(notes)} 条笔记进行总结，生成一份摘要报告：

{json.dumps(notes, ensure_ascii=False, indent=2)}

请提供：
1. 整体主题概述
2. 每条笔记的要点总结
3. 笔记之间的关联性分析
4. 建议的后续行动
"""
        return [TextContent(type="text", text=prompt)]
    
    if name == "review_note":
        note_id = arguments.get("note_id")
        note = load_note(note_id)
        
        if not note:
            return [TextContent(
                type="text",
                text=f"笔记 {note_id} 不存在"
            )]
        
        prompt = f"""请审查以下笔记：

标题：{note['title']}
内容：
{note['content']}

标签：{', '.join(note.get('tags', []))}

请提供：
1. 内容质量评估
2. 结构和可读性分析
3. 改进建议
4. 推荐添加的标签
"""
        return [TextContent(type="text", text=prompt)]
    
    raise ValueError(f"Unknown prompt: {name}")

第六步：启动服务器​

async def main():
    """主函数：启动 MCP 服务器"""
    # 使用 STDIO 传输运行服务器
    async with stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            app.create_initialization_options()
        )


if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

测试服务器​

配置 Claude Desktop​

编辑 Claude Desktop 配置文件：

  • macOS: ~/Library/Application Support/Claude/claude_desktop_config.json
  • Windows: %APPDATA%\Claude\claude_desktop_config.json
  • Linux: ~/.config/Claude/claude_desktop_config.json

添加服务器配置：

{
  "mcpServers": {
    "notes": {
      "command": "python",
      "args": ["/absolute/path/to/server.py"]
    }
  }
}

重启 Claude Desktop 后，你就可以在对话中使用这个笔记服务器了。

测试对话示例​

创建笔记：

用户：帮我创建一条笔记，记录今天的会议要点：讨论了 Q2 目标和团队扩展计划

Claude：我来帮你创建这条笔记。
[调用 create_note 工具]
笔记创建成功！ID: 20260327143000，标题: 会议记录

搜索笔记：

用户：搜索包含「会议」的笔记

Claude：[调用 search_notes 工具]
找到 1 条笔记：
- [20260327143000] 会议记录
  标签: 无

使用提示模板：

用户：总结所有笔记

Claude：[使用 summarize_notes 提示模板]
根据你的笔记，以下是摘要报告：
1. 整体主题：团队管理相关记录
2. 要点：Q2 目标规划、团队扩展...

完整代码​

完整的 server.py 文件：

import json
from datetime import datetime
from pathlib import Path
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent, Resource, Prompt, PromptArgument

app = Server("note-server")
NOTES_DIR = Path.home() / ".notes"
NOTES_DIR.mkdir(exist_ok=True)

# 辅助函数
def get_note_path(note_id: str) -> Path:
    return NOTES_DIR / f"{note_id}.json"

def load_note(note_id: str) -> dict | None:
    path = get_note_path(note_id)
    if path.exists():
        with open(path, "r", encoding="utf-8") as f:
            return json.load(f)
    return None

def save_note(note_id: str, content: dict):
    path = get_note_path(note_id)
    with open(path, "w", encoding="utf-8") as f:
        json.dump(content, f, ensure_ascii=False, indent=2)

def list_all_notes() -> list[dict]:
    notes = []
    for path in NOTES_DIR.glob("*.json"):
        with open(path, "r", encoding="utf-8") as f:
            note = json.load(f)
            note["id"] = path.stem
            notes.append(note)
    return sorted(notes, key=lambda x: x.get("created", ""), reverse=True)

# 工具定义
@app.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="create_note",
            description="创建一条新笔记",
            inputSchema={
                "type": "object",
                "properties": {
                    "title": {"type": "string", "description": "笔记标题"},
                    "content": {"type": "string", "description": "笔记内容"},
                    "tags": {"type": "array", "items": {"type": "string"}, "default": []}
                },
                "required": ["title", "content"]
            }
        ),
        Tool(
            name="search_notes",
            description="搜索笔记",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string", "description": "搜索关键词"}
                },
                "required": ["query"]
            }
        ),
        Tool(
            name="delete_note",
            description="删除笔记",
            inputSchema={
                "type": "object",
                "properties": {
                    "note_id": {"type": "string", "description": "笔记 ID"}
                },
                "required": ["note_id"]
            }
        )
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict | None) -> list[TextContent]:
    if arguments is None:
        arguments = {}
    
    if name == "create_note":
        note_id = datetime.now().strftime("%Y%m%d%H%M%S")
        note = {
            "title": arguments["title"],
            "content": arguments["content"],
            "tags": arguments.get("tags", []),
            "created": datetime.now().isoformat()
        }
        save_note(note_id, note)
        return [TextContent(type="text", text=f"笔记创建成功！ID: {note_id}")]
    
    elif name == "search_notes":
        query = arguments["query"].lower()
        results = [n for n in list_all_notes() 
                   if query in n.get("title", "").lower() 
                   or query in n.get("content", "").lower()]
        return [TextContent(type="text", text=json.dumps(results, ensure_ascii=False, indent=2))]
    
    elif name == "delete_note":
        note_id = arguments["note_id"]
        path = get_note_path(note_id)
        if path.exists():
            path.unlink()
            return [TextContent(type="text", text=f"笔记 {note_id} 已删除")]
        return [TextContent(type="text", text=f"笔记 {note_id} 不存在")]
    
    raise ValueError(f"Unknown tool: {name}")

# 资源定义
@app.list_resources()
async def list_resources() -> list[Resource]:
    notes = list_all_notes()
    resources = [Resource(uri="notes://list", name="所有笔记", mimeType="application/json")]
    for note in notes:
        resources.append(Resource(
            uri=f"notes://{note['id']}",
            name=note["title"],
            mimeType="application/json"
        ))
    return resources

@app.read_resource()
async def read_resource(uri: str) -> list[TextContent]:
    if uri == "notes://list":
        return [TextContent(type="text", text=json.dumps(list_all_notes(), ensure_ascii=False, indent=2))]
    if uri.startswith("notes://"):
        note = load_note(uri[8:])
        if note:
            return [TextContent(type="text", text=json.dumps(note, ensure_ascii=False, indent=2))]
    raise ValueError(f"Unknown resource: {uri}")

# 提示定义
@app.list_prompts()
async def list_prompts() -> list[Prompt]:
    return [
        Prompt(name="summarize_notes", description="总结所有笔记"),
        Prompt(name="review_note", description="审查笔记",
               arguments=[PromptArgument(name="note_id", required=True)])
    ]

@app.get_prompt()
async def get_prompt(name: str, arguments: dict | None) -> list[TextContent]:
    if arguments is None:
        arguments = {}
    
    if name == "summarize_notes":
        notes = list_all_notes()
        return [TextContent(type="text", text=f"请总结以下笔记：\n{json.dumps(notes, ensure_ascii=False, indent=2)}")]
    
    if name == "review_note":
        note_id = arguments.get("note_id")
        note = load_note(note_id)
        if note:
            return [TextContent(type="text", text=f"请审查笔记：\n{json.dumps(note, ensure_ascii=False, indent=2)}")]
        return [TextContent(type="text", text=f"笔记 {note_id} 不存在")]
    
    raise ValueError(f"Unknown prompt: {name}")

# 启动服务器
async def main():
    async with stdio_server() as (read_stream, write_stream):
        await app.run(read_stream, write_stream, app.create_initialization_options())

if __name__ == "__main__":
    import asyncio
    asyncio.run(main())

下一步​