乐闻世界logo
5月28日 06:54

如何实现一个 MCP 服务器？有哪些最佳实践？

MCP 服务器的核心概念

MCP（Model Context Protocol）是一个开放协议，让大语言模型能够通过标准化接口连接外部数据源和工具。理解服务器实现之前，需要先厘清三个核心概念：

  • Tool（工具）：服务器暴露给客户端的可调用函数，接受结构化参数并返回结果
  • Resource（资源）：服务器提供的可读数据源，客户端可以按 URI 访问
  • Prompt（提示模板）：服务器预设的提示词模板，客户端可以复用

服务器与客户端之间通过 Capability 协商 建立连接——客户端在初始化时声明自己支持哪些能力，服务器也声明自己提供哪些能力，双方只使用共同支持的功能。

选择传输方式

MCP 支持两种传输协议，选择哪种直接影响部署架构：

Stdio 传输：客户端以子进程方式启动服务器，通过标准输入/输出通信。适合本地工具集成，延迟低，无需处理网络和认证问题。Claude Desktop 和 Cursor 的本地服务器都走这条路径。

Streamable HTTP 传输：服务器以独立 HTTP 服务运行，客户端通过 POST 请求通信。适合远程部署、多客户端共享、需要认证的场景。注意，旧版基于 SSE 的 HTTP 传输已被废弃，新项目应直接使用 Streamable HTTP。

python
# Stdio 模式启动
mcp.run(transport="stdio")

# HTTP 模式启动
mcp.run(transport="streamable-http", host="0.0.0.0", port=8080)

用 Python 实现一个 MCP 服务器

Python SDK（mcp 包）提供了 FastMCP 高阶 API，可以快速搭建服务器。

安装依赖

bash
pip install "mcp[cli]"

定义工具

工具是 MCP 服务器的核心能力。每个工具需要声明名称、描述和参数 Schema：

python
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("my-server")

@mcp.tool()
def search_docs(query: str, limit: int = 5) -> str:
    """在文档库中搜索相关内容，返回匹配结果摘要"""
    # 实际场景中对接搜索引擎或向量数据库
    results = do_search(query, limit)
    return format_results(results)

关键要点：工具函数的 docstring 会成为客户端看到的工具描述，直接影响模型能否正确调用，务必写得准确具体。参数类型注解会自动转换为 JSON Schema。

定义资源

资源用于暴露可读数据，客户端通过 URI 主动拉取：

python
@mcp.resource("config://app/settings")
def get_settings() -> str:
    """返回当前应用配置"""
    return json.dumps(load_settings(), ensure_ascii=False)

@mcp.resource("data://users/{user_id}/profile")
def get_user_profile(user_id: str) -> str:
    """按用户 ID 返回用户资料"""
    profile = fetch_profile(user_id)
    return json.dumps(profile, ensure_ascii=False)

URI 模板支持路径参数（如 {user_id}），SDK 会自动解析并传入函数。

定义提示模板

提示模板是预设的 Prompt，客户端可以按名称获取：

python
@mcp.prompt()
def code_review(code: str) -> str:
    """代码审查提示模板"""
    return f"请审查以下代码，关注安全性和性能问题：\n\n```\n{code}\n```"

启动服务器

python
if __name__ == "__main__":
    mcp.run()

mcp run 命令默认使用 stdio 传输。也可通过 CLI 指定：

bash
mcp run server.py --transport streamable-http

用 TypeScript 实现一个 MCP 服务器

TypeScript SDK 是官方维护的首选 SDK，生态最完整：

typescript
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
import { z } from "zod";

const server = new McpServer({
  name: "my-server",
  version: "1.0.0",
});

server.tool(
  "search_docs",
  "在文档库中搜索相关内容",
{ query: z.string(), limit: z.number().default(5) },
async ({ query, limit }) => {
    const results = await doSearch(query, limit);
    return { content: [{ type: "text", text: formatResults(results) }] };
  }
);

const transport = new StdioServerTransport();
await server.connect(transport);

TypeScript SDK 使用 zod 定义参数 Schema，类型安全且自动生成 JSON Schema。

生产环境的最佳实践

错误处理

不要让异常穿透到客户端变成不友好的报错。在每个工具函数中做防御性处理：

python
@mcp.tool()
def query_database(sql: str) -> str:
    """执行只读 SQL 查询"""
    try:
        if not sql.strip().upper().startswith("SELECT"):
            return "错误：仅允许 SELECT 查询"
        result = execute_readonly_query(sql)
        return json.dumps(result, ensure_ascii=False)
    except ConnectionError:
        return "错误：数据库连接失败，请稍后重试"
    except Exception as e:
        return f"查询执行出错：{type(e).__name__}"

注意上面捕获了具体异常类型，对用户屏蔽了内部实现细节。

输入验证

永远不要信任客户端传入的参数，即使模型通常不会构造恶意输入：

python
@mcp.tool()
def run_command(cmd: str) -> str:
    """执行预定义的系统命令（仅允许白名单命令）"""
    # 不要直接拼接执行用户输入！
    allowed = {"list_files", "check_status", "get_version"}
    if cmd not in allowed:
        return f"错误：不支持的命令，允许的命令：{', '.join(allowed)}"
    return execute_safe(cmd)

认证与安全

远程部署的 MCP 服务器必须实现认证。2026 协议推荐使用 OAuth 2.1：

  • 为每个客户端颁发独立的 access token，绑定最小权限 scope
  • 工具级别的权限控制：不同 scope 对应不同的工具访问权限
  • 传输层强制 HTTPS，禁止明文 HTTP
  • 本地 stdio 服务器依赖操作系统用户隔离，不需要额外认证

异步与性能

工具函数如果涉及 I/O 操作（网络请求、文件读写、数据库查询），应该使用异步实现：

python
@mcp.tool()
async def fetch_url(url: str) -> str:
    """获取 URL 内容"""
    async with httpx.AsyncClient(timeout=10) as client:
        resp = await client.get(url)
        return resp.text[:5000]  # 限制返回长度

其他性能策略：

  • 对频繁调用的工具实现结果缓存，避免重复计算
  • 为耗时操作设置超时，防止客户端无限等待
  • 大量数据分页返回，不要一次吐出全部内容

日志与监控

生产服务器必须有可观测性：

python
import logging

logger = logging.getLogger("mcp-server")

@mcp.tool()
def search_docs(query: str) -> str:
    logger.info(f"search_docs called: query={query!r}")
    result = do_search(query)
    logger.info(f"search_docs returned {len(result)} results")
    return result

建议记录：每次工具调用的入参摘要、执行耗时、异常信息。远程服务器还应实现健康检查端点供负载均衡器探活。

常见陷阱

在工具中执行任意代码：绝不能用 eval()、exec() 或直接拼接 shell 命令。这是最严重的安全漏洞，恶意输入可以完全控制服务器。

忽略 Capability 协商：服务器声明的能力必须与实际实现一致。声明了 tools capability 却没有注册任何工具，或者工具描述与实际行为不符，都会导致客户端调用失败。

过度暴露工具：一个服务器注册几十个工具会占用客户端大量上下文窗口。按功能域拆分成多个独立服务器，每个只暴露相关工具，客户端按需连接。

同步阻塞：在异步框架中使用同步的阻塞 I/O 会拖慢整个服务器。使用 asyncio.to_thread 将同步调用包装为异步，或直接使用异步库。

面试追问

Q：MCP 的 Stdio 传输和 HTTP 传输分别适合什么场景？ Stdio 适合本地单客户端场景，客户端以子进程方式启动服务器，零配置、低延迟、无需认证。HTTP 适合远程部署、多客户端共享、需要认证和网关管理的生产环境。2026 协议推荐新项目使用 Streamable HTTP 替代已废弃的 SSE 传输。

Q：如何在 MCP 服务器中防止工具被滥用？ 三层防线：输入验证拒绝非法参数，OAuth scope 控制工具访问权限，速率限制防止高频调用。核心原则是永远不信任客户端输入，白名单优于黑名单，最小权限优于便利性。

Q：一个 MCP 服务器应该注册多少个工具？ 建议控制在 10 个以内。工具过多会占用客户端上下文窗口，影响模型选择准确性。按功能域拆分多个服务器，客户端按需连接，比一个臃肿服务器更易维护。

标签：MCP