OpenClaw技能安全执行MySQL增删改查：让大模型只懂抽参，执行器守护数据库

简介：针对OpenClaw官方技能库中暂无现成MySQL-CRUD技能的现状，这份资源面向需要自行扩展数据库操作能力的开发者，演示了基于Python编程语言与常用MySQL驱动（如pymysql）封装数据库技能的实现路径。内容围绕增删改查四类核心操作展开，覆盖数据库连接配置、参数化查询防注入、异常捕获与事务处理等关键环节，并给出技能输入输出格式的封装示例，方便直接嵌入OpenClaw框架调用。压缩包共3个文件，包含Python脚本、Markdown说明文档和HTML演示页面，整体仅3KB，轻量但结构完整，适合快速阅读后二次开发。已有165人学习下载，可帮助开发者避开常见的配置与调试弯路，快速掌握OpenClaw自定义数据操作技能的设计思路。

1. 让 OpenClaw 技能直接操作 MySQL：一次把「能聊」变成「能干」

OpenClaw 接入 MySQL 做增删改查，第一反应通常是让模型直接写 SQL 然后执行——我劝你千万别这么干。模型写的 SQL 没经过参数校验、没有 LIMIT、甚至可能漏掉 DELETE 的 WHERE 条件，生产库经不起一次这样的翻车。OpenClaw 的技能（Skill）机制恰好把这个口子堵住了：大模型只负责把用户的自然语言转成结构化参数，真正执行增删改查的是一个你写好的 Python 执行器。这篇文章把整套技能从文件结构、执行器代码到避坑清单完整拆解，适合已经部署了 OpenClaw、想让它帮你查库、改数据、干实际活的工程师照着复现。

2. 先拆 OpenClaw 的 Skill 机制：技能文件、参数触发与最小执行器

2.1 Skill 在 OpenClaw 里是什么：从提示词模板到可执行工具的边界

在使用 OpenClaw 的过程中你会发现，它默认能聊、能总结、能调用一些内置工具，但要让它做一件具体的业务操作，比如「查一下订单表里昨天有多少条记录」，就必须把这件事封装成一个技能。技能的本质是两部分的组合：一份描述文件，告诉大模型这个工具在什么场景下用、需要哪些参数、参数长什么样；一个执行脚本，真正连数据库、跑 SQL、返回结果。

关键边界在于：大模型永远不直接执行代码。它只负责从用户的话里抽参数，比如用户说「把 id 为 5 的用户状态改成禁用」，模型要抽出来的是table=users、set_clause=status='disabled'、where_clause=id=5，然后把这三个参数交给执行器。这样即使有人通过提示注入诱导模型写出DROP TABLE这种语句，执行器层面也可以直接拒绝，不会真跑到数据库里去。

2.2 最小技能包：一个目录、三个文件

OpenClaw 的技能通常就是一个独立目录，放进技能目录后由框架动态加载。我一般会这样组织：

mysql_crud/ ├── SKILL.md # 技能描述 + 参数 schema ├── handler.py # 增删改查执行器 └── requirements.txt # pymysql / sqlalchemy 等依赖

SKILL.md是整个技能的「说明书」，大模型靠它来判断什么时候使用这个技能、参数怎么填。handler.py是真正干活的脚本，它收到模型抽取的参数后，完成数据库连接、SQL 拼接、执行、返回结构化结果。requirements.txt声明运行时依赖，部署到新环境时pip install -r requirements.txt一条命令搞定。

这种最小结构的好处是把「模型的判断」和「代码的执行」彻底分开，各自演进互不干扰。模型侧的 prompt 调优只改 SKILL.md，数据库侧的连接策略只改 handler.py，两者不需要同时改。

2.3 参数定义与触发逻辑：让模型知道什么时候该出手

SKILL.md里的参数 schema 决定了模型能从自然语言里抽出什么。下面是一个可用的最小描述文件：

name: mysql_crud description: 当用户需要查询、插入、更新或删除 MySQL 数据库中的记录时，使用此技能。 parameters: type: object properties: query_type: type: string enum: [SELECT, INSERT, UPDATE, DELETE] description: 要执行的操作类型 table_name: type: string description: 目标表名 data: type: object description: INSERT 时插入的字段键值对，或 UPDATE 时待更新的字段键值对 where_clause: type: object description: 查询或更新条件，例如 {"id": 5} limit: type: integer description: SELECT 返回的最大行数，默认 50 required: [query_type, table_name]

这段配置的逻辑是：模型先判断用户意图属于增删改查中的哪一种，填入query_type；再从用户话里提取表名和条件。limit是可选的，但我会在描述里明确要求模型对查询类操作带上这个参数，防止一次查出几十万行把上下文撑爆。触发条件写在description里，只有当用户请求涉及数据库操作时，模型才会调用这个技能，普通闲聊不会触发。

参数抽好了，接下来就看执行器怎么接住这些参数并安全地跑起来，这是下一章的重点。

3. 封装 MySQL 增删改查：连接方式、执行器与技能注册一条龙

3.1 先选连接方式：直连、连接池还是走网关

让执行器连 MySQL，常见的有三种方式：用 PyMySQL 直连、用 SQLAlchemy 管理连接池、或者通过 HTTP 网关间接访问。三者的区别直接决定了技能的可靠性和并发能力，我做了个对比：

我一般推荐 SQLAlchemy。原因很实际：OpenClaw 跑起来之后可能会有多个会话同时触发同一个技能，直连模式在并发稍高时就会出现Too many connections报错，而连接池可以让连接数稳定在一个可控范围内。如果你的 OpenClaw 部署在 Docker 容器里，MySQL 跑在宿主机上，记得连接串里的 host 要写宿主机在容器网络中的可达地址；反过来 MySQL 跑在容器里，也要确认端口映射到了宿主机的哪个端口。

3.2 写一个带超时、重试和关闭动作的执行器

执行器的核心职责是：接参数、拼 SQL、执行、返回结构化结果。下面这个handler.py是我在实际环境里调过的一版，包含连接池、超时、重试和显式关闭连接：

import os import json import time from sqlalchemy import create_engine, text from sqlalchemy.pool import QueuePool DB_USER = os.getenv("MYSQL_USER", "claw_bot") DB_PASSWORD = os.getenv("MYSQL_PASSWORD", "") DB_HOST = os.getenv("MYSQL_HOST", "127.0.0.1") DB_PORT = os.getenv("MYSQL_PORT", "3306") DB_NAME = os.getenv("MYSQL_DB", "app_db") engine = create_engine( f"mysql+pymysql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}?charset=utf8mb4", poolclass=QueuePool, pool_size=5, max_overflow=3, pool_recycle=3600, pool_pre_ping=True, connect_args={"connect_timeout": 5}, ) def execute(params: dict) -> dict: query_type = params.get("query_type", "").upper() table = params.get("table_name", "") if query_type not in ("SELECT", "INSERT", "UPDATE", "DELETE"): return {"ok": False, "error": f"unsupported query_type: {query_type}"} data = params.get("data") or {} where = params.get("where_clause") or {} limit = int(params.get("limit") or 50) # 危险操作拦截：UPDATE/DELETE 必须有 where 条件 if query_type in ("UPDATE", "DELETE") and not where: return {"ok": False, "error": "UPDATE/DELETE must have where_clause"} sql, bind = build_sql(query_type, table, data, where, limit) for attempt in range(2): try: with engine.connect() as conn: result = conn.execute(text(sql), bind) if query_type == "SELECT": rows = [dict(r) for r in result.fetchmany(limit)] return {"ok": True, "rows": rows, "count": len(rows)} conn.commit() return {"ok": True, "affected_rows": result.rowcount} except Exception as e: if attempt == 1: return {"ok": False, "error": str(e)} time.sleep(0.3)

逻辑说明：执行器首先对query_type做枚举校验，避免模型传进来一个奇怪的字符串；然后对 UPDATE 和 DELETE 做强制条件校验——没有where_clause直接拒绝，这一步是整条安全链的底线。build_sql函数负责根据操作类型拼接带绑定参数的 SQL，绑定参数用:key占位而不是字符串拼接，既防注入又能让 MySQL 走预编译缓存。重试逻辑只做一次，间隔 0.3 秒，主要应对网络抖动造成的瞬时报错；连接池的pool_pre_ping=True会在取连接前先测活，避免拿到失效连接。

参数说明：pool_size=5表示连接池保持 5 个连接，max_overflow=3允许高峰期再多创建 3 个，这样最多 8 个连接，对绝大多数小团队够用。pool_recycle=3600强制连接每小时的回收，规避 MySQL 的wait_timeout把空闲连接断掉的问题。connect_timeout=5让技能在数据库不可达时快速失败返回错误信息，而不是让大模型干等。

3.3 四个核心操作如何映射到参数：每个操作的返回语义

增删改查不是简单地把参数拼进 SQL，而是每一类操作都有明确的语义约定，模型才能把结果转述给用户：

• SELECT：返回rows列表，每行是字段名到值的字典。返回行数受limit约束，默认 50 行，超出部分在描述里明确告知模型「结果可能不完整」。
• INSERT：返回affected_rows，值为 1 表示插入成功；需要拿到自增 ID 的话，可以在执行器里加result.lastrowid一并返回。
• UPDATE：返回affected_rows，这个数字对用户很有意义——更新了 0 行说明条件可能写错了。
• DELETE：返回affected_rows，同一张表的删除操作我会建议模型在执行前先跑一个 SELECT 确认影响范围。

这四种操作的共同原则是：执行器只返回「结构化的事实」，不返回夸大的成功描述。比如 UPDATE 影响了 0 行，执行器就如实返回 0，模型会把「没有匹配到记录」告诉用户，而不是含糊地回一句「已更新」。

3.4 把技能注册到 OpenClaw：配置目录和触发测试

技能文件写好后，放进 OpenClaw 配置指定的技能目录，然后在配置文件里声明：

skills: enabled: true dir: ./skills

把mysql_crud/整个目录放进./skills/下，重新加载配置或重启 OpenClaw 进程即可。Windows 上部署的话，路径用绝对路径更省心，比如dir: D:/openclaw/skills，注意反斜杠要转义或直接用正斜杠。验证是否加载成功有个笨但有效的办法：在对话里直接问一句「帮我查一下 users 表有多少条记录」，然后看执行器的日志有没有对应输出。

如果返回的是「我没有权限执行」或者模型答非所问，先检查技能目录名是否和 SKILL.md 里的name一致，再确认 OpenClaw 进程有权限读 handler.py。这套流程跑通之后，增删改查就真的交给 OpenClaw 了，但下一步要面对的是各种意想不到的坑。

4. 避坑：OpenClaw 操作 MySQL 的 5 个常见翻车点

4.1 模型生成 SQL 跑偏：把 DROP 表当查询做

现象：用户问「这个表没用了删掉吧」，模型真的调用了技能，执行器如果没拦截，表就没了。

原因：大模型对「删除」的理解是语义级的，它区分不清「删除某条记录」和「删除整张表」在 SQL 里的天壤之别。

解决：在执行器里加一条硬校验——table_name必须来自一个白名单，执行前检查SHOW TABLES的结果；同时用正则禁止 SQL 中出现DROP、ALTER、TRUNCATE、CREATE这些非白名单关键词。就算模型抽参数抽错了，执行器也不会放行。

4.2 连不上 Docker 容器里的 MySQL

现象：技能执行时报Can't connect to MySQL server on '127.0.0.1'，但 MySQL 容器明明在跑。

原因：MySQL 跑在 Docker 容器里时，3306 端口没有映射到宿主机，或者映射到了别的宿主端口。还有一个隐蔽坑：MySQL 用户授权只给了容器网段的 IP，宿主机连接时被拒绝。

解决：启动容器时加-p 3306:3306映射；进入容器用SELECT user, host FROM mysql.user确认claw_bot账号的 host 是'%'或宿主机可达的 IP 段，然后GRANT ... TO 'claw_bot'@'%'并FLUSH PRIVILEGES。

4.3 Windows 下 MySQL 服务没起来导致技能假死

现象：OpenClaw 在 Windows 上部署好后，技能调用超时，日志里只有连接超时没有具体错误。

原因：Windows 上 MySQL 安装后服务默认是手动启动，机器重启后服务没跟着起来；还有一种情况是之前net start mysql启动的是 5.x 服务，但连接串上写的端口还是 3306，实际新版占了别的端口。

解决：先在命令行确认服务和端口：net start | findstr mysql和mysql -h 127.0.0.1 -P 3306 -u claw_bot -p手动连通后再让技能重试。连接串里 host 写127.0.0.1而不是localhost，后者在部分 Windows 环境会走命名管道导致 PyMySQL 连不上。

4.4 中文乱码和 emoji 写入报错

现象：SELECT 查出来的中文是???，INSERT 带 emoji 直接报Incorrect string value。

原因：连接串没指定字符集，MySQL 会话默认用了 latin1；或者表本身是utf8，而 utf8 在 MySQL 里最多存 3 字节，存不下 emoji 这类 4 字节字符。

解决：连接串统一加?charset=utf8mb4，建表时用DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci。修改已有表用ALTER TABLE ... CONVERT TO CHARACTER SET utf8mb4，只改连接串不改表结构，写入依然会报错。

4.5 长事务和行锁等待把技能卡死

现象：技能第一次调用 UPDATE 很慢，第二次调用直接报Lock wait timeout exceeded。

原因：第一次的 UPDATE 条件没走索引，扫了全表，把大量行锁住了；事务没提交，锁一直不释放，后续会话全部堵住。

解决：在技能描述里强制要求模型对 UPDATE/DELETE 先做 SELECT 预览：影响行数超过 50 就要求用户确认后再执行；MySQL 侧设置innodb_lock_wait_timeout=10，让它快速失败而不是无限等锁。

5. 从能跑变好用：最小权限、事务兜底与操作审计

5.1 给技能开一个专用账号，别用 root

OpenClaw 技能连 MySQL，最忌讳用 root 账号，一旦模型抽参抽错或者提示注入生效，代价是整个库。我一般会给技能单独建一个账号，权限精确到库和操作类型：

CREATE USER 'claw_bot'@'localhost' IDENTIFIED BY '替换成强密码'; GRANT SELECT, INSERT, UPDATE, DELETE ON app_db.* TO 'claw_bot'@'localhost'; FLUSH PRIVILEGES;

逻辑说明：app_db.*把权限限制在单个业务库内，技能连不到其他库；只授权 SELECT、INSERT、UPDATE、DELETE 四个 DML 操作，没有 DDL 权限，即使执行器被绕过，也做不了DROP TABLE。@'localhost'限定了来源 IP，如果 OpenClaw 在另一台机器上，改成对应 IP 或网段。

参数说明：不要把GRANT ALL PRIVILEGES ON *.*给这个账号，这类授权是给 DBA 用的，不是给智能体用的。改完权限后记得FLUSH PRIVILEGES，否则部分环境不会立即生效。

5.2 事务兜底：技能调用要么全成功，要么全回滚

执行器里的事务策略是：每次调用开启一个短事务，执行成功立即提交，任何一条异常都回滚并返回错误信息。engine.connect()默认开启事务，所以commit()和rollback()的位置很重要：

for attempt in range(2): try: with engine.begin() as conn: result = conn.execute(text(sql), bind) if query_type == "SELECT": rows = [dict(r) for r in result.fetchmany(limit)] return {"ok": True, "rows": rows, "count": len(rows)} return {"ok": True, "affected_rows": result.rowcount} except Exception as e: if attempt == 1: return {"ok": False, "error": str(e)} time.sleep(0.3)

逻辑说明：engine.begin()是 SQLAlchemy 推荐的事务写法，进入with块自动开启事务，正常走完自动提交，抛异常自动回滚，不需要手动commit()/rollback()，少一个「忘了回滚」的隐患。重试只针对连接层的瞬时报错，事务一旦开始执行并报了业务错误，不会重试，避免把重复数据写进去。

5.3 结果人话化：让模型拿到能看懂的事实

执行器返回的裸结果对用户不友好。SELECT 返回 50 行原始字典，模型转述时会非常吃力，所以执行器要做一层格式化：

def format_result(query_type: str, result: dict) -> dict: if query_type != "SELECT": return result rows = result.get("rows", []) if len(rows) > 10: return { "ok": True, "message": f"共查到 {result['count']} 行，仅展示前 10 行", "preview": rows[:10], } return {"ok": True, "rows": rows, "count": len(rows)}

逻辑说明：超过 10 行时不再把全部数据塞进模型上下文，而是只给前 10 行预览和总数，让模型告诉用户「数据太多，已展示前 10 条」。这样既避免上下文被撑爆，也让模型的行为更可靠——用户如果想看更多，会明确要求增加 limit，而不是让模型在截断的数据上瞎猜结论。

5.4 审计日志：每次操作都要留痕

技能放给团队用之后，审计日志就是后悔药。执行器每执行完一个操作，往本地日志文件追加一条 JSON：

def audit_log(skill_name, params, sql, result, elapsed_ms): entry = { "time": time.strftime("%Y-%m-%d %H:%M:%S"), "skill": skill_name, "query_type": params.get("query_type"), "table": params.get("table_name"), "sql": sql, "ok": result.get("ok"), "elapsed_ms": elapsed_ms, } with open("openclaw_audit.log", "a", encoding="utf-8") as f: f.write(json.dumps(entry, ensure_ascii=False) + "\n")

逻辑说明：日志记录的是执行层的客观事实，不记录完整参数里的敏感数据，这样排查问题时能回答「这个技能在什么时间做了什么操作」，又不会把大批量导出的数据明文落盘。每次操作都记录行数而不是完整数据内容，是日志安全和可观测性之间平衡的做法。

6. 先跑通 SELECT 再放开写操作：渐进开放与验证清单

技能上线最稳妥的路径是分阶段放开。第一步只开放 SELECT，让 OpenClaw 先成为一个能查库的问答工具；跑几天确认模型抽参稳定、没有误触发之后，再开放 INSERT；UPDATE 和 DELETE 放到最后，且必须保留执行器里的无条件拦截。每个阶段用固定的验证问题集去压一遍：查不存在的表、查空表、插入重复主键、更新不存在的记录、删除带条件的记录，每一条都要看执行器返回的错误是不是能被模型转述成用户能懂的话。

我自己的血泪教训是：当初觉得 SELECT 跑得很稳，直接把 UPDATE 一起放开了，结果一条「把这个用户的余额清零」的指令，因为where_clause被模型抽成了{"id": "unknown"}，加上测试库里恰好有三条脏数据带这个 id，一次改了三条。从那以后，执行器里加了一条硬规矩：UPDATE 和 DELETE 的where_clause必须包含主键字段，否则拒绝执行。不是所有业务场景都适用这条，但它在绝大多数情况下能拦住最危险的那类误操作。

验证清单可以按这个顺序过一遍：第一，危险指令测试——故意问「把 users 表清空」「把所有商品价格改成 0」，看技能怎么反应；第二，并发测试——同时开三个会话查询同一张表，看连接池是否稳定；第三，可用性测试——停掉 MySQL 再启动，看技能能否在连接池重建后自动恢复。这三轮过完，基本可以放心交给业务方用了。

如果你只记住一个结论，那就是：OpenClaw 的技能机制本身不负责安全，安全全写在你的执行器里。参数白名单、无条件拦截、影响行数限制、审计日志，缺一条都可能在未来某个时刻翻车。这套方案我已经跑了几个季度，前期多花半天把执行器做扎实，后面维护成本会低到你几乎感受不到它的存在。希望帮到你。

本文还有配套的精品资源，点击获取