MCP Server 开发实战：构建你自己的 MCP 服务

引言

上一篇文章我们介绍了 MCP 协议的设计理念和架构。理论终究要落地——本文将带你从零开始构建两个真实可用的 MCP Server，分别连接 SQLite 数据库和天气 API，让你彻底掌握 MCP Server 的开发全流程。

环境准备

MCP 官方提供了 Python 和 TypeScript 两种 SDK。本文以 Python 为例。

# 创建项目目录
mkdir my-mcp-server && cd my-mcp-server

# 安装 MCP SDK
pip install mcp

# 推荐的辅助工具：MCP Inspector（调试必备）
npx @anthropic-ai/mcp-inspector

Python SDK 的核心模块包括：

• mcp.server：Server 端框架，提供 Server 类和装饰器。
• mcp.types：类型定义，包括 Tool、TextContent、Resource 等。
• mcp.server.stdio：stdio 传输层实现。

定义工具：三个核心要素

MCP 中的工具定义由三个核心要素构成：

1. 名称（name）

工具的唯一标识符，AI 模型通过名称来指定调用哪个工具。命名建议：使用 snake_case，动词开头，清晰表达功能。例如 query_database 优于 db，get_weather 优于 w。

2. 描述（description）

工具的用途说明，直接影响 AI 模型能否正确选择和使用工具。描述应当清晰、准确，包含使用场景提示。这是 AI 判断「何时调用此工具」的核心依据。

3. 输入模式（inputSchema）

使用 JSON Schema 定义工具接受的参数。包括参数名称、类型、是否必填、默认值、枚举约束等。

from mcp.types import Tool

tool_def = Tool(
    name="query_database",
    description="执行 SQL 查询并返回结果。仅支持 SELECT 查询，不支持修改操作。",
    inputSchema={
        "type": "object",
        "properties": {
            "sql": {
                "type": "string",
                "description": "要执行的 SQL 查询语句，仅支持 SELECT"
            }
        },
        "required": ["sql"]
    }
)

实战一：SQLite 数据库查询 MCP Server

这个 Server 将为 AI 模型提供数据库查询能力，让模型能够通过自然语言交互来查询数据。

完整的 Server 实现

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

# 创建 Server 实例
server = Server("sqlite-explorer")

# 初始化数据库连接
db = sqlite3.connect("data.db")
db.row_factory = sqlite3.Row

@server.list_tools()
async def list_tools() -> list[Tool]:
    """暴露可用工具列表"""
    return [
        Tool(
            name="query_database",
            description="在 SQLite 数据库中执行 SELECT 查询，返回结果集。",
            inputSchema={
                "type": "object",
                "properties": {
                    "sql": {
                        "type": "string",
                        "description": "要执行的 SELECT 查询语句"
                    }
                },
                "required": ["sql"]
            }
        ),
        Tool(
            name="list_tables",
            description="列出数据库中所有的表及其结构信息。",
            inputSchema={
                "type": "object",
                "properties": {}
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    """处理工具调用请求"""
    if name == "query_database":
        sql = arguments["sql"]
        # 安全检查：仅允许 SELECT 语句
        if not sql.strip().upper().startswith("SELECT"):
            return [TextContent(
                type="text",
                text="错误：仅允许执行 SELECT 查询。"
            )]

        try:
            cursor = db.execute(sql)
            rows = cursor.fetchall()
            # 格式化结果
            if not rows:
                return [TextContent(type="text", text="查询结果为空。")]

            columns = rows[0].keys()
            lines = [",".join(columns)]
            for row in rows:
                lines.append(",".join(str(v) for v in row))
            return [TextContent(
                type="text",
                text=f"查询结果（{len(rows)} 行）：\n" + "\n".join(lines[:50])
            )]
        except Exception as e:
            return [TextContent(
                type="text",
                text=f"查询失败：{str(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" + "\n".join(tables)
        )]

    else:
        return [TextContent(
            type="text",
            text=f"未知工具：{name}"
        )]

# 启动 Server
async def main():
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream,
                         server.create_initialization_options())

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

关键设计要点

安全控制：在 query_database 处理函数中，我们显式检查 SQL 语句是否以 SELECT 开头，防止 AI 模型被诱导执行 DROP、DELETE 等危险操作。这是 MCP Server 安全设计的第一道防线。

结果截断：通过 lines[:50] 限制返回行数，防止海量数据撑爆 AI 的上下文窗口。

友好的错误信息：捕获所有异常并返回清晰的错误描述，避免将堆栈信息直接暴露给 AI 模型。

实战二：天气 API MCP Server

接下来构建一个对接外部天气服务的 MCP Server，演示如何将第三方 API 封装为 AI 可调用的工具。

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

server = Server("weather-service")

# 混元天气 API 配置（示例）
API_KEY = "your_api_key_here"
BASE_URL = "https://api.weather.com/v1"

@server.list_tools()
async def list_tools() -> list[Tool]:
    return [
        Tool(
            name="get_current_weather",
            description="获取指定城市的实时天气信息，包括温度、湿度、风速和天气状况。",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称，例如 '北京'、'上海'"
                    }
                },
                "required": ["city"]
            }
        ),
        Tool(
            name="get_forecast",
            description="获取指定城市未来7天的天气预报。",
            inputSchema={
                "type": "object",
                "properties": {
                    "city": {
                        "type": "string",
                        "description": "城市名称"
                    },
                    "days": {
                        "type": "integer",
                        "description": "预报天数，1-7",
                        "default": 3,
                        "minimum": 1,
                        "maximum": 7
                    }
                },
                "required": ["city"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
    async with httpx.AsyncClient(timeout=10.0) as client:
        if name == "get_current_weather":
            city = arguments["city"]
            response = await client.get(
                f"{BASE_URL}/current",
                params={"city": city, "key": API_KEY}
            )
            response.raise_for_status()
            data = response.json()
            return [TextContent(
                type="text",
                text=f"{city} 当前天气：\n"
                     f"温度：{data['temp']}°C\n"
                     f"湿度：{data['humidity']}%\n"
                     f"风速：{data['wind_speed']} m/s\n"
                     f"天气：{data['condition']}"
            )]

        elif name == "get_forecast":
            city = arguments["city"]
            days = arguments.get("days", 3)
            response = await client.get(
                f"{BASE_URL}/forecast",
                params={"city": city, "key": API_KEY, "days": days}
            )
            response.raise_for_status()
            data = response.json()
            forecast_lines = [
                f"{d['date']}: {d['condition']}, "
                f"{d['temp_low']}°C ~ {d['temp_high']}°C"
                for d in data["forecasts"]
            ]
            return [TextContent(
                type="text",
                text=f"{city} 未来{days}天预报：\n" + "\n".join(forecast_lines)
            )]

# 启动逻辑同示例一

API 集成注意事项

1. 超时控制：使用 httpx.AsyncClient(timeout=10.0) 设置超时，防止外部服务无响应导致 AI 调用卡死。
2. 错误传播：通过 raise_for_status() 将 HTTP 错误转化为异常，由异常处理逻辑统一返回友好提示。
3. 速率限制：生产环境中应实现令牌桶或滑动窗口限流，防止 AI 频繁调用导致 API 配额耗尽。

调试利器：MCP Inspector

MCP Inspector 是官方提供的调试工具，可以独立测试你的 MCP Server：

npx @anthropic-ai/mcp-inspector python my_server.py

运行后会启动一个 Web 界面，你可以在其中：

• 查看 Server 暴露的所有工具和资源
• 手动构造参数并调用工具
• 查看完整的 JSON-RPC 请求和响应
• 模拟 AI 模型的调用流程

这是开发阶段不可或缺的调试手段。

安全设计清单

在将 MCP Server 投入生产前，请逐项检查以下安全措施：

1. 输入验证：对所有工具参数做严格的类型检查和格式校验，使用 JSON Schema 的 enum、pattern、minimum、maximum 等约束。
2. 访问控制：根据 MCP Client 的身份限制可调用的工具范围。在生产部署中，考虑集成 OAuth 或 API Key 认证。
3. 速率限制：为每个工具设置合理的调用频率上限，防止滥用。
4. 操作审计：记录所有工具调用的详细日志（调用者、工具名、参数、结果、时间戳）。
5. 敏感信息脱敏：工具返回结果中避免包含密码、Token 等敏感信息。

发布与分享

MCP Server 开发完成后，可以通过以下方式分享：

1. 发布到 PyPI / npm：如果你的 Server 有通用价值，打包发布到包管理器。
2. 上传到 MCP Registry：社区维护的 MCP Server 目录，方便其他开发者发现。
3. 提交到 Claude Desktop 配置：在你的 claude_desktop_config.json 中添加 Server 配置，直接使用。

{
  "mcpServers": {
    "my-sqlite": {
      "command": "python",
      "args": ["path/to/my_server.py"]
    }
  }
}

总结

构建一个 MCP Server 的核心工作流可以总结为：定义工具 Schema → 实现处理函数 → 处理异常 → 安全加固 → 测试调试 → 部署发布。掌握这个流程后，你可以将任何 API、数据库、文件系统封装为 AI 可用的标准化工具，让你的 AI 应用具备前所未有的能力边界。

在下一篇文章中，我们将探讨如何将多个 MCP Server 与企业级 AI 框架（如 LangChain）集成，构建生产级 AI 应用。