MCP 2026-05-14

MCP Server 开发实战：从零构建生产级工具服务

手把手教你使用 TypeScript 和 Python 构建高质量的 MCP Server，涵盖 Tools、Resources、Prompts 三大能力的完整实现。

MCP Server 是 Model Context Protocol 生态中的核心组件——它将外部服务的能力暴露给 LLM，让 AI Agent 能够与真实世界交互。这篇文章将带你从环境搭建到生产部署，完整实现一个功能丰富的 MCP Server。

开发环境准备

在开始之前，确保你的开发环境满足以下要求：

Node.js 18+（TypeScript 版本）或 Python 3.10+（Python 版本）
一个支持 MCP 的客户端（如 Claude Desktop）

TypeScript 项目初始化

mkdir my-mcp-server && cd my-mcp-server
npm init -y
npm install @modelcontextprotocol/sdk zod
npm install -D typescript @types/node
npx tsc --init

tsconfig.json 需要配置 ESM 模块：

{
  "compilerOptions": {
    "target": "ES2022",
    "module": "Node16",
    "moduleResolution": "Node16",
    "outDir": "./dist",
    "strict": true,
    "esModuleInterop": true
  }
}

Python 项目初始化

mkdir my-mcp-server && cd my-mcp-server
python -m venv .venv
source .venv/bin/activate
pip install mcp

理解 MCP Server 的核心概念

在编写代码之前，我们需要理解 MCP Server 的三种核心能力各自的设计意图和使用场景。

Tool：让 LLM 执行操作

Tool 是 MCP Server 最重要的能力。它代表一个可以被 LLM 调用的操作，类似于 API 的 endpoint。每个 Tool 需要定义：

name：唯一标识符，使用 snake_case
description：清晰的功能描述，LLM 根据这个描述决定何时调用
inputSchema：JSON Schema 格式的参数规范

Tool 的描述质量直接影响 LLM 的调用准确率。好的描述应该回答三个问题：这个工具做什么？什么时候应该用？参数的含义是什么？

Resource：让 LLM 读取数据

Resource 代表只读的数据源。与 Tool 不同，Resource 不执行任何操作，只是返回数据。Resource 通过 URI 标识，支持静态和动态两种模式。

Prompt：引导 LLM 正确使用

Prompt 是预定义的提示模板，帮助 LLM 理解如何组合使用 Server 提供的工具和资源。

实战：构建数据库查询 Server

让我们构建一个实用的 MCP Server——SQLite 数据库查询工具。这个 Server 将允许 AI Agent 查询数据库、查看表结构和执行分析。

TypeScript 实现

import { McpServer } from '@modelcontextprotocol/sdk/server/mcp.js';
import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js';
import { z } from 'zod';
import Database from 'better-sqlite3';

const db = new Database(process.argv[2] || 'data.db');
const server = new McpServer({
  name: 'sqlite-query',
  version: '1.0.0',
});

// 查询工具
server.tool(
  'query',
  '执行 SQL 查询并返回结果。仅支持 SELECT 语句。',
  {
    sql: z.string().describe('要执行的 SQL SELECT 语句'),
  },
  async ({ sql }) => {
    // 安全校验：只允许 SELECT
    const trimmed = sql.trim().toUpperCase();
    if (!trimmed.startsWith('SELECT')) {
      return {
        content: [{ type: 'text', text: '错误：只允许执行 SELECT 查询' }],
        isError: true,
      };
    }

    try {
      const rows = db.prepare(sql).all();
      return {
        content: [{
          type: 'text',
          text: JSON.stringify(rows, null, 2),
        }],
      };
    } catch (error) {
      return {
        content: [{ type: 'text', text: `查询错误：${error.message}` }],
        isError: true,
      };
    }
  }
);

// 表结构查看
server.tool(
  'list_tables',
  '列出数据库中的所有表',
  {},
  async () => {
    const tables = db.prepare(
      "SELECT name FROM sqlite_master WHERE type='table'"
    ).all();
    return {
      content: [{
        type: 'text',
        text: tables.map((t: any) => t.name).join('\n'),
      }],
    };
  }
);

// 表结构详情
server.tool(
  'describe_table',
  '查看指定表的列定义和数据类型',
  { tableName: z.string().describe('表名') },
  async ({ tableName }) => {
    // 防止 SQL 注入
    if (!/^[a-zA-Z_][a-zA-Z0-9_]*$/.test(tableName)) {
      return {
        content: [{ type: 'text', text: '无效的表名' }],
        isError: true,
      };
    }

    const columns = db.prepare(`PRAGMA table_info(${tableName})`).all();
    return {
      content: [{
        type: 'text',
        text: JSON.stringify(columns, null, 2),
      }],
    };
  }
);

// 资源：数据库元信息
server.resource(
  'db-info',
  'sqlite://info',
  async (uri) => {
    const tables = db.prepare(
      "SELECT name FROM sqlite_master WHERE type='table'"
    ).all();
    const info = tables.map((t: any) => {
      const count = db.prepare(`SELECT COUNT(*) as cnt FROM ${t.name}`).get();
      return `${t.name}: ${count.cnt} 行`;
    });

    return {
      contents: [{
        uri: uri.href,
        text: `数据库表统计：\n${info.join('\n')}`,
      }],
    };
  }
);

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

Python 实现

import sqlite3
import sys
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import Tool, TextContent

db_path = sys.argv[1] if len(sys.argv) > 1 else "data.db"
db = sqlite3.connect(db_path)
server = Server("sqlite-query")

@server.list_tools()
async def list_tools():
    return [
        Tool(
            name="query",
            description="执行 SQL SELECT 查询",
            inputSchema={
                "type": "object",
                "properties": {
                    "sql": {
                        "type": "string",
                        "description": "SQL SELECT 语句"
                    }
                },
                "required": ["sql"]
            }
        ),
        Tool(
            name="list_tables",
            description="列出所有数据库表",
            inputSchema={"type": "object", "properties": {}}
        ),
    ]

@server.call_tool()
async def call_tool(name, arguments):
    if name == "query":
        sql = arguments["sql"]
        if not sql.strip().upper().startswith("SELECT"):
            return [TextContent(type="text", text="仅支持 SELECT 查询")]
        try:
            cursor = db.execute(sql)
            rows = cursor.fetchall()
            columns = [desc[0] for desc in cursor.description]
            result = [dict(zip(columns, row)) for row in rows]
            return [TextContent(type="text", text=str(result))]
        except Exception as e:
            return [TextContent(type="text", text=f"错误：{e}")]

    elif name == "list_tables":
        cursor = db.execute(
            "SELECT name FROM sqlite_master WHERE type='table'"
        )
        tables = [row[0] for row in cursor.fetchall()]
        return [TextContent(type="text", text="\n".join(tables))]

async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream)

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

Tool 设计的最佳实践

经过大量实践，我总结了以下 Tool 设计原则：

描述要精确

模糊的描述会导致 LLM 错误调用。比较以下两个描述：

# 差
description: "查询数据"

# 好
description: "执行 SQL SELECT 查询语句，返回查询结果。仅支持读取操作，不支持 INSERT/UPDATE/DELETE。"

参数校验要严格

使用 Zod（TypeScript）或 Pydantic（Python）进行严格的参数校验。这不仅能防止错误输入，还能帮助 LLM 理解参数的约束条件。

server.tool(
  'search',
  '在指定目录中搜索文件',
  {
    directory: z.string().describe('搜索目录的绝对路径'),
    pattern: z.string().describe('文件名匹配模式，支持通配符 * 和 ?'),
    maxResults: z.number().int().min(1).max(100).default(10)
      .describe('最大返回数量，默认 10'),
  },
  handler
);

错误信息要有用

当工具执行失败时，返回清晰的错误信息，帮助 LLM 理解问题所在并做出调整。

catch (error) {
  return {
    content: [{
      type: 'text',
      text: `文件不存在：${filePath}\n请检查路径是否正确，或使用 list_files 工具查看可用文件。`,
    }],
    isError: true,
  };
}

注册到 Claude Desktop

开发完成后，需要在 Claude Desktop 的配置文件中注册你的 MCP Server。

编辑 ~/Library/Application Support/Claude/claude_desktop_config.json（macOS）或 %APPDATA%\Claude\claude_desktop_config.json（Windows）：

{
  "mcpServers": {
    "sqlite": {
      "command": "node",
      "args": ["/path/to/dist/index.js", "/path/to/database.db"]
    }
  }
}

重启 Claude Desktop 后，你就可以在对话中使用这些工具了。

常见问题（FAQ）

MCP Server 如何处理并发请求？

stdio 传输模式下，Server 按顺序处理请求。如果需要并发支持，可以使用 HTTP+SSE 传输模式，配合 Node.js 的异步处理能力实现并发。

工具调用的超时时间如何设置？

MCP 协议本身不定义超时时间，由 Client 端控制。建议在 Server 端也实现自己的超时机制，避免长时间运行的查询阻塞整个服务。

如何测试 MCP Server？

可以使用 MCP Inspector 工具进行交互式测试。运行 npx @modelcontextprotocol/inspector node dist/index.js 即可启动一个 Web 界面来测试你的 Server。

总结

构建一个高质量的 MCP Server 需要关注三个核心要素：清晰的 Tool 描述、严格的参数校验和有用的错误信息。遵循这些原则，你的 Server 就能被 LLM 准确、高效地使用。

下一步，我们将探讨如何构建 MCP Client，让你的 AI Agent 能够同时管理多个 MCP Server 的连接。