from mcp.server.fastmcp import FastMCP

@mcp.tool() / @mcp.resource()

mcp.run() 

uv init mcp-text2cd mcp-text2uv venv && source .venv/bin/activateuv add "mcp[cli]" sqlalchemy sqlglot pydantic

# 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")

# 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")