实战:路径与工作流

MCP Server 开发实战:从零构建你的第一个 MCP Server

更新 原创整合
标签
mcpserver-developmenttutorialtools

为什么要自己写 MCP Server

社区已有大量现成的 MCP Server(GitHub、Slack、PostgreSQL 等),直接拿来用就行。但以下场景需要你自己写:

  1. 内部工具集成:你公司自建的 API、内部数据库、私有部署的服务,没有现成 Server 可用
  2. 自定义工作流:你需要把多个步骤封装成一个 Tool,比如"检查代码质量 + 跑测试 + 生成报告"一站式完成
  3. 数据安全与隔离:敏感数据不能经过第三方 Server,需要在本地或私有环境运行

一句话判断:凡是 LLM 需要调用但你不想让第三方经手的操作,都值得写一个自己的 MCP Server。


MCP Server 核心概念速览

MCP Server 可以向客户端暴露三种能力:

能力 说明 类比 谁触发
Tools 模型可调用的函数 API endpoint LLM 自动调用
Resources 可读取的结构化数据 GET 请求 应用主动拉取
Prompts 预设的提示模板 快捷指令 用户选择触发

本文聚焦 Tools,因为它是让 LLM "做事" 的核心机制,也是 90% 的实际需求所在。

Tool 的工作流程

1. Server 注册 Tool(名称、描述、参数 schema)
2. Client 通过 tools/list 发现可用 Tool
3. LLM 根据上下文决定调用某个 Tool
4. Client 发送 tools/call 请求
5. Server 执行并返回结果
6. LLM 拿到结果,继续推理

关键点:Tool 的描述(description)决定了 LLM 是否会正确地调用它。 描述写得好,模型用得准;描述模糊,模型会乱调或漏调。


环境准备

Python(推荐入门路径)

# 安装 uv(Python 包管理器,比 pip 快 10-100x)
curl -LsSf https://astral.sh/uv/install.sh | sh

# 创建项目
uv init snippet-server
cd snippet-server
uv venv && source .venv/bin/activate

# 安装 MCP SDK
uv add "mcp[cli]"

要求:Python 3.10+,MCP SDK 1.2.0+。

TypeScript(适合前端/全栈开发者)

mkdir snippet-server && cd snippet-server
npm init -y
npm install @modelcontextprotocol/sdk zod@3
npm install -D @types/node typescript

要求:Node.js 16+。


从零开始:构建一个实用的 MCP Server

我们要构建一个 Snippet Manager:管理代码片段的工具服务器。它提供三个 Tool:

  • save_snippet:保存代码片段,带标题和标签
  • search_snippets:按关键词或标签搜索片段
  • list_snippets:列出所有片段或按标签过滤

这是一个比天气 Demo 更实用的例子——你日常开发中真的会用它来管理代码片段。

第一步:创建 Server 实例

# snippet_server.py
import json
from pathlib import Path
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("snippet-manager")

SNIPPETS_FILE = Path.home() / ".snippets.json"


def _load_snippets() -> list[dict]:
    if not SNIPPETS_FILE.exists():
        return []
    return json.loads(SNIPPETS_FILE.read_text(encoding="utf-8"))


def _save_snippets(snippets: list[dict]) -> None:
    SNIPPETS_FILE.write_text(
        json.dumps(snippets, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )

FastMCP 是 Python MCP SDK 提供的高层封装。它会自动根据类型注解和 docstring 生成 Tool 的 inputSchema,你不需要手写 JSON Schema。

第二步:定义 Tools

@mcp.tool()
def save_snippet(title: str, code: str, tags: str = "") -> str:
    """Save a code snippet with title and optional tags.

    Args:
        title: A short descriptive title for the snippet
        code: The code content to save
        tags: Comma-separated tags for categorization (e.g. "python,async,http")
    """
    snippets = _load_snippets()

    for s in snippets:
        if s["title"] == title:
            s["code"] = code
            s["tags"] = [t.strip() for t in tags.split(",") if t.strip()]
            _save_snippets(snippets)
            return f"Updated snippet: {title}"

    snippets.append({
        "title": title,
        "code": code,
        "tags": [t.strip() for t in tags.split(",") if t.strip()],
    })
    _save_snippets(snippets)
    return f"Saved snippet: {title}"


@mcp.tool()
def search_snippets(query: str) -> str:
    """Search snippets by keyword in title, code, or tags.

    Args:
        query: Search keyword to match against title, code content, or tags
    """
    snippets = _load_snippets()
    query_lower = query.lower()
    results = [
        s for s in snippets
        if query_lower in s["title"].lower()
        or query_lower in s["code"].lower()
        or any(query_lower in t.lower() for t in s["tags"])
    ]

    if not results:
        return f"No snippets found matching '{query}'"

    output = []
    for s in results:
        tag_str = ", ".join(s["tags"]) if s["tags"] else "no tags"
        output.append(f"## {s['title']} [{tag_str}]\n{s['code']}")

    return f"Found {len(results)} snippet(s):\n\n" + "\n\n".join(output)


@mcp.tool()
def list_snippets(tag: str = "") -> str:
    """List all saved snippets, optionally filtered by tag.

    Args:
        tag: Optional tag to filter snippets by
    """
    snippets = _load_snippets()

    if tag:
        tag_lower = tag.lower()
        snippets = [
            s for s in snippets
            if any(tag_lower == t.lower() for t in s["tags"])
        ]

    if not snippets:
        if tag:
            return f"No snippets found with tag '{tag}'"
        return "No snippets saved yet. Use save_snippet to add one."

    lines = []
    for s in snippets:
        tag_str = ", ".join(s["tags"]) if s["tags"] else "no tags"
        lines.append(f"- {s['title']} [{tag_str}]")

    header = f"Snippets" + (f" tagged '{tag}'" if tag else "")
    return f"{header} ({len(snippets)}):\n" + "\n".join(lines)

注意几个关键设计:

  1. docstring 即 schema:FastMCP 会从 docstring 的 Args: 部分提取参数描述,自动生成 inputSchema
  2. 类型注解即验证title: str 会生成 {"type": "string"} 的 schema;tags: str = "" 会标记为非必需参数
  3. 错误处理用文本返回:Tool 不应该抛异常,而是返回描述性错误文本让 LLM 理解发生了什么

第三步:运行 Server

def main():
    mcp.run(transport="stdio")


if __name__ == "__main__":
    main()

启动:

uv run snippet_server.py

Server 会通过 stdio 监听 JSON-RPC 消息。此时你不会看到任何输出,这是正常的——stdio 模式下 Server 通过 stdin/stdout 与客户端通信,不能往 stdout 打印任何东西。


FastMCP 如何自动生成 Tool 定义

这是 FastMCP 最省心的地方。以上面的 save_snippet 为例,FastMCP 自动生成的 Tool 定义等价于:

{
  "name": "save_snippet",
  "description": "Save a code snippet with title and optional tags.",
  "inputSchema": {
    "type": "object",
    "properties": {
      "title": {
        "type": "string",
        "description": "A short descriptive title for the snippet"
      },
      "code": {
        "type": "string",
        "description": "The code content to save"
      },
      "tags": {
        "type": "string",
        "description": "Comma-separated tags for categorization (e.g. \"python,async,http\")",
        "default": ""
      }
    },
    "required": ["title", "code"]
  }
}

你写 Python 函数,FastMCP 帮你搞定协议层。类型映射关系:

Python 类型 JSON Schema 类型
str string
int integer
float number
bool boolean
list[str] array[string]
dict object

接入 Claude Code / Cursor

Server 写好了,需要让客户端知道它的存在。

Claude Desktop / Claude Code

编辑配置文件:

# macOS
code ~/Library/Application\ Support/Claude/claude_desktop_config.json

# 或 Claude Code 使用项目级配置
# .mcp.json 放在项目根目录

配置内容:

{
  "mcpServers": {
    "snippet-manager": {
      "command": "uv",
      "args": [
        "--directory",
        "/ABSOLUTE/PATH/TO/snippet-server",
        "run",
        "snippet_server.py"
      ]
    }
  }
}

command 字段填 uv 的完整路径(用 which uv 获取),--directory 指向项目目录,必须是绝对路径。

Cursor

在项目根目录创建 .cursor/mcp.json

{
  "mcpServers": {
    "snippet-manager": {
      "command": "uv",
      "args": [
        "--directory",
        "/ABSOLUTE/PATH/TO/snippet-server",
        "run",
        "snippet_server.py"
      ]
    }
  }
}

Claude Code(项目级配置)

在项目根目录创建 .mcp.json

{
  "mcpServers": {
    "snippet-manager": {
      "command": "uv",
      "args": [
        "--directory",
        "/ABSOLUTE/PATH/TO/snippet-server",
        "run",
        "snippet_server.py"
      ],
      "env": {}
    }
  }
}

配置完成后重启客户端。在 Claude Desktop 中输入消息时,工具图标旁应该能看到你的 Server 名称。


调试技巧

第一原则:绝不要往 stdout 写东西

stdio 模式下,stdout 是 JSON-RPC 通信通道。往 stdout 写任何内容都会破坏协议消息,导致客户端断连。

import sys

# 错误 - 会破坏 MCP 通信
print("Server started")

# 正确 - 写到 stderr
print("Server started", file=sys.stderr)

# 正确 - 用 logging(默认输出到 stderr)
import logging
logging.basicConfig(level=logging.DEBUG)
logging.info("Server started")

TypeScript 同理:

// 错误
console.log("Server started");

// 正确
console.error("Server started");

使用 MCP Inspector 测试

在接入客户端之前,先用 Inspector 验证 Server 是否正常工作:

npx @modelcontextprotocol/inspector uv --directory /path/to/snippet-server run snippet_server.py

Inspector 提供可视化界面,可以:

  1. 查看 tools/list 返回的所有 Tool 定义
  2. 手动调用 tools/call 验证输入输出
  3. 检查 schema 是否正确生成

详细的 Inspector 使用方法参见 MCP Inspector 调试实战

常见错误排查

现象 原因 解决
客户端看不到 Tool Server 启动失败或 schema 错误 用 Inspector 独立测试
调用后无响应 Tool 函数内部抛异常 在函数内 try/except,返回错误文本
Server 启动后立即退出 依赖缺失或语法错误 单独运行 uv run snippet_server.py 看报错
中文参数乱码 编码问题 文件读写指定 encoding="utf-8"

用 CLI 模式快速验证

# 列出所有 Tool
npx @modelcontextprotocol/inspector --cli uv run snippet_server.py --method tools/list

# 调用 save_snippet
npx @modelcontextprotocol/inspector --cli uv run snippet_server.py \
  --method tools/call \
  --tool-name save_snippet \
  --tool-arg title="hello" \
  --tool-arg code='print("hello")'

进阶:Remote Server + HTTP 传输

stdio 适合本地使用。如果你的 Server 需要被多个客户端远程访问(比如团队共享),需要使用 HTTP 传输。

三种传输方式对比

传输 场景 启动方式 客户端配置
stdio 本地单机 客户端启动进程 command + args
SSE(旧版) 远程服务 Server 自行启动 HTTP url 字段
streamable-http(推荐) 远程服务 Server 自行启动 HTTP url 字段

切换到 HTTP 传输

只需改一行启动代码:

mcp.run(transport="streamable-http", host="0.0.0.0", port=8080)

客户端配置也相应改变:

{
  "mcpServers": {
    "snippet-manager": {
      "url": "http://your-server:8080/mcp"
    }
  }
}

OAuth 认证

远程 Server 暴露在公网时,需要认证保护。MCP 支持标准 OAuth 2.1 流程:

  1. Server 声明需要认证(返回 401 + WWW-Authenticate header)
  2. Client 引导用户到授权页面
  3. 用户授权后,Client 拿到 access token
  4. 后续请求携带 Bearer token

如果你用 Cloudflare Workers / Vercel 等平台部署 MCP Server,平台通常提供内置的认证方案,不需要自己实现完整的 OAuth 流程。

大多数个人开发者和团队场景下,stdio 模式就够用了。只有在"多用户远程访问"或"Server 部署在云端"时才需要考虑 HTTP 传输和认证。


完整代码参考

# snippet_server.py
import json
from pathlib import Path
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("snippet-manager")

SNIPPETS_FILE = Path.home() / ".snippets.json"


def _load_snippets() -> list[dict]:
    if not SNIPPETS_FILE.exists():
        return []
    return json.loads(SNIPPETS_FILE.read_text(encoding="utf-8"))


def _save_snippets(snippets: list[dict]) -> None:
    SNIPPETS_FILE.write_text(
        json.dumps(snippets, ensure_ascii=False, indent=2),
        encoding="utf-8",
    )


@mcp.tool()
def save_snippet(title: str, code: str, tags: str = "") -> str:
    """Save a code snippet with title and optional tags.

    Args:
        title: A short descriptive title for the snippet
        code: The code content to save
        tags: Comma-separated tags for categorization (e.g. "python,async,http")
    """
    snippets = _load_snippets()

    for s in snippets:
        if s["title"] == title:
            s["code"] = code
            s["tags"] = [t.strip() for t in tags.split(",") if t.strip()]
            _save_snippets(snippets)
            return f"Updated snippet: {title}"

    snippets.append({
        "title": title,
        "code": code,
        "tags": [t.strip() for t in tags.split(",") if t.strip()],
    })
    _save_snippets(snippets)
    return f"Saved snippet: {title}"


@mcp.tool()
def search_snippets(query: str) -> str:
    """Search snippets by keyword in title, code, or tags.

    Args:
        query: Search keyword to match against title, code content, or tags
    """
    snippets = _load_snippets()
    query_lower = query.lower()
    results = [
        s for s in snippets
        if query_lower in s["title"].lower()
        or query_lower in s["code"].lower()
        or any(query_lower in t.lower() for t in s["tags"])
    ]

    if not results:
        return f"No snippets found matching '{query}'"

    output = []
    for s in results:
        tag_str = ", ".join(s["tags"]) if s["tags"] else "no tags"
        output.append(f"## {s['title']} [{tag_str}]\n{s['code']}")

    return f"Found {len(results)} snippet(s):\n\n" + "\n\n".join(output)


@mcp.tool()
def list_snippets(tag: str = "") -> str:
    """List all saved snippets, optionally filtered by tag.

    Args:
        tag: Optional tag to filter snippets by
    """
    snippets = _load_snippets()

    if tag:
        tag_lower = tag.lower()
        snippets = [
            s for s in snippets
            if any(tag_lower == t.lower() for t in s["tags"])
        ]

    if not snippets:
        if tag:
            return f"No snippets found with tag '{tag}'"
        return "No snippets saved yet. Use save_snippet to add one."

    lines = []
    for s in snippets:
        tag_str = ", ".join(s["tags"]) if s["tags"] else "no tags"
        lines.append(f"- {s['title']} [{tag_str}]")

    header = f"Snippets" + (f" tagged '{tag}'" if tag else "")
    return f"{header} ({len(snippets)}):\n" + "\n".join(lines)


def main():
    mcp.run(transport="stdio")


if __name__ == "__main__":
    main()

调试清单

Q: FastMCP 支持异步函数吗?

支持。@mcp.tool() 同时接受同步和异步函数。如果你的 Tool 需要发起 HTTP 请求或做 IO 操作,用 async def 即可:

@mcp.tool()
async def fetch_api_data(url: str) -> str:
    """Fetch data from a URL."""
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.text

Q: Tool 的返回值只能是字符串吗?

不是。FastMCP 支持返回字符串(自动包装为 text content)、图片(Image 对象)等。但 95% 的场景返回字符串就够了——LLM 本质上处理的就是文本。

Q: 如何在 Tool 中传递环境变量(API Key 等)?

在客户端配置的 env 字段中设置,Tool 代码中通过 os.environ 读取:

{
  "mcpServers": {
    "my-server": {
      "command": "uv",
      "args": ["run", "server.py"],
      "env": {
        "API_KEY": "sk-xxx"
      }
    }
  }
}
import os
api_key = os.environ.get("API_KEY")

Q: 多个 MCP Server 可以同时运行吗?

可以。在客户端配置的 mcpServers 中添加多个条目即可,每个 Server 独立进程、独立通信。Claude Code 和 Cursor 都支持同时连接多个 Server。

Q: 本地开发时如何热重载?

stdio 模式下没有原生热重载。开发时可以用 uvicornwatchfiles 配合 HTTP 传输模式;或者每次改代码后重启客户端(Claude Code 用 /mcp 命令可重新连接)。