AI Agent 技能分享｜从零实现 MCP Server，让 Agent 安全读取数据库

AI Agent 技能分享｜从零实现 MCP Server，让 Agent 安全读取数据库

把数据库接给 Agent，最省事的办法是让模型生成 SQL，然后直接执行。这个方案做演示很快，我不建议照搬到业务系统里。数据库里装的是实打实的业务数据，模型偶尔选错表、漏掉租户条件，代价可能比一次回答错误大得多。

这篇用 MCP Python SDK v2、SQLAlchemy 和 SQL Server 做一个订单查询服务。Agent 只能调用事先定义好的订单工具，拿不到任意 SQL 的执行入口。

先说说直接执行 SQL 的问题

很多 Text-to-SQL 示例采用下面这条链路：

用户问题 → 大模型生成 SQL → 数据库执行 SQL → 返回结果

链路很短，风险却不少：

1. 模型可能生成UPDATE、DELETE或高消耗查询；
2. 用户可能通过提示词诱导模型查询无权访问的数据；
3. 即使只允许SELECT，仍可能出现跨租户查询、敏感字段泄露和全表扫描；
4. SQL 语法检查无法判断一条查询在业务上是否越权；
5. 将完整表结构交给模型，还会增加上下文长度和泄密风险。

我更倾向于把数据库查询收进几个业务工具里：

用户问题 ↓ AI Agent：判断需要调用哪个工具 ↓ MCP Server：校验参数、身份和权限 ↓ 固定 SQL + 参数化查询 + 只读账号 ↓ 裁剪后的结构化结果

比如允许 Agent 调用search_orders，但不提供execute_sql。查哪些字段、最多返回多少条，都写死在工具内部。模型负责选择工具，不负责决定数据库边界。

MCP 放在这条链路的什么位置

MCP（Model Context Protocol）可以理解为一套面向 AI 应用的标准接口协议。MCP Server 可以向不同的 AI 客户端暴露：

• Tools：允许 Agent 调用的操作；
• Resources：客户端可以读取的上下文资源；
• Prompts：可复用的提示模板。

下面只用 Tool。它和普通 REST API 并不冲突：业务接口仍然可以保留，MCP Server 负责把合适的能力整理成模型容易理解的工具定义。

这里有个版本坑。现在pip install mcp安装的是 v2，服务类叫MCPServer；不少搜索结果还是 v1 的FastMCP写法，混用以后会直接卡在导入阶段。

把项目跑起来

目录

mcp-order-server/ ├─ server.py ├─ requirements.txt └─ .env.example

依赖

requirements.txt：

mcp[cli]>=2,<3 SQLAlchemy>=2,<3 pyodbc>=5,<6 pydantic>=2,<3

安装：

python-mvenv .venv# Windows.venv\Scripts\activate pipinstall-rrequirements.txt

本机还需要安装 Microsoft ODBC Driver 18 for SQL Server。

连接字符串

.env.example：

SQLSERVER_URL=mssql+pyodbc://mcp_reader:请替换密码@127.0.0.1:1433/OrderDb?driver=ODBC+Driver+18+for+SQL+Server&TrustServerCertificate=yes MCP_TENANT_ID=tenant_demo

实际运行时使用环境变量，不要把密码提交到 Git 仓库：

$env:SQLSERVER_URL="mssql+pyodbc://mcp_reader:***@127.0.0.1:1433/OrderDb?driver=ODBC+Driver+18+for+SQL+Server&TrustServerCertificate=yes"$env:MCP_TENANT_ID="tenant_demo"

数据库权限先收紧

不要让 MCP Server 复用管理员账号，也不要图省事授予整库读取权限。单独建只读登录，只开放 Agent 确实要用的视图。

USE[OrderDb];GOCREATEUSER[mcp_reader]FORLOGIN[mcp_reader];GOGRANTSELECTONOBJECT::dbo.v_AgentOrderSummaryTO[mcp_reader];GO

再建一个专用视图，把手机号、身份证、密码散列这类字段挡在视图外面：

CREATEVIEWdbo.v_AgentOrderSummaryASSELECTo.TenantId,o.OrderId,o.OrderNo,o.CustomerName,o.OrderStatus,o.TotalAmount,o.CreatedAtFROMdbo.OrdersASoWHEREo.IsDeleted=0;GO

这样即使以后有人改了工具代码，也还有数据库权限兜底：

• MCP 代码只查询规定的视图；
• 数据库账号本身也只能读取这个视图。

应用层限制和数据库授权最好都保留。只做其中一层，时间一久很容易被后续改动绕过去。

写 MCP Server

下面的server.py可以直接改。几个看似啰嗦的限制——关键字长度、状态枚举、返回数量——都是故意加上的。SQL 也全部走参数，不拼接用户输入。

importloggingimportosimporttimefromdatetimeimportdate,datetimefromdecimalimportDecimalfromtypingimportAnnotated,Literalfrommcp.serverimportMCPServerfrompydanticimportFieldfromsqlalchemyimportcreate_engine,event,text logging.basicConfig(level=logging.INFO,format="%(asctime)s %(levelname)s %(message)s",)logger=logging.getLogger("order-mcp")database_url=os.environ["SQLSERVER_URL"]# Demo 用环境变量固定租户。# 生产环境应从经过验证的访问令牌中读取 tenant_id，不能让模型传入。tenant_id=os.environ["MCP_TENANT_ID"]engine=create_engine(database_url,pool_pre_ping=True,pool_size=5,max_overflow=5,)@event.listens_for(engine,"connect")defset_query_timeout(dbapi_connection,_connection_record)->None:# pyodbc 的 connection.timeout 表示查询超时秒数。dbapi_connection.timeout=5mcp=MCPServer("Safe Order Query Server")defjson_value(value):"""把数据库类型转换为适合 MCP 传输的 JSON 值。"""ifisinstance(value,(datetime,date)):returnvalue.isoformat()ifisinstance(value,Decimal):returnfloat(value)returnvaluedefrow_to_dict(row)->dict:return{key:json_value(value)forkey,valueinrow._mapping.items()}@mcp.tool()defsearch_orders(customer_keyword:Annotated[str,Field(max_length=50,description="客户名称关键字；不需要按客户筛选时传空字符串",),]="",status:Literal["ALL","PENDING","PAID","SHIPPED","CLOSED"]="ALL",limit:Annotated[int,Field(ge=1,le=50)]=20,)->dict:"""查询当前租户的订单摘要，最多返回 50 条，不包含手机号等敏感字段。"""started_at=time.perf_counter()sql=text(""" SELECT TOP (:limit) OrderId, OrderNo, CustomerName, OrderStatus, TotalAmount, CreatedAt FROM dbo.v_AgentOrderSummary WHERE TenantId = :tenant_id AND (:customer_keyword = '' OR CustomerName LIKE :customer_pattern) AND (:status = 'ALL' OR OrderStatus = :status) ORDER BY CreatedAt DESC, OrderId DESC """)params={"limit":limit,"tenant_id":tenant_id,"customer_keyword":customer_keyword,"customer_pattern":f"%{customer_keyword}%","status":status,}try:withengine.connect()asconnection:rows=connection.execute(sql,params).fetchall()elapsed_ms=round((time.perf_counter()-started_at)*1000,2)logger.info("tool=search_orders tenant=%s status=%s limit=%s rows=%s elapsed_ms=%s",tenant_id,status,limit,len(rows),elapsed_ms,)return{"count":len(rows),"items":[row_to_dict(row)forrowinrows],"truncated":len(rows)==limit,}exceptException:# 详细异常只进入服务端日志，不把连接信息和 SQL 细节返回给模型。logger.exception("tool=search_orders failed tenant=%s",tenant_id)raiseRuntimeError("订单查询暂时失败，请稍后重试")@mcp.tool()defget_order_status(order_no:Annotated[str,Field(min_length=6,max_length=32,pattern=r"^[A-Za-z0-9_-]+$"),],)->dict:"""根据订单号查询当前租户的一条订单状态。"""sql=text(""" SELECT TOP (1) OrderNo, OrderStatus, TotalAmount, CreatedAt FROM dbo.v_AgentOrderSummary WHERE TenantId = :tenant_id AND OrderNo = :order_no """)withengine.connect()asconnection:row=connection.execute(sql,{"tenant_id":tenant_id,"order_no":order_no},).fetchone()ifrowisNone:return{"found":False,"order":None}return{"found":True,"order":row_to_dict(row)}if__name__=="__main__":# 本地可使用 stdio；远程部署推荐 streamable-http。mcp.run("streamable-http")

先在本地把边界测出来

启动开发工具：

mcp dev server.py

也可以直接启动 Streamable HTTP 服务：

python server.py

默认 MCP 地址为：

http://127.0.0.1:8000/mcp

在 MCP Inspector 中可以先调用：

{"customer_keyword":"张","status":"PAID","limit":10}

别只测正常查询，下面几种输入更值得试：

1. 正常查询能否返回结构化数据；
2. limit=1000是否会在进入数据库前被拒绝；
3. 非法状态值是否会被 Schema 校验拒绝；
4. 客户关键字中带单引号时，是否仍按普通参数处理。

接到 Agent

下面以 OpenAI Agents SDK 为例连接远程 MCP Server：

importasyncioimportosfromagentsimportAgent,Runnerfromagents.mcpimportMCPServerStreamableHttp,create_static_tool_filterasyncdefmain()->None:asyncwithMCPServerStreamableHttp(name="Order MCP",params={"url":"http://127.0.0.1:8000/mcp","headers":{"Authorization":f"Bearer{os.environ['MCP_SERVER_TOKEN']}"},"timeout":10,},cache_tools_list=True,max_retry_attempts=2,tool_filter=create_static_tool_filter(allowed_tool_names=["search_orders","get_order_status"]),require_approval="never",)asserver:agent=Agent(name="订单助手",instructions=("你只能依据工具返回的数据回答订单问题。""查询结果为空时直接说明没有找到，不要猜测订单状态。"),mcp_servers=[server],)result=awaitRunner.run(agent,"查询张姓客户最近已支付的 10 个订单")print(result.final_output)asyncio.run(main())

tool_filter只是让当前 Agent 少看到一些无关工具。我把它当作防误用配置，不把它当权限系统。真正的鉴权仍在 MCP Server 和数据库里做。

上线前容易漏掉的地方

租户参数不要暴露给模型

不要设计成下面这样：

defsearch_orders(tenant_id:str,...):...

因为tenant_id会暴露在 Tool Schema 中，模型能够自行填写。正确做法是从 OAuth 访问令牌、网关签名或可信运行上下文中取得租户与用户身份，然后在服务端强制追加租户条件。

“只允许 SELECT”没有想象中安全

SQL 必须以 SELECT 开头并不等于安全。复杂子查询、跨库访问、系统函数和高消耗查询仍然可能产生风险。优先提供业务级工具，而不是execute_sql(sql: str)。

返回值限量

我一般会同时卡住这些指标：

• 最大行数；
• 最大字段数；
• 最大单字段长度；
• 最大响应体积；
• 是否允许返回敏感字段；
• 查询超时时间。

日志要能查清一次调用

至少记录：

trace_id、user_id、tenant_id、tool_name、参数摘要、结果行数、耗时、状态、错误码

手机号、证件号、Token、数据库密码等内容必须脱敏或完全不记录。

远程服务别裸奔

MCP 官方安全文档建议远程服务使用 OAuth 2.1，并按 Tool 或能力拆分 Scope。生产部署还应配置 TLS、允许的 Host/Origin、限流以及密钥轮换。

发版前注意

我会先拿数据库账号单独登录一次，确认它只能查询指定视图，直接访问原表会被拒绝。随后绕过 Agent 直接调用 MCP Tool，测试超长参数、非法状态、单引号、空结果和超过上限的limit。这一步能把“Prompt 看起来写得很严”造成的错觉去掉。

最后再查网络侧配置：远程地址是否强制 TLS，Token 是否放在请求头，Host/Origin 白名单和限流是否生效。日志里应该能按trace_id找到调用人、租户、工具名、耗时和行数，但不能搜到数据库密码、Token、手机号等原始敏感信息。

最后

这套方案会比“模型生成 SQL 后直接执行”多写一些代码，但边界很清楚：模型只负责选工具和填业务参数，SQL、租户条件、可见字段以及返回上限都由后端掌握。

如果现有系统已经有订单查询 API，也没必要为了 MCP 再写一套数据库访问层。让 MCP Server 调已有 API，同样能达到目的。关键在于不要把原本由后端控制的权限和数据范围交给模型临场判断。

下一篇接着处理工具上线后的几个麻烦：接口超时了要不要重试，重复调用怎么去重，以及权限到底应该放在哪一层。