AI Agent 技能分享从零实现 MCP Server让 Agent 安全读取数据库把数据库接给 Agent最省事的办法是让模型生成 SQL然后直接执行。这个方案做演示很快我不建议照搬到业务系统里。数据库里装的是实打实的业务数据模型偶尔选错表、漏掉租户条件代价可能比一次回答错误大得多。这篇用 MCP Python SDK v2、SQLAlchemy 和 SQL Server 做一个订单查询服务。Agent 只能调用事先定义好的订单工具拿不到任意 SQL 的执行入口。先说说直接执行 SQL 的问题很多 Text-to-SQL 示例采用下面这条链路用户问题 → 大模型生成 SQL → 数据库执行 SQL → 返回结果链路很短风险却不少模型可能生成UPDATE、DELETE或高消耗查询用户可能通过提示词诱导模型查询无权访问的数据即使只允许SELECT仍可能出现跨租户查询、敏感字段泄露和全表扫描SQL 语法检查无法判断一条查询在业务上是否越权将完整表结构交给模型还会增加上下文长度和泄密风险。我更倾向于把数据库查询收进几个业务工具里用户问题 ↓ AI Agent判断需要调用哪个工具 ↓ MCP Server校验参数、身份和权限 ↓ 固定 SQL 参数化查询 只读账号 ↓ 裁剪后的结构化结果比如允许 Agent 调用search_orders但不提供execute_sql。查哪些字段、最多返回多少条都写死在工具内部。模型负责选择工具不负责决定数据库边界。MCP 放在这条链路的什么位置MCPModel 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.txtmcp[cli]2,3 SQLAlchemy2,3 pyodbc5,6 pydantic2,3安装python-mvenv .venv# Windows.venv\Scripts\activate pipinstall-rrequirements.txt本机还需要安装 Microsoft ODBC Driver 18 for SQL Server。连接字符串.env.exampleSQLSERVER_URLmssqlpyodbc://mcp_reader:请替换密码127.0.0.1:1433/OrderDb?driverODBCDriver18forSQLServerTrustServerCertificateyes MCP_TENANT_IDtenant_demo实际运行时使用环境变量不要把密码提交到 Git 仓库$env:SQLSERVER_URLmssqlpyodbc://mcp_reader:***127.0.0.1:1433/OrderDb?driverODBCDriver18forSQLServerTrustServerCertificateyes$env:MCP_TENANT_IDtenant_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.IsDeleted0;GO这样即使以后有人改了工具代码也还有数据库权限兜底MCP 代码只查询规定的视图数据库账号本身也只能读取这个视图。应用层限制和数据库授权最好都保留。只做其中一层时间一久很容易被后续改动绕过去。写 MCP Server下面的server.py可以直接改。几个看似啰嗦的限制——关键字长度、状态枚举、返回数量——都是故意加上的。SQL 也全部走参数不拼接用户输入。importloggingimportosimporttimefromdatetimeimportdate,datetimefromdecimalimportDecimalfromtypingimportAnnotated,Literalfrommcp.serverimportMCPServerfrompydanticimportFieldfromsqlalchemyimportcreate_engine,event,text logging.basicConfig(levellogging.INFO,format%(asctime)s %(levelname)s %(message)s,)loggerlogging.getLogger(order-mcp)database_urlos.environ[SQLSERVER_URL]# Demo 用环境变量固定租户。# 生产环境应从经过验证的访问令牌中读取 tenant_id不能让模型传入。tenant_idos.environ[MCP_TENANT_ID]enginecreate_engine(database_url,pool_pre_pingTrue,pool_size5,max_overflow5,)event.listens_for(engine,connect)defset_query_timeout(dbapi_connection,_connection_record)-None:# pyodbc 的 connection.timeout 表示查询超时秒数。dbapi_connection.timeout5mcpMCPServer(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_length50,description客户名称关键字不需要按客户筛选时传空字符串,),],status:Literal[ALL,PENDING,PAID,SHIPPED,CLOSED]ALL,limit:Annotated[int,Field(ge1,le50)]20,)-dict:查询当前租户的订单摘要最多返回 50 条不包含手机号等敏感字段。started_attime.perf_counter()sqltext( 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:rowsconnection.execute(sql,params).fetchall()elapsed_msround((time.perf_counter()-started_at)*1000,2)logger.info(toolsearch_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(toolsearch_orders failed tenant%s,tenant_id)raiseRuntimeError(订单查询暂时失败请稍后重试)mcp.tool()defget_order_status(order_no:Annotated[str,Field(min_length6,max_length32,patternr^[A-Za-z0-9_-]$),],)-dict:根据订单号查询当前租户的一条订单状态。sqltext( SELECT TOP (1) OrderNo, OrderStatus, TotalAmount, CreatedAt FROM dbo.v_AgentOrderSummary WHERE TenantId :tenant_id AND OrderNo :order_no )withengine.connect()asconnection:rowconnection.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}别只测正常查询下面几种输入更值得试正常查询能否返回结构化数据limit1000是否会在进入数据库前被拒绝非法状态值是否会被 Schema 校验拒绝客户关键字中带单引号时是否仍按普通参数处理。接到 Agent下面以 OpenAI Agents SDK 为例连接远程 MCP ServerimportasyncioimportosfromagentsimportAgent,Runnerfromagents.mcpimportMCPServerStreamableHttp,create_static_tool_filterasyncdefmain()-None:asyncwithMCPServerStreamableHttp(nameOrder MCP,params{url:http://127.0.0.1:8000/mcp,headers:{Authorization:fBearer{os.environ[MCP_SERVER_TOKEN]}},timeout:10,},cache_tools_listTrue,max_retry_attempts2,tool_filtercreate_static_tool_filter(allowed_tool_names[search_orders,get_order_status]),require_approvalnever,)asserver:agentAgent(name订单助手,instructions(你只能依据工具返回的数据回答订单问题。查询结果为空时直接说明没有找到不要猜测订单状态。),mcp_servers[server],)resultawaitRunner.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 看起来写得很严”造成的错觉去掉。最后再查网络侧配置远程地址是否强制 TLSToken 是否放在请求头Host/Origin 白名单和限流是否生效。日志里应该能按trace_id找到调用人、租户、工具名、耗时和行数但不能搜到数据库密码、Token、手机号等原始敏感信息。最后这套方案会比“模型生成 SQL 后直接执行”多写一些代码但边界很清楚模型只负责选工具和填业务参数SQL、租户条件、可见字段以及返回上限都由后端掌握。如果现有系统已经有订单查询 API也没必要为了 MCP 再写一套数据库访问层。让 MCP Server 调已有 API同样能达到目的。关键在于不要把原本由后端控制的权限和数据范围交给模型临场判断。下一篇接着处理工具上线后的几个麻烦接口超时了要不要重试重复调用怎么去重以及权限到底应该放在哪一层。