MCP SQL

MySQL MCP服务器

MySQL数据库的模型上下文协议（MCP）服务器。该服务器为通过MCP探索和查询MySQL数据库提供了一个统一的接口。

特性

• MySQL支持：连接到MySQL数据库
• 统一接口：MySQL操作的一致性工具和API
• 数据库特定优化：使用MySQL优化的SQL语法
• 模式探索：列出数据库、表和关系
• 查询执行：使用适当的参数处理运行SQL查询
• 资源支持：表数据的MCP资源端点
• LangGraph文本到SQL代理：智能代理，通过自动模式探索和错误恢复将自然语言转换为SQL查询

安装

1. 安装依赖项：

pip install -r requirements.txt

要求包括：

• 用于数据库访问的MySQL连接器
• 服务器的MCP和FastMCP
• 用于文本到SQL代理的LangChain和LangGraph
• LangChain OpenAI集成支持LLM

用法

连接串

MySQL

# Using connection string
python mysql-db-server.py --conn "mysql://user:password@host:port/database"

# Using environment variable
export DATABASE_CONNECTION_STRING="mysql://user:password@host:port/database"
python mysql-db-server.py

命令行选项

python mysql-db-server.py [OPTIONS]

Options:
  --conn TEXT           MySQL connection string (format: mysql://user:password@host:port/database)
  --transport TEXT      Transport protocol: stdio, sse, or streamable-http (default: stdio)
  --host TEXT           Host to bind for SSE/HTTP transports (default: 127.0.0.1)
  --port INTEGER        Port to bind for SSE/HTTP transports (default: 8000)
  --mount TEXT          Optional mount path for SSE transport (e.g., /mcp)
  --readonly            Enable read-only mode (prevents INSERT, UPDATE, DELETE, etc.)
  --help                Show this message and exit

环境变量

• DATABASE_CONNECTION_STRING：MySQL连接字符串
• DATABASE_READONLY：设置为“true”、“1”或“yes”以启用只读模式
• DATABASE_STATEMENT_TIMEOUT_MS：查询超时（毫秒）
• MCP_TRANSPORT：传输协议（stdio、sse、可流式传输http）
• MCP_HOST：网络传输主机
• MCP_PORT：网络传输端口
• MCP_SSE_MOUNT：SSE运输的装载路径

只读模式

只读模式禁止任何写操作（INSERT、UPDATE、DELETE、DROP等），只允许SELECT、SHOW、WITH、VALUES和EXPLAIN查询。

通过命令行启用：

python mysql-db-server.py --conn "mysql://user:password@host:port/database" --readonly

通过环境变量启用：

export DATABASE_READONLY="true"
python mysql-db-server.py --conn "mysql://user:password@host:port/database"

检查是否启用了只读模式：

# Get tools from MCP client
tools = await client.get_tools()
server_info_tool = next((t for t in tools if t.name == "server_info"), None)

if server_info_tool:
    result = await server_info_tool.ainvoke({})
    print(result["readonly"])  # True if read-only mode is enabled

LangGraph文本到SQL代理

该项目包括一个复杂的基于LangGraph的代理，可以将自然语言问题转换为SQL查询。代理会自动探索数据库模式，生成SQL查询，执行它们，并在出现错误时优化查询。

特性

• 智能模式探索:

- 使用LLM分析自动识别相关表（对于≤3个表跳过LLM调用） - 只探索查询所需的表（快得多！） - 描述具有精确列名的表结构 - 获取外键关系以获得更好的JOIN - 跨查询缓存架构信息（基于会话）

• 查询验证:

- 在生成SQL之前，验证用户查询实际上是数据库问题 - 首先使用快速启发式（基于集合的关键字匹配），然后仅对模棱两可的情况使用LLM - 尽早拒绝胡言乱语、问候语和非数据库问题 - 优化以避免90%以上的案件需要LLM

• 智能SQL生成:

- 使用LLM和思维链推理逐步生成查询 - 处理包含子查询的多部分问题（例如，“查找X，然后查找Y”） - 窗口功能支持：对于“每组前N名”、组内排名、比较行、运行总计、百分位数、移动平均值等适当场景，自动使用窗口函数（ROW_NUMBER（）、RANK（）、DENSE_RANK（））、LAG（）、LEAD（）、SUM（）OVER等） - 领带处理：通过返回所有具有最大/最小值的实体，正确处理“最大/最小”查询中的关系 - 验证SQL语法并自动修复简单问题

• 信心评分和自动优化:

- 根据实际查询执行结果计算置信度分数 - 显式检查SQL是否回答了问题 （不仅仅是语法正确性） - 当置信度低或检测到错误时自动优化查询 - 为每个查询提供详细的分析和推理

• 错误恢复:

- 智能解析SQL错误以提取可操作的信息 - 执行失败时自动重试改进的查询 - 在修复简单错误（例如列名）时保留查询结构

• 性能优化:

- 并行执行模式探索操作 - 编译正则表达式模式以实现更快的文本处理 - 尽可能重用测试查询结果（避免重复执行） - 基于状态的SQL存储（避免重新提取） - O（1）表/列检查的基于集合的查找 - 模式和列缓存的LRU缓存管理（防止无限增长） - LLM和数据库调用的超时处理（防止挂起） - 集中式辅助函数减少了代码重复，提高了可维护性 - 所有提示均按以下方式组织 prompts.py 更容易更新 - 组织良好的代码结构，具有清晰的部分（图构建、辅助方法、节点方法、边缘方法、公共方法）

• 稳健的错误处理:

- 输入验证（拒绝None、空或无效查询） - 所有LLM和数据库调用的超时保护（验证超时是否大于0） - 当LLM返回空/格式错误的响应时，出现优雅的回退 - 使用try/except对所有转换（float、int）进行类型安全 - 正则表达式组验证（使用前检查组是否为空） - 缓存结构验证（优雅地处理损坏的缓存） - 安全字符串解析（处理格式错误的表资源、错误消息） - 工具调用结构验证（处理前验证字典结构） - 调试综合日志记录（可选，可禁用）

• MCP集成：无缝使用MCP工具进行数据库操作

设置

1. 启动MCP服务器 （在一个终端中）：

python mysql-db-server.py --conn "mysql://user:password@host:port/database" --transport streamable-http --port 8000

1. 使用代理 （在Python/Jupyter中）：

from langchain_mcp_adapters.client import MultiServerMCPClient
from langchain_openai import ChatOpenAI
from text_to_sql_agent import TextToSQLAgent

# Connect to MCP server
client = MultiServerMCPClient({
    "mysql-server": {
        "url": "http://localhost:8000/mcp",
        "transport": "streamable_http"
    }
})

# Initialize LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Create agent with optional configuration
agent = TextToSQLAgent(
    mcp_client=client,
    llm=llm,
    max_query_attempts=3,  # Maximum retry attempts
    llm_timeout=60,  # Timeout for LLM calls (seconds)
    query_timeout=30,  # Timeout for database queries (seconds)
    max_schema_cache_size=1000,  # Maximum table descriptions to cache
    max_column_cache_size=500,  # Maximum column name extractions to cache
    enable_logging=True  # Enable logging for debugging
)
llm = ChatOpenAI(
    api_key="your-openai-api-key",
    model="gpt-4o-mini",
    temperature=0
)

# Create the agent
agent = TextToSQLAgent(
    mcp_client=client,
    llm=llm,
    max_query_attempts=3  # Maximum retry attempts
)

用法示例

基本查询

# Ask a natural language question
result = await agent.query("How many authors are in the database?")

# Get the final answer
answer = agent.get_final_answer(result)
print(answer)

带筛选的查询

# Complex queries with filters
result = await agent.query("Show me all authors born after 1950")
print(agent.get_final_answer(result))

集合查询

# Statistical queries
result = await agent.query("What is the average birth year of all authors?")
print(agent.get_final_answer(result))

运作原理

代理遵循优化的工作流程：

1. 模式探索 （带缓存和并行化）：

- 列出所有可用表（在第一次调用后缓存） - 使用LLM智能识别相关表（跳过≤3个表） - 仅描述相关表的表结构（并行执行） - 获取外键关系以获得更好的JOIN（并行执行） - 为后续查询缓存所有架构信息

1. 查询验证 （生成SQL之前）：

- 验证用户查询是否为有效的数据库问题 - 首先使用快速启发式（关键字匹配） - 仅在模棱两可的情况下才回到LLM验证 - 尽早拒绝无效查询，以避免不必要的处理

1. SQL生成 （有信心评分）：

- 使用LLM和思维链推理进行逐步生成 - 在提示中包括架构、外键和列名 - 执行测试查询（LIMIT 3）以获取示例结果 - 计算置信度得分和分析（单次LLM调用） - 置信度评分明确检查SQL是否回答了问题 - 验证SQL语法并检测关键问题 - 为边缘布线设置细化标志（不直接细化） - 将最终SQL存储在状态中以避免重新提取

1. SQL优化 （如果需要）：

- 当置信度低或检测到错误时，单独的节点处理细化 - 如果需要，获取缺失的架构 - 使用分析和错误上下文优化SQL - 重新执行测试查询并重新计算置信度

1. 查询执行 （优化以避免冗余）：

- 使用状态存储的SQL（避免重新提取） - 如果测试查询结果包含所有数据，则重用它们（避免冗余执行） - 在重复使用测试结果时，尊重原始的LIMIT条款 （例如，LIMIT 1返回1行，而不是所有测试结果） - 仅在需要时执行完整查询

1. 错误恢复 （具有智能解析功能）：

- 智能解析SQL错误（提取错误类型、列、表） - 将错误上下文传递给SQL生成（可选细化） - 修复简单错误时保留查询结构 - 自动重试失败的查询（最多 max_query_attempts)

代理状态图

代理使用具有以下节点的LangGraph状态机：

• exploreschema：发现并缓存数据库架构（使用条件路由完成）
• generate.sql：使用LLM将自然语言转换为SQL，计算置信度，设置细化标志
• refine_sql：根据置信度得分和分析优化SQL（为清晰起见，单独的节点）
• execute_query：通过MCP工具运行SQL查询（尽可能重用测试结果）
• refine_query：改进基于错误反馈的查询（路由回generate_sql）
• 工具：处理模式探索的工具调用

建筑：代理遵循LangGraph最佳实践，并明确区分：

• 节点：进程状态和返回更新（无条件逻辑）
• 边缘：根据状态标志（所有编排逻辑）做出路由决策

有关体系结构、工作流图以及如何扩展代理的详细说明，请参阅 AGENT_ARCHITECTURE.md.

高级用法

自定义配置

# Run with custom LangGraph config
result = await agent.query(
    "Find all authors with more than 5 books",
    config={"recursion_limit": 50}
)

访问完整状态

# Get complete agent state including all messages
result = await agent.query("Show me the database schema")

# Access messages, schema info, and query attempts
messages = result["messages"]
schema_info = result["schema_info"]
attempts = result["query_attempts"]

清洁结果的辅助功能

from langchain_core.messages import ToolMessage

def get_answer(result):
    """Extract the final answer from agent result"""
    messages = result.get("messages", [])
    for msg in reversed(messages):
        if isinstance(msg, ToolMessage) and "successfully" in msg.content.lower():
            return msg.content
    return result.get("messages", [])[-1].content if result.get("messages") else "No answer"

# Use it
result = await agent.query("How many tables are in the database?")
print(get_answer(result))

最佳结果提示

1. 具体：清晰、具体的问题最有效

- ✅ “显示1950年以后出生的所有作者” - ❌ “作者的东西”

1. 使用表名：如果你知道表名，就提出来

- ✅ “列出作者表中的所有书籍” - ✅ “作者表中有多少条记录？”

1. 指定筛选器：明确过滤条件

- ✅ “查找出生年份大于1950的作者” - ✅ “显示名称以'G'开头的作者”

1. 请求聚合：代理处理COUNT、SUM、AVG等。

- ✅ “平均出生年份是多少？” - ✅ “统计作者总数”

局限性

• 目前针对SELECT查询（只读操作）进行了优化
• 可配置最大重试次数（默认值：3）
• LLM功能需要OpenAI API密钥
• 架构信息跨同一代理实例中的查询缓存，以获得更好的性能

改进代理

有关增强代理性能、准确性和功能的全面指南，请参阅 改进.md.

✅ 已实现的功能：

• 智能桌面选择：使用LLM仅识别相关表（跳过≤3个表）
• 外键关系：获取FK信息以更好地理解JOIN
• 架构缓存：跨查询缓存架构信息（基于会话）
• 思维链推理：逐步生成查询以提高准确性
• 信心评分：根据实际查询执行结果计算置信度
• 自动精炼：检测到问题时自动改进查询
• 智能错误分析：从错误消息中提取可操作的信息
• 性能优化：编译正则表达式、并行执行、结果重用、基于状态的存储
• 代码组织:

- 集中提示 prompts.py - 辅助函数减少了重复（代码从2362行减少到2180行，减少了约7.7%） - 清晰的代码结构，有组织的部分：图构建、辅助方法、节点方法、边缘方法和公共方法

计划改进：

• 几个射击示例：添加示例查询以指导更好的SQL生成模式
• 查询说明：解释生成的SQL查询的作用
• 查询历史和学习：从过去成功的查询中学习

看 改进.md 详细的实现示例和分步说明。

MySQL功能

• 数据库级组织
• SHOW DATABASES 和 SHOW TABLES 为了性能
• DESCRIBE 表结构
• SHOW CREATE TABLE 有关详细的表格信息

工具

服务器提供以下MCP工具：

• server_info：获取服务器和数据库信息（数据库类型：MySQL，只读状态，MySQL连接器版本）
• db_identity：获取当前数据库标识详细信息（数据库类型：MySQL、数据库名称、用户、主机、端口、服务器版本）
• run_query：使用类型化输入执行SQL查询（返回markdown或JSON字符串）
• run_query_json：使用类型化输入执行SQL查询（返回JSON列表）
• list_table_resources：将表作为MCP资源URI列出(table://schema/table)-返回结构化列表
• read_table_resource：读取表格 数据 （行）通过MCP资源协议-返回JSON列表
• list_tables：列出数据库中的表-返回 标记语言 字符串（人类可读）
• describe_table：获取桌子 结构 （列、类型、约束）-返回markdown字符串
• get_foreign_keys：获取外键关系（通过SHOW CREATE TABLE）

调用MCP工具

MultiServerMCPClient 没有 call_tool() 方法。相反，您需要：

1. 使用获取工具 get_tools() 返回LangChain StructuredTool 物体
2. 按名称查找工具
3. 使用调用它 tool.ainvoke() 有适当的论据

高效的工具查找

为了获得更好的性能（特别是使用许多工具），请为O（1）查找创建一个字典：

# Get all tools
tools = await client.get_tools()

# Create dictionary for fast O(1) lookup (recommended)
tool_dict = {t.name: t for t in tools}

# Now you can call tools efficiently
server_info = await tool_dict["server_info"].ainvoke({})
tables = await tool_dict["list_tables"].ainvoke({"db_schema": None})

获取服务器信息

# Get tools
tools = await client.get_tools()
tool_dict = {t.name: t for t in tools}

# Get server info and connection details
result = await tool_dict["server_info"].ainvoke({})
# Returns: {
#   "name": "MySQL Database Explorer",
#   "database_type": "MySQL",  # Database type is explicitly included
#   "readonly": False,
#   "mysql_connector_version": "8.0.33"
# }

db_info = await tool_dict["db_identity"].ainvoke({})
# Returns: {
#   "database_type": "MySQL",  # Database type is explicitly included
#   "database": "mydatabase",
#   "user": "root@localhost",
#   "host": "localhost",
#   "port": 3306,
#   "server_version": "8.0.33"
# }

列表表格

tools = await client.get_tools()
tool_dict = {t.name: t for t in tools}

# Lists tables in the current database
result = await tool_dict["list_tables"].ainvoke({"db_schema": "mydatabase"})

# Or get tables as MCP resources
resources = await tool_dict["list_table_resources"].ainvoke({"schema": "mydatabase"})

探索表结构

tools = await client.get_tools()
tool_dict = {t.name: t for t in tools}

# Get table structure
result = await tool_dict["describe_table"].ainvoke({
    "table_name": "users",
    "db_schema": "mydatabase"
})

# Get foreign key relationships
fks = await tool_dict["get_foreign_keys"].ainvoke({
    "table_name": "users",
    "db_schema": "mydatabase"
})

执行查询

tools = await client.get_tools()
tool_dict = {t.name: t for t in tools}

# Execute query with markdown output
result = await tool_dict["run_query"].ainvoke({
    "input": {
        "sql": "SELECT * FROM users LIMIT 10",
        "format": "markdown"
    }
})

# Execute query with JSON output
result = await tool_dict["run_query_json"].ainvoke({
    "input": {
        "sql": "SELECT * FROM users LIMIT 10",
        "row_limit": 100
    }
})

MCP资源工具

这些工具实现了MCP（模型上下文协议）资源模式，该模式允许将表视为可通过标准化URI访问的可发现资源。

list_table_resources

列出模式中的所有表，并将其作为MCP资源URI返回。这使MCP客户端能够动态发现可用表。

它的作用：

• 查询数据库以从指定架构中获取所有表名
• 将每个表格式化为资源URI： table://schema/table_name
• 返回这些URI的列表

例子：

# Get all tables as resource URIs
tools = await client.get_tools()
tool_dict = {t.name: t for t in tools}
resources = await tool_dict["list_table_resources"].ainvoke({"schema": "mydatabase"})
# Returns: ["table://mydatabase/users", "table://mydatabase/orders", "table://mydatabase/products"]

使用案例：

• MCP客户端中的动态表发现
• 构建列出可用表的UI
• 与MCP资源感知工具集成

read_table_resource

使用MCP资源协议从特定表读取数据。这提供了一种访问表数据的标准化方法。

它的作用：

• 执行 SELECT * FROM schema.table 有行限制
• 以JSON格式的字典列表返回表行
• 每个字典代表一行，列名作为关键字

例子：

# Read table data via MCP resource protocol
tools = await client.get_tools()
tool_dict = {t.name: t for t in tools}
data = await tool_dict["read_table_resource"].ainvoke({
    "schema": "mydatabase",
    "table": "users",
    "row_limit": 50  # Limits number of rows returned
})
# Returns: [
#   {"id": 1, "name": "Alice", "email": "alice@example.com"},
#   {"id": 2, "name": "Bob", "email": "bob@example.com"},
#   ...
# ]

使用案例：

• 无需编写SQL即可快速预览表
• MCP资源感知客户端，可以通过URI获取表数据
• 数据探索和检查工具

主要区别在于 run_query:

• read_table_resource：读取整个表的简单、标准化的方法（不需要SQL）
• run_query：灵活，允许使用自定义WHERE子句、JOIN等进行任何SQL查询。

比较：资源工具与常规工具

list_table_resources 对比 list_tables

这两个表都列出了，但用途不同：

示例比较：

# Get tools once
tools = await client.get_tools()
tool_dict = {t.name: t for t in tools}

# list_table_resources - structured for programs
resources = await tool_dict["list_table_resources"].ainvoke({"schema": "mydb"})
# Returns: ["table://mydb/users", "table://mydb/orders"]

# list_tables - formatted for humans
tables = await tool_dict["list_tables"].ainvoke({"db_schema": "mydb"})
# Returns: "| Tables_in_mydb |\n|-----------------|\n| users          |\n| orders         |"

read_table_resource 对比 describe_table

这些服务 完全不同的目的 -它们是互补的，而不是多余的：

示例比较：

# Get tools once
tools = await client.get_tools()
tool_dict = {t.name: t for t in tools}

# read_table_resource - get the DATA
data = await tool_dict["read_table_resource"].ainvoke({
    "schema": "mydb", "table": "users", "row_limit": 10
})
# Returns: [{"id": 1, "name": "Alice", "email": "alice@example.com"}, ...]

# describe_table - get the STRUCTURE
structure = await tool_dict["describe_table"].ainvoke({
    "table_name": "users", "db_schema": "mydb"
})
# Returns: "| Field | Type | Null | Key | Default | Extra |\n|-------|------|------|-----|---------|-------|\n| id | int | NO | PRI | NULL | auto_increment |\n| name | varchar(100) | NO | | NULL | |"

摘要：

• 使用 read_table_resource 当你想 查看数据 在表格中
• 使用 describe_table 当你想 查看架构/结构 一张桌子
• 他们回答不同的问题：“表中有什么？”与“表是如何定义的？”

备注

• MySQL连接字符串必须以开头 mysql://
• 格式为： mysql://user:password@host:port/database
• 为了兼容性，所有数据库名称都被视为模式

文档

• AGENT_ARCHITECTURE.md：详细的架构说明、工作流程图和扩展指南

- 代理如何在内部工作 - 可视化工作流程图 - 添加节点和工具的分步指南 - 代码示例和模式

• 改进.md：全面改进指南

- 通过优先级排名增强想法 - 实施示例 - 性能优化 - 高级功能

改进代理

有关增强代理性能、准确性和功能的全面指南，请参阅 改进.md.

改进指南包括：

• 优先级排名 每项改进的（高/中/低）
• 详细的代码示例 带有实现片段
• 分步集成说明
• 工作流程图 展示代理如何随着改进而进化
• 测试策略 以及验证方法
• 效益分析 对于每一项增强

快速开始：从高优先级项目开始，如外键关系、更好的错误解析和列名建议，以立即产生影响。

故障排除

连接问题

• 确保MySQL服务器正在运行且可访问
• 检查连接字符串格式和凭据
• 验证网络连接和防火墙设置

MySQL特定问题

• 确保MySQL服务器支持连接协议
• 检查指定数据库的用户权限
• 验证MySQL连接器版本兼容性