Python 小白也能学会:手把手写一个连接 MySQL 的 MCP 服务端

目标：用 Python 写一个 MCP 服务端，连接 MySQL 查询用户信息，并配置到 OpenClaw

一、MCP 是什么

MCP（Model Context Protocol） 是 Anthropic 推出的开放协议，让 AI 助手（如 Claude）能够安全地调用外部工具和数据源。

打个比方：

MCP 服务端可以暴露三种能力：

1. Tools（工具）：AI 可调用的函数（如查询用户）
2. Resources（资源）：只读数据（如文件内容）
3. Prompts（提示词模板）：预定义的提示词

本文重点讲 Tools。

二、环境准备

2.1 系统要求

• Python 3.10 或更高版本
• MySQL 数据库（已安装并运行）
• pip 包管理器

2.2 安装依赖

# 创建项目目录mkdir mysql-mcp-servercd mysql-mcp-server# 创建虚拟环境python -m venv .venv# 激活虚拟环境# macOS/Linux:source .venv/bin/activate# Windows:.venv\Scripts\activate# 安装 MCP SDK 和 MySQL 驱动pip install "mcp[cli]" mysql-connector-python

注意：mcp[cli] 需要 1.2.0 或更高版本。

三、编写 MCP 服务端

3.1 项目结构

mysql-mcp-server/├── .venv/├── server.py          # 主服务端代码├── config.json        # 数据库配置（敏感信息，勿提交）└── requirements.txt   # 依赖列表

3.2 数据库配置文件

创建 config.json，存放数据库连接信息：

{"host": "localhost","port": 3306,"user": "your_username","password": "your_password","database": "your_database"}

⚠️ 安全提醒：此文件包含敏感信息，务必加入 .gitignore，不要提交到代码仓库。

3.3 核心服务端代码

创建 server.py：

"""MySQL MCP Server - 查询用户基本信息"""import jsonimport sysimport loggingfrom typing import Anyimport mysql.connectorfrom mcp.server.fastmcp import FastMCP# 配置日志（写入 stderr，避免干扰 stdio 通信）logging.basicConfig(    level=logging.INFO,format="%(asctime)s - %(levelname)s - %(message)s",    stream=sys.stderr)logger = logging.getLogger(__name__)# 初始化 FastMCP 服务端mcp = FastMCP("mysql-user-query")# 数据库配置文件路径CONFIG_FILE = "config.json"def load_config() -> dict:"""加载数据库配置"""try:with open(CONFIG_FILE, "r", encoding="utf-8") as f:return json.load(f)except FileNotFoundError:        logger.error(f"配置文件 {CONFIG_FILE} 不存在")raiseexcept json.JSONDecodeError as e:        logger.error(f"配置文件 JSON 格式错误: {e}")raisedef get_connection():"""获取数据库连接"""    config = load_config()return mysql.connector.connect(        host=config["host"],        port=config.get("port", 3306),        user=config["user"],        password=config["password"],        database=config["database"]    )def row_to_dict(cursor, row) -> dict:"""将数据库行转换为字典"""return {desc[0]: value for desc, value in zip(cursor.description, row)}# ============ 定义 MCP Tools ============@mcp.tool()async def query_user_by_id(user_id: int) -> str:"""根据用户 ID 查询用户基本信息。    Args:        user_id: 用户 ID（整数）    Returns:        用户信息的 JSON 字符串，如果未找到则返回提示信息    """try:        conn = get_connection()        cursor = conn.cursor(dictionary=True)        cursor.execute("SELECT id, username, email, created_at, status FROM users WHERE id = %s",            (user_id,)        )        row = cursor.fetchone()        cursor.close()        conn.close()if row:# 处理 datetime 等不可 JSON 序列化的类型            result = {}for key, value in row.items():if hasattr(value, 'isoformat'):                    result[key] = value.isoformat()else:                    result[key] = valuereturn json.dumps(result, ensure_ascii=False, indent=2)else:return f"未找到 ID 为 {user_id} 的用户"except mysql.connector.Error as e:        logger.error(f"数据库错误: {e}")return f"数据库查询失败: {str(e)}"except Exception as e:        logger.error(f"未知错误: {e}")return f"查询失败: {str(e)}"@mcp.tool()async def search_users_by_name(name: str) -> str:"""根据用户名模糊搜索用户。    Args:        name: 用户名关键词（支持模糊匹配）    Returns:        匹配用户列表的 JSON 字符串    """try:        conn = get_connection()        cursor = conn.cursor(dictionary=True)        cursor.execute("SELECT id, username, email, status FROM users WHERE username LIKE %s LIMIT 20",            (f"%{name}%",)        )        rows = cursor.fetchall()        cursor.close()        conn.close()if rows:            result = []for row in rows:                item = {}for key, value in row.items():if hasattr(value, 'isoformat'):                        item[key] = value.isoformat()else:                        item[key] = value                result.append(item)return json.dumps(result, ensure_ascii=False, indent=2)else:return f"未找到用户名包含 '{name}' 的用户"except mysql.connector.Error as e:        logger.error(f"数据库错误: {e}")return f"数据库查询失败: {str(e)}"except Exception as e:        logger.error(f"未知错误: {e}")return f"查询失败: {str(e)}"@mcp.tool()async def list_users(page: int = 1, page_size: int = 10) -> str:"""分页查询用户列表。    Args:        page: 页码，从 1 开始（默认 1）        page_size: 每页数量（默认 10，最大 100）    Returns:        用户列表和分页信息的 JSON 字符串    """# 参数校验    page = max(1, page)    page_size = min(max(1, page_size), 100)    offset = (page - 1) * page_sizetry:        conn = get_connection()        cursor = conn.cursor(dictionary=True)# 查询总数        cursor.execute("SELECT COUNT(*) as total FROM users")        total = cursor.fetchone()["total"]# 分页查询        cursor.execute("SELECT id, username, email, status FROM users ORDER BY id LIMIT %s OFFSET %s",            (page_size, offset)        )        rows = cursor.fetchall()        cursor.close()        conn.close()        result = {"page": page,"page_size": page_size,"total": total,"total_pages": (total + page_size - 1) // page_size,"users": []        }for row in rows:            item = {}for key, value in row.items():if hasattr(value, 'isoformat'):                    item[key] = value.isoformat()else:                    item[key] = value            result["users"].append(item)return json.dumps(result, ensure_ascii=False, indent=2)except mysql.connector.Error as e:        logger.error(f"数据库错误: {e}")return f"数据库查询失败: {str(e)}"except Exception as e:        logger.error(f"未知错误: {e}")return f"查询失败: {str(e)}"# ============ 启动服务端 ============if __name__ == "__main__":# 检查配置文件try:        load_config()        logger.info("数据库配置加载成功")except Exception as e:        logger.error(f"无法加载配置: {e}")        sys.exit(1)# 启动 MCP 服务端（stdio 模式）    logger.info("启动 MySQL MCP 服务端...")    mcp.run()

3.4 创建依赖文件

创建 requirements.txt：

mcp[cli]>=1.2.0mysql-connector-python>=8.0.0

四、本地测试

4.1 使用 MCP Inspector 测试

MCP SDK 提供了一个调试工具 mcp dev，可以在浏览器中可视化测试：

# 在项目目录下运行mcp dev server.py

浏览器会自动打开调试界面，你可以：

1. 查看服务端暴露的所有 Tools
2. 输入参数测试每个 Tool
3. 查看返回结果

4.2 命令行测试

也可以直接运行服务端，通过 stdio 交互：

python server.py

服务端启动后，会通过标准输入输出（stdio）等待 JSON-RPC 请求。

五、配置到 OpenClaw

5.1 OpenClaw MCP 配置格式

OpenClaw 支持 stdio 和 HTTP 两种 MCP 传输方式。

方式一：Stdio 模式（推荐）

Stdio 模式下，OpenClaw 会启动一个子进程运行你的 MCP 服务端：

{"mcp": {"servers": {"mysql-user-query": {"command": "python","args": ["/绝对路径/mysql-mcp-server/server.py"],"env": {"PYTHONUNBUFFERED": "1"}}}}}

字段说明：

⚠️ 注意：args 中的路径必须是绝对路径。

方式二：HTTP 模式

如果你将 MCP 服务端部署为 HTTP 服务，可以使用 HTTP 模式：

{"mcp": {"servers": {"mysql-user-query": {"url": "http://localhost:3100/mcp","transport": "streamable-http","headers": {"Authorization": "Bearer ${MY_SECRET_TOKEN}"},"connectionTimeoutMs": 30000}}}}

字段说明：

5.2 配置文件位置

OpenClaw 的 MCP 配置可以放在多个位置：

位置一：全局配置

编辑 ~/.openclaw/config.json（或 OpenClaw 配置文件路径）：

{"mcp": {"servers": {"mysql-user-query": {"command": "python","args": ["/home/user/mysql-mcp-server/server.py"],"env": {"PYTHONUNBUFFERED": "1"}}}}}

位置二：项目级配置

在项目根目录创建 .openclaw/pi.json：

{"mcpServers": {"mysql-user-query": {"command": "python","args": ["/home/user/mysql-mcp-server/server.py"]}}}

位置三：通过 Bundle 配置

如果你创建了一个 OpenClaw 插件包，可以在 openclaw.plugin.json 中配置 MCP：

{"id": "mysql-mcp-bundle","name": "MySQL MCP Bundle","description": "MySQL 用户查询 MCP 服务","mcp": {"servers": {"mysql-user-query": {"command": "python","args": ["${PLUGIN_DIR}/server.py"]}}}}

5.3 重启 OpenClaw Gateway

配置完成后，重启 OpenClaw Gateway 使配置生效：

openclaw gateway restart

5.4 验证配置

检查 MCP 服务端是否正常加载：

# 查看 Gateway 日志openclaw gateway logs --tail 50# 或在 OpenClaw 会话中询问 AI# 例如："你能查询数据库中 ID 为 1 的用户信息吗？"

六、完整的 OpenClaw 配置示例

假设你的 MySQL MCP 服务端位于 /home/yourname/mysql-mcp-server/，完整的 OpenClaw 配置文件示例如下：

config.json（OpenClaw 主配置）

{"gateway": {"port": 18789},"mcp": {"servers": {"mysql-user-query": {"command": "/home/yourname/mysql-mcp-server/.venv/bin/python","args": ["/home/yourname/mysql-mcp-server/server.py"],"env": {"PYTHONUNBUFFERED": "1","PYTHONIOENCODING": "utf-8"}}}}}

💡 最佳实践：command 使用虚拟环境中的 Python 解释器路径，确保依赖一致。

七、常见问题排查

问题 1：服务端启动失败

症状：OpenClaw 日志显示 MCP 服务端连接失败

排查步骤：

# 1. 手动运行服务端，检查是否有报错cd /home/yourname/mysql-mcp-serversource .venv/bin/activatepython server.py# 2. 检查配置文件路径是否正确cat config.json# 3. 检查数据库连接python -c "import mysql.connector; conn = mysql.connector.connect(host='localhost', user='your_user', password='your_pass', database='your_db'); print('OK')"

问题 2：日志输出干扰 stdio

症状：MCP 调用返回乱码或解析失败

原因：print() 输出到 stdout 会干扰 JSON-RPC 通信

解决方案：

# ❌ 错误print("正在处理请求...")# ✅ 正确print("正在处理请求...", file=sys.stderr)# ✅ 或使用 loggingimport logginglogging.info("正在处理请求...")

问题 3：中文字符乱码

解决方案：

# 在服务端代码中设置编码import syssys.stdout.reconfigure(encoding='utf-8')# 或在配置中添加环境变量{"env": {"PYTHONIOENCODING": "utf-8"  }}

八、总结

本文从零开始，手把手教你：

1. ✅ 理解 MCP 协议的核心概念
2. ✅ 用 Python FastMCP SDK 编写服务端
3. ✅ 实现三个实用的 MySQL 查询工具
4. ✅ 配置到 OpenClaw（stdio 和 HTTP 两种模式）
5. ✅ 常见问题排查

核心代码要点：

• 使用 @mcp.tool() 装饰器定义工具
• 日志输出到 stderr，不要用 print() 输出到 stdout
• OpenClaw 配置使用绝对路径
• 数据库密码等敏感信息放在单独的配置文件中

参考资料：

• MCP 官方文档：https://modelcontextprotocol.io
• OpenClaw 插件文档：https://docs.openclaw.ai/plugins
• Python MCP SDK：https://github.com/modelcontextprotocol/python-sdk