AI导哈

数据库MCP服务器

一个基于FastMCP框架的企业级微能力平台（MCP）服务器模板，提供安全的数据库访问网关，支持Oracle和PostgreSQL等多种数据库，具备中间件认证和数据库抽象层功能。

  • by Simulation-Theory
  • Python
  • 本地部署
  • 更新于
在 GitHub 查看 Simulation-Theory/coremcp

使用说明

企业级数据库MCP服务器

该项目提供了一个生产质量模板，用于使用FastMCP框架构建安全的微能力平台（MCP）服务器。它展示了创建具有强大身份验证和干净架构设计的企业数据库（Oracle、PostgreSQL等）AI可访问网关的最佳实践。

架构概述

此模板实现了几个关键的架构模式：

  • 基于中间件的身份验证：承载令牌身份验证在中间件层处理，与业务逻辑完全解耦。
  • 数据库抽象层：工具与存根函数交互，而不是直接与数据库交互，从而实现了关注点的清晰分离和简化的测试。
  • 混合工具设计:
    • 低级工具：使用架构上下文执行复杂查询的通用SQL
    • 高级工具：专门为通用操作构建的接口，降低了复杂性
  • 架构验证：基于Pydantic的验证可确保数据完整性，并为所有工具参数提供丰富的元数据。

先决条件

  • Ubuntu 22.04 LTS（或兼容的Linux发行版）
  • Python 3.10或更高版本
  • 包安装的网络访问

第1节：Ubuntu服务器部署

第一步：系统准备

更新系统并安装所需的软件包：

sudo apt update
sudo apt install python3-pip python3.10-venv -y

步骤2：项目设置

创建项目目录

mkdir -p ~/mcp-server
cd ~/mcp-server

复制服务器代码

放置 server.py 项目目录中的文件。

创建Python虚拟环境

python3 -m venv venv

激活虚拟环境

source venv/bin/activate

创建需求文件

echo "fastmcp[server]" > requirements.txt

再进行

pip install -r requirements.txt

步骤3：身份验证配置

创建承载令牌文件：

# Generate a secure token (example using openssl)
openssl rand -hex 32 > mcp_secret.token

# Or manually create with your own token
echo "YOUR_SECURE_TOKEN_HERE" > mcp_secret.token

安全说明：确保此文件具有受限权限：

chmod 600 mcp_secret.token

步骤4：初步测试

直接运行服务器以验证功能：

python server.py

步骤5：使用Systemd进行生产部署

对于生产环境，将服务器配置为systemd服务，以确保在出现故障时自动启动和重新启动。

创建服务文件

sudo nano /etc/systemd/system/mcp-server.service

添加服务配置

将/home/ubuntu/mcp服务器替换为实际的项目路径：

Description=Enterprise Database MCP Server
After=network.target

[Service]
Type=simple
User=ubuntu
Group=ubuntu
WorkingDirectory=/home/ubuntu/mcp-server
ExecStart=/home/ubuntu/mcp-server/venv/bin/python /home/ubuntu/mcp-server/server.py
Restart=always
RestartSec=3
StandardOutput=journal
StandardError=journal

[Install]
WantedBy=multi-user.target```

Enable and Start Service

```bash
sudo systemctl daemon-reload
sudo systemctl enable mcp-server.service
sudo systemctl start mcp-server.service

验证服务状态

bash

sudo系统ctl状态mcp-server.service

bash

sudo系统ctl状态mcp-server.service 预期输出应以绿色显示活动（运行）。

查看服务日志

bash

sudo journalctl-u mcp-server.service-f 第2节：验证测试 使用以下curl命令验证服务器功能。将YOUR_SERVER_IP替换为服务器的IP地址，将YOUR_SECURE_TOKEN_HERE替换为mcp_sesecret.TOKEN中的令牌。

测试1：身份验证失败（预期） 此测试验证没有正确身份验证的请求是否被拒绝：

bash

curl-X POSThttp://YOUR_SERVER_IP:9009/mcp
-H“内容类型：应用程序/json”
d “jsonrpc”：“2.0”， “id”：1， “方法”：“工具/调用”， “参数”：{ “name”：“get_employee_details”， “论点”：{ “employee_id”：101 } } }' 预期响应：

JSON

｛“detail”：“缺少授权标头。”｝ 测试2：高级工具-获取员工详细信息 此测试调用一个特定的、专门构建的工具：

bash

curl-X POSThttp://YOUR_SERVER_IP:9009/mcp
-H“内容类型：应用程序/json”
-H“授权：持有人YOUR_SECURE TOKEN_HERE”
d “jsonrpc”：“2.0”， “id”：1， “方法”：“工具/调用”， “参数”：{ “name”：“get_employee_details”， “论点”：{ “employee_id”：101 } } }' 预期响应：

JSON

{ “jsonrpc”：“2.0”， “id”：1， “结果”：{ “内容”：[ { “类型”：“文本”， “text”：“｛”状态“：”成功“，”员工数据“：｛”员工ID“：101，”公司名称“：”约翰“，”姓氏“：”Doe“，”工资“：95000，”部门ID“：2｝｝” } ] } } 测试3：低级工具-执行SQL查询 此测试演示了通用SQL执行能力：

bash

curl-X POSThttp://YOUR_SERVER_IP:9009/mcp
-H“内容类型：应用程序/json”
-H“授权：持有人YOUR_SECURE TOKEN_HERE”
d “jsonrpc”：“2.0”， “id”：1， “方法”：“工具/调用”， “参数”：{ “name”：“execute_sql_query”， “论点”：{ “query”：“从部门中选择部门ID、部门名称” } } }' 预期响应：

JSON

{ “jsonrpc”：“2.0”， “id”：1， “结果”：{ “内容”：[ { “类型”：“文本”， “text”：“｛”status“：”success“，”row_count“：2，”results“：〔｛”DEPARTMENT_ID“：1，”DEPARTMENT_NAME“：”Engineering“｝，｛”DEPARTMENT_ID”：2，“DEPARTMENT_ NAME”：“Sales”｝ } ] } } 测试4：安全验证 验证是否拒绝了非SELECT查询：

curl-X POSThttp://YOUR_SERVER_IP:9009/mcp
-H“内容类型：应用程序/json”
-H“授权：持有人YOUR_SECURE TOKEN_HERE”
d “jsonrpc”：“2.0”， “id”：1， “方法”：“工具/调用”， “参数”：{ “name”：“execute_sql_query”， “论点”：{ “query”：“从EMPLOYEE_ID=101的员工中删除” } } }' 预期响应：

JSON

{ “jsonrpc”：“2.0”， “id”：1， “结果”：{ “内容”：[ { “类型”：“文本”， “text”：“｛”状态“：”错误“，”消息“：”安全违规：只允许SELECT查询。”｝ } ] } } 第3节：生产数据库集成 提供的模板使用存根函数进行数据库交互。要与真实数据库集成：

对于Oracle数据库 安装Oracle客户端库

按照Oracle的官方文档在您的平台上安装Instant Client。

安装Python驱动程序

bash

pip安装oracledb 更新execute_readonly_query函数

将存根实现替换为：

python

导入oracledb

def execute_readonly_query（sql:str，params:dict=None）->列表[dict[str，Any]]： “”“对Oracle数据库执行只读SQL查询。”“” 如果不是sql.strip（）.upper（）.startswith（“SELECT”）： raise ValueError（“安全违规：只允许SELECT查询。”）

connection = oracledb.connect(
    user=os.getenv("DB_USER"),
    password=os.getenv("DB_PASSWORD"),
    dsn=os.getenv("DB_DSN")
)

try:
    cursor = connection.cursor()
    cursor.execute(sql, params or {})
    columns = [col[0] for col in cursor.description]
    results = [dict(zip(columns, row)) for row in cursor.fetchall()]
    return results
finally:
    connection.close()

对于PostgreSQL 安装Python驱动程序

bash

pip安装psycopg2二进制文件 更新execute_readonly_query函数

将存根实现替换为：

python

导入psycopg2 导入psycopg2.extras

def execute_readonly_query（sql:str，params:dict=None）->列表[dict[str，Any]]： “”“对PostgreSQL数据库执行只读SQL查询。”“” 如果不是sql.strip（）.upper（）.startswith（“SELECT”）： raise ValueError（“安全违规：只允许SELECT查询。”）

connection = psycopg2.connect(
    host=os.getenv("DB_HOST"),
    database=os.getenv("DB_NAME"),
    user=os.getenv("DB_USER"),
    password=os.getenv("DB_PASSWORD")
)

try:
    cursor = connection.cursor(cursor_factory=psycopg2.extras.RealDictCursor)
    cursor.execute(sql, params or {})
    results = [dict(row) for row in cursor.fetchall()]
    return results
finally:
    connection.close()

第4节：Simtheory平台部署 [保留用于Simtheory特定部署说明]

故障排除 服务无法启动 检查服务日志：

bash

sudo journalctl-u mcp-server.service-n 50 常见问题：

缺少mcp_secret.token文件 文件权限不正确 Python虚拟环境配置不正确 身份验证错误 验证令牌文件是否存在并且包含正确的令牌：

bash

猫mcp-secret.token 确保令牌与您在Authorization标头中发送的内容完全匹配。

端口已在使用中 检查端口9009是否已被占用：

sudo lsof ：9009 要使用其他端口，请修改server.py中的最后一行：

uvicon.run（应用程序，主机=“0.0.0.0”，端口=YOUR_port_NUMBER） 安全考虑 令牌管理：安全地存储承载令牌。永远不要将mcp_sesecret.token提交到版本控制。 数据库凭据：使用环境变量作为数据库凭据。永远不要在源文件中硬编码它们。 只读访问：为MCP服务器配置专用的只读数据库用户。 网络安全：使用防火墙规则限制仅授权客户端访问端口9009。 HTTPS：在生产环境中，将MCP服务器置于具有SSL/TLS终止的反向代理（nginx、Apache）后面。

相关 MCP