把自然语言变成SQL 查询、把需求描述变成可执行的 Python 程序，这两件事听起来像“让模型写两段字符串”。但真正落地时，你很快会遇到三类硬问题：上下文怎么给：数据库 schema、业务词表、样例值、运行时约束……怎么稳定、标准化地提供给模型？动作怎么做：SQL 到底在哪执行？Python 在什么隔离环境里跑？失败怎么回滚？安全怎么兜底：提示注入、越权读写、工具链漏洞、审计与授权……不处理就不可能上线。MCP（Model Context Protocol）的价值，恰好是把“给上下文”和“让模型做动作”这两件事，从具体模型/具体应用里抽离出来，做成标准接口：Host（LLM 应用）↔ Client（连接器）↔ Server（能力提供方），通信基于JSON-RPC 2.0、支持能力协商、并且把Resources / Tools / Prompts做成统一语义。 下面这篇文章按“从零开始”的方式，给出一套Text2SQL + Text2Python 的 MCP Server 设计与实现骨架，并把关键原理讲透：怎么把它做成真正可用、可控、可评估的工程系统。1. 先对齐：你要做的是“Text2SQL/2Python”，还是“可上线的查询/执行系统”？
而现实里，Text2SQL 的标杆数据集（如 Spider / BIRD）之所以难，是因为它考察的是跨库泛化、复杂查询、真实数据值、以及执行正确性：- Spider 1.0：10,181 个问题、200 个多表数据库、复杂 SQL，训练/测试库不重合，强调对新 schema 的泛化。
- BIRD：12,751 对 question-SQL，95 个大库，总数据量 33.4GB，强调“脏值/外部知识/效率”等真实问题。
这意味着：你的 MCP 实现不能只产出 SQL 字符串，而要产出一个可控系统：- 能把正确的 schema 片段喂给模型（减少噪声）
- 能把 SQL 安全地执行在受控连接上（只读/超时/限行数）
- 能把执行反馈（错误、统计、Explain）结构化返回给模型做自修复
2. MCP 在这里怎么用：Resources / Tools / Prompts 的职责边界
2.1 Resources：把“上下文”做成可寻址、可缓存的对象
- db://schema/{db}：库级 schema（表/列/类型/外键）
- db://dict/{db}：业务词表/字段别名/枚举含义
- runtime://python：Python 沙箱能力说明（允许包、资源限制、禁网等）
- Resources 的关键不是“内容多”，而是稳定、可复用、可做缓存与权限控制。
2.2 Tools：把“动作”做成可审计、可限权的函数
- sql.execute(sql, params, db, max_rows, timeout_ms)：执行只读 SQL
- sql.explain(sql, db)：Explain/Query Plan
- python.run(code, files, limits)：在隔离环境运行代码并返回 stdout/stderr/产物摘要
- MCP 规范强调工具调用需要清晰的用户授权/控制语义，并提醒“工具代表任意代码执行路径，必须慎重”。
2.3 Prompts：把“最佳实践提示词模板”变成可复用资产
例如：一个专门用于 Text2SQL 的 prompt 模板，强制模型先做 schema linking 再产出 SQL AST。3. Text2SQL 原理拆解：从 Schema Linking 到“可执行正确性闭环”
把 Text2SQL 做成工程系统，一般是下面这条流水线（你可以把它映射成 Host 侧 Agent 的多步调用，也可以封装进 Server 侧工具里）。3.1 Schema Linking：决定“给模型哪些表/列”
LLM Text2SQL 的一个核心工程技巧是 schema linking：先筛选相关表/列，再让模型生成 SQL，减少噪声与 token 成本。但 schema linking 本身也有风险：漏召回关键列会直接导致不可修复的错误；过度裁剪会破坏结构完整性。RSL-SQL 这类工作专门讨论了这些风险并用双向链接、上下文增强与多轮自修正来提高召回。 工程实现上，schema linking 通常是三路信号融合：- 向量检索（question ↔ column/table description）
关键点：别把 linking 当成一次性过滤器，而要当成“候选生成器 + 风险兜底机制”。实践里会保留一个full schema 模式和一个linked schema 模式，失败时回退或投票。3.2 SQL 生成：别让模型“自由写字符串”
- 结构化输出：让模型输出 SQL 的结构（AST/JSON），再由你序列化成 SQL
- 语法解析与白名单：生成后用解析器（如 sqlglot）解析，确保单语句、只读、禁用危险关键词
3.3 执行与反馈：执行正确性 > 字符串相似
- 把数据库错误结构化返回给模型（错误码/位置/期望类型）
这样你才能构建自修复闭环：模型 → 执行 → 结构化报错 → 模型修正 → 再执行。4. Text2Python 原理拆解：把“写代码”变成“受控执行任务”
Text2Python 真正的难点不是生成代码，而是安全、可重复、可诊断：4.1 先定义 I/O 合同（Contract）
没有 contract，模型只会“写到能跑为止”，并且难以评估。4.2 沙箱：不要幻想“Python 里限制 builtins 就安全”
安全执行 Python 的底线是进程级隔离（容器/虚拟机/受限用户），并配合：如果你把 Python 直接 exec() 在服务进程里，迟早会出事故。4.3 诊断闭环：让模型基于“可操作反馈”修复
- 可选：单测结果、lint/type check 结果
5. 用 MCP Python SDK（FastMCP）搭两个 Server
下面用 MCP 官方 Python SDK 的FastMCP写骨架。官方文档与 Quickstart 的写法是：from mcp.server.fastmcp import FastMCP
@mcp.tool() / @mcp.resource()
STDIO 模式日志注意：不要向 stdout 打日志，会破坏 JSON-RPC 消息；写 stderr 或用 logging。
5.1 依赖与目录
uv init mcp-text2cd mcp-text2uv venv && source .venv/bin/activateuv add "mcp[cli]" sqlalchemy sqlglot pydantic
（SQL 驱动按需加：psycopg、pymysql 等）6. SQL MCP Server：schema 资源 + 安全执行工具
6.1 关键设计点
资源：提供 schema（并可加入字段别名、注释、样例值摘要）工具：执行 SQL（只读、单语句、自动补 LIMIT、超时、限行数）返回结构：建议用 JSON（便于模型稳定消费），FastMCP 支持 json_response=True。 ()6.2 示例代码（骨架）
# sql_server.pyfrom __future__ import annotationsimport osfrom typing import Any, Optionalimport sqlalchemy as saimport sqlglotfrom mcp.server.fastmcp import FastMCPfrom pydantic import BaseModelmcp = FastMCP("text2sql", json_response=True)ENGINE_URL = os.environ.get("DB_URL", "sqlite:///./demo.db")engine = sa.create_engine(ENGINE_URL, future=True)class QueryResult(BaseModel): columns: list[str] rows: list[list[Any]] row_count: int truncated: booldef _is_readonly(ast: sqlglot.Expression) -> bool: # 只允许 SELECT（可按需扩展到 WITH + SELECT） return ast.key.upper() in {"SELECT", "WITH"}def _normalize_sql(sql: str) -> str: # 禁止多语句 if ";" in sql.strip().rstrip(";"): raise ValueError("Multiple statements are not allowed.") return sql.strip().rstrip(";")@mcp.resource("db://schema")def get_schema() -> str: """Return database schema (tables/columns/types).""" insp = sa.inspect(engine) out = [] for t in insp.get_table_names(): cols = insp.get_columns(t) out.append( { "table": t, "columns": [{"name": c["name"], "type": str(c["type"])} for c in cols], "pk": insp.get_pk_constraint(t).get("constrained_columns", []), "fks": insp.get_foreign_keys(t), } ) return sa.json.dumps(out) # 或者用 json.dumps@mcp.tool()def sql_execute(sql: str, max_rows: int = 200) -> QueryResult: """Execute a read-only SQL query safely. Args: sql: SQL statement (SELECT only). max_rows: Maximum rows to return (truncate if exceeded). """ sql = _normalize_sql(sql) ast = sqlglot.parse_one(sql) if not _is_readonly(ast): raise ValueError("Only SELECT queries are allowed.") # 可选：自动补 LIMIT（示意，生产建议更严谨） if "LIMIT" not in sql.upper(): sql = f"{sql} LIMIT {max_rows + 1}" with engine.connect() as conn: res = conn.execute(sa.text(sql)) cols = list(res.keys()) rows = res.fetchmany(max_rows + 1) truncated = len(rows) > max_rows rows = rows[:max_rows] return QueryResult( columns=cols, rows=[list(r) for r in rows], row_count=len(rows), truncated=truncated, )if __name__ == "__main__": mcp.run(transport="stdio")
- 超时：驱动级/连接级 statement_timeout（Postgres）或应用侧 kill
- Explain：单独 sql_explain 工具，做 cost 阈值
- 参数化：避免模型直接拼接值（尤其文本条件），要求传 params 并在服务端绑定
7. Python MCP Server：隔离执行 + 结构化结果
7.1 两种落地方式
- 方式 A（推荐）：在 MCP Server 里调用“沙箱执行器”（容器/微虚拟机/受限用户进程），Server 只负责编排与回传结果
- 方式 B（仅本地 Demo）：subprocess + 资源限制（不等于强安全，但能展示闭环）
7.2 示例代码（本地 Demo 骨架）
# python_server.pyfrom __future__ import annotationsimport subprocessimport tempfilefrom typing import Anyfrom mcp.server.fastmcp import FastMCPfrom pydantic import BaseModelmcp = FastMCP("text2python", json_response=True)class PyRunResult(BaseModel): exit_code: int stdout: str stderr: str@mcp.tool()def python_run(code: str, timeout_s: int = 3) -> PyRunResult: """Run python code in a subprocess (demo sandbox). Args: code: Python code to execute. timeout_s: Execution timeout. """ with tempfile.TemporaryDirectory() as td: p = subprocess.run( ["python", "-I", "-c", code], # -I: isolate (no user site-packages) cwd=td, capture_output=True, text=True, timeout=timeout_s, ) return PyRunResult( exit_code=p.returncode, stdout=p.stdout[-8000:], # 截断策略 stderr=p.stderr[-8000:], )if __name__ == "__main__": mcp.run(transport="stdio")
重要：-I 只能减少环境污染，不等于安全隔离。真正要上线，还是要进程级隔离 + 禁网 + 配额。
8. “把 Text2SQL/Text2Python 封装进 MCP Tool”——用 Sampling 做 Server 侧生成（可选高级）
如果你希望 Host 只调用一个工具：text2sql.query(question)，由 Server 自己完成“生成→执行→修复”，可以用 MCP 的Sampling：Server 在工具内部请求 Host 提供一次 LLM 采样。Python SDK 示例是 ctx.session.create_message(...)。
注意：MCP 规范对 sampling 的用户控制有明确要求（是否采样、发送的 prompt、返回结果可见性等）。 9. 安全与攻防：MCP + 执行动作 = 必须默认“零信任”
当你把“读数据、跑代码、改文件”的能力接给模型，安全问题会从“模型胡说”升级为“模型能做事”。MCP 规范本身就把安全与用户控制作为关键原则。 - 工具链漏洞：官方 MCP Git Server 曾出现路径校验问题（CVE-2025-68145），需要升级修复版本；本质是“工具提供文件系统能力时，边界校验必须严密”。
- 间接提示注入：来自外部资源（文档、工单、网页）的文本可能携带“隐藏指令”，诱导模型越权调用工具。Microsoft 给出过针对 MCP 场景的缓解建议（隔离指令与数据、最小权限、显式用户确认等）。
- SQL：只读账号、schema 白名单、限时限量、Explain 阈值、审计日志
- Python：容器/VM 隔离、禁网、配额、工作目录白名单、产物扫描
- MCP：工具调用需要明确授权 UI、敏感资源需要 OAuth/Token、版本升级与漏洞跟踪
10. 真正要追求的指标：可评估、可回归、可演进
Text2SQL：建议用 Spider/BIRD 风格的“执行正确性”做回归思路，至少在你的业务库上建立query 测试集 + 可执行断言（行数、聚合结果、关键字段）。 Text2Python：用“单测 + 输出契约 + 资源上限”做回归，把“能跑”变成“稳定产出”。
