【技术干货】Text2SQL MCP服务端：用Python实现自然语言转SQL的终极方案

Text2SQL MCP服务端。这个工具能将自然语言转换为SQL查询，让不懂SQL的业务人员也能轻松操作数据库！

学编程的头没秃

722人浏览 · 2025-05-24 15:00:58

学编程的头没秃 · 2025-05-24 15:00:58 发布

Text2SQL MCP服务端。这个工具能将自然语言转换为SQL查询，让不懂SQL的业务人员也能轻松操作数据库！

一、项目背景

在日常工作中，我们经常遇到这样的场景：

产品经理想直接从数据库查数据但不会写SQL
数据分析师需要频繁查询数据库但SQL不熟练
开发人员需要为不同业务方编写重复的SQL查询接口

Text2SQL MCP服务端就是为了解决这些问题而生的！它基于LangChain和SQLAlchemy，MCP构建。

二、核心功能

1. 多数据库支持

支持SQLite、MySQL、PostgreSQL等多种数据库，通过统一的接口进行操作。

2. 四大核心能力

SQL查询：执行SELECT查询并返回JSON格式结果
SQL执行：执行INSERT/UPDATE/DELETE等操作
表结构查询：获取数据库表结构信息
方言识别：自动识别数据库类型

三、技术实现

1. 创建项目

# 使用 pip 安装（推荐）
pip install uv

# 使用官网脚本下载
# Linux/macOS
curl -LsSf https://astral.sh/uv/install.sh | sh

# Windows (PowerShell)
irm https://astral.sh/uv/install.ps1 | iex

uv init  # 初始化

uv venv  # 默认创建 `.venv` 目录

# Linux/macOS (bash/zsh)
source .venv/bin/activate  # 默认路径
# 或
source myenv/bin/activate  # 自定义路径

# Windows (PowerShell)
.\.venv\Scripts\activate  # 默认路径
# 或
.\myenv\Scripts\activate  # 自定义路径

# Windows (CMD)
.venv\Scripts\activate.bat

安装依赖

uv add mcp
uv add sqlalchemy
uv add langchain_community

2. 数据库连接核心类

from sqlalchemy import create_engine, text
from sqlalchemy.exc import SQLAlchemyError
from langchain_community.utilities import SQLDatabase
from typing import Optional, Dict, Any, List, Union

class UniversalDBConnector:
    """通用数据库连接器，支持多种数据库类型和SQL操作"""

    def __init__(self, connection_str: str, **kwargs):
        self.engine = create_engine(connection_str, **kwargs)
        self.db = SQLDatabase(self.engine)

    def get_table_info(self, table_names: Optional[List[str]] = None) -> str:
        """获取数据库表结构信息"""
        return self.db.get_table_info(table_names=table_names)

    def query(self, sql: str) -> list[dict[str, Any] | dict[Any, Any]] | dict[str, str]:
        """执行查询SQL语句"""
        with self.engine.connect() as conn:
            try:
                result = conn.execute(text(sql))
                return [dict(row._asdict()) if hasattr(row, '_asdict') else dict(row) for row in result]
            except SQLAlchemyError as e:
                return {"error": f"SQL执行错误: {str(e)}"}

    def execute(self, sql: str) -> Union[int, None]:
        """执行非查询SQL语句"""
        with self.engine.begin() as conn:
            try:
                result = conn.execute(text(sql))
                return result.rowcount
            except SQLAlchemyError as e:
                raise SQLAlchemyError(f"SQL执行错误: {str(e)}")

    @property
    def dialect(self) -> str:
        """获取数据库方言"""
        return self.db.dialect

    def close(self) -> None:
        """关闭数据库连接"""
        self.engine.dispose()

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.close()

3. MCP服务端实现

from mcp.server import FastMCP
from SQLMCP.Text2SQL import UniversalDBConnector

# 初始化MCP服务器
mcp = FastMCP("SQL-mcp-server")
mcp.version = "1.2.0"
mcp.description = """基于Langchain+sqlalchemy的SQLMCP服务器
"""

@mcp.tool(name='query', description='执行SQL查询语句(尽量限制返回数量)')
def query(sql: str, db_path: str = None) -> dict:
    """执行SQL查询语句"""
    if db_path isNone:
        return {"error": "db_path is required", "data": None}

    try:
        db = UniversalDBConnector(db_path)
        result = db.query(sql)
        db.close()

        serializable_result = []
        for row in result:
            serializable_row = {}
            for k, v in row.items():
                try:
                    import json
                    json.dumps(v)
                    serializable_row[k] = v
                except (TypeError, ValueError):
                    serializable_row[k] = str(v)
            serializable_result.append(serializable_row)

        return {"error": None, "data": serializable_result}
    except Exception as e:
        return {"error": str(e), "data": None}

@mcp.tool(name='execute', description='执行SQL更新语句')
def execute(sql: str, db_path: str = None) -> dict:
    """执行SQL更新语句"""
    if db_path isNone:
        return {"error": "db_path is required"}
    db = UniversalDBConnector(db_path)
    result = db.execute(sql)
    db.close()
    return {"data": result}

@mcp.tool(name='get_table_info', description='获取数据库表信息')
def get_table_info(db_path: str = None, table_names: list[str]=None) -> dict:
    """获取数据库表信息"""
    if db_path isNone:
        return {"error": "db_path is required"}
    db = UniversalDBConnector(db_path)
    result = db.get_table_info(table_names=table_names)
    db.close()
    return {"data": result}

@mcp.tool(name='dialect', description='获取数据库方言')
def dialect(db_path: str = None) -> dict:
    """获取数据库方言"""
    if db_path isNone:
        return {"error": "db_path is required"}
    db = UniversalDBConnector(db_path)
    result = db.dialect()
    db.close()
    return {"data": result}

四、使用示例

五、项目优势

开箱即用：简单配置即可快速搭建服务
跨数据库：一套代码支持多种数据库
安全可靠：完善的错误处理和资源管理
易于扩展：可轻松集成LLM实现自然语言转SQL

那么，如何系统的去学习大模型LLM？

作为一名从业五年的资深大模型算法工程师，我经常会收到一些评论和私信，我是小白，学习大模型该从哪里入手呢？我自学没有方向怎么办？这个地方我不会啊。如果你也有类似的经历，一定要继续看下去！这些问题啊，也不是三言两语啊就能讲明白的。

所以我综合了大模型的所有知识点，给大家带来一套全网最全最细的大模型零基础教程。在做这套教程之前呢，我就曾放空大脑，以一个大模型小白的角度去重新解析它，采用基础知识和实战项目相结合的教学方式，历时3个月，终于完成了这样的课程，让你真正体会到什么是每一秒都在疯狂输出知识点。

由于篇幅有限，⚡️ 朋友们如果有需要全套《2025全新制作的大模型全套资料》，扫码获取~
在这里插入图片描述

👉大模型学习指南+路线汇总👈

我们这套大模型资料呢，会从基础篇、进阶篇和项目实战篇等三大方面来讲解。
在这里插入图片描述

👉①.基础篇👈

基础篇里面包括了Python快速入门、AI开发环境搭建及提示词工程，带你学习大模型核心原理、prompt使用技巧、Transformer架构和预训练、SFT、RLHF等一些基础概念，用最易懂的方式带你入门大模型。
在这里插入图片描述

👉②.进阶篇👈

接下来是进阶篇，你将掌握RAG、Agent、Langchain、大模型微调和私有化部署，学习如何构建外挂知识库并和自己的企业相结合，学习如何使用langchain框架提高开发效率和代码质量、学习如何选择合适的基座模型并进行数据集的收集预处理以及具体的模型微调等等。
在这里插入图片描述

👉③.实战篇👈

实战篇会手把手带着大家练习企业级的落地项目（已脱敏），比如RAG医疗问答系统、Agent智能电商客服系统、数字人项目实战、教育行业智能助教等等，从而帮助大家更好的应对大模型时代的挑战。
在这里插入图片描述

👉④.福利篇👈

最后呢，会给大家一个小福利，课程视频中的所有素材，有搭建AI开发环境资料包，还有学习计划表，几十上百G素材、电子书和课件等等，只要你能想到的素材，我这里几乎都有。我已经全部上传到CSDN，朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】
在这里插入图片描述
相信我，这套大模型系统教程将会是全网最齐全最易懂的小白专用课！！