CoPaw Skill机制解析：为Agent封装可控的数据库查询能力

最近在给一个内部项目搭 AI Agent 能力时我把注意力重点放在了 CoPaw 的 Skill 机制上。说白了CoPaw Skill 就是给 Agent 装上的“技能插件”让原本只会聊天的模型能真正执行一些具体任务。我们团队第一个落地的场景就是开发一个从关系数据库中查询数据的 Skill把 Agent 和 MySQL/PostgreSQL 这类传统数据源打通。这个选题其实很典型。很多人在玩 Agent 时都会遇到同一个问题模型再聪明也没法直接摸到你的业务数据库甚至不让它直接摸才是对的。把“查数据库”这件事做成一个受控的 Skill既能让 Agent 具备数据查询能力又能把 SQL 执行、连接管理、权限控制这些风险隔离在脚本层。这篇文章就围绕这个目标来写适合已经跑通 CoPaw 基础流程、想给 Agent 增加真实工具能力的开发者也适合正在纠结“Skill 到底怎么写、参数怎么设计”的朋友。1. 先想清楚为什么数据库查询要封装成 Skill1.1 Skill 和 Agent 到底什么关系网上很多人搜“skill 和 agent 的区别”其实这俩不是一个层级的玩意儿。Agent 是那个会思考、会规划、会决定下一步干什么的大脑Skill 是大脑可以调用的“肌肉记忆”是预先写好的、带明确输入输出约定的能力单元。我习惯这样理解Agent 是项目经理Skill 是执行某项具体任务的外包团队。项目经理不需要懂 SQL 怎么写它只需要知道“有一个查数据库的 Skill输入是表名和条件输出是结构化结果”然后根据用户的自然语言去调用。Skill 内部怎么连库、怎么拼 SQL、怎么处理超时Agent 一概不关心。这种解耦非常关键因为它让 Skill 可以独立测试、独立发布、独立加固。1.2 为什么不能让 Agent 直接连库理论上你也可以让 Agent 自己生成 SQL然后把 SQL 直接丢给数据库执行。但这么做等于让一个概率模型直接操作你的生产库风险非常大。模型生成的 SQL 可能有语法错误可能查全表导致数据库负载飙升更恐怖的是如果提示词被注入它可能生成DROP TABLE这种破坏性语句。所以行业的通用做法是把数据库访问收敛到固定的脚本里Agent 只能传参数不能执行任意 SQL。这样即使模型发疯脚本层的白名单和参数化校验也能兜底。这是我在设计 CoPaw Skill 时第一个确定的原则——Skill 是安全边界不是便利贴。1.3 相比普通 Function Calling 和 MCP 工具Skill 的优势在哪如果你用过 Function Calling会发现它和 Skill 有一部分重叠都是让模型调用外部工具。但 Skill 更强调“完整技能封装”——它不仅有调用入口还包含使用说明什么时候该用、参数怎么填、脚本实现、可能还有配套的资源文件。也就是说Skill 自带使用文档模型会先读 SKILL.md 里的描述再决定要不要调用、怎么调用。拿我们这个场景对比Function Calling 通常只给模型一个函数签名模型并不知道这个函数适合哪些查询场景而 CoPaw Skill 的 SKILL.md 里会写明“此 Skill 只负责单表查询、支持条件过滤和排序、不支持多表 JOIN 和写操作”模型读到这些约束后调用就会克制很多。这比裸的 Function Calling 更接近真正的“能力封装”。2. Skill 的骨架目录、元信息与调用约定2.1 最小 Skill 目录结构CoPaw Skill 目前没有强制统一的目录标准但社区里比较通行的做法是每个 Skill 一个独立目录里面至少包含一个SKILL.md文件和脚本目录。我个人的推荐结构是my-skills/ └── db-query/ ├── SKILL.md ├── scripts/ │ ├── query_db.py │ ├── requirements.txt │ └── .env.example └── assets/ └── sample_output.mdSKILL.md是技能的“说明书入口”CoPaw 加载 Skill 时首先读它scripts/存放实际执行逻辑的脚本一般用 Python生态成熟、驱动齐全.env.example记录需要注入的环境变量比如数据库连接串实际运行时通过环境变量注入避免把密码写死在代码里。这个结构不是拍脑袋定的。实际调试时你会发现如果把脚本和说明文档混在一起Agent 很容易读到一堆无关代码干扰模型对 Skill 的理解。分开之后SKILL.md保持纯净只描述技能能力边界和参数规范脚本内部细节一概不写模型反而调用得更准确。2.2 SKILL.md 里的 YAML frontmatter 怎么写SKILL.md的开头通常有一段 YAML 元信息CoPaw 靠它做技能索引。以下是我在 db-query Skill 里用的精简版--- name: db_query description: 用于从关系数据库中查询数据的技能。支持 MySQL、PostgreSQL仅限单表查询支持 WHERE 过滤、ORDER BY 排序、LIMIT 限制。不支持多表 JOIN、不支持 UPDATE/DELETE/DDL。 when_to_use: 当用户需要查询业务数据、统计数据、核查记录时使用。当用户要求修改或删除数据时明确拒绝并解释此 Skill 只读。 version: 1.0.0 ---description和when_to_use这两项是模型最关心的。写的时候我踩过一个坑一开始 description 写得太泛比如“用于查询数据库”结果模型在用户问“帮我算出上个月所有订单的总金额”时直接没有调用 Skill因为模型不确定这个 Skill 能不能做聚合统计。后来我在 description 里明确写了“支持 COUNT、SUM、AVG 等聚合函数”调用率一下子就上来了。所以给 Skill 写描述时宁可啰嗦不能模糊。你要站在模型的角度想它看到的是一堆 Skill 的 description它需要快速判断哪个 Skill 最适合当前用户意图。描述越具体匹配越准。2.3 参数怎么设计才不会被大模型乱传这是 Skill 开发里最容易被忽视、但实际最容易出问题的地方。模型不是人它填参数时不会像工程师一样仔细经常把字符串值加上多余引号或者漏传可选参数。所以参数设计要遵循几个原则第一参数全部走命名参数不搞位置参数。CoPaw 在调用脚本时一般会按约定的参数名传值所以脚本里用argparse定义清晰的--table、--where、--limit不要依赖sys.argv[1]这种位置取值。第二可选参数要有默认值且容错。比如--limit一定要有默认值 100否则模型一旦不传脚本可能返回全表数据直接把对话上下文撑爆。第三凡是模型可能填错的参数脚本要做二次校验。以--table为例模型可能传一个不存在的表名脚本应该在项目启动时加载一份表名白名单传入了白名单之外的直接报错返回而不是把错误 SQL 抛给数据库。这样既省了一次无效数据库连接也能防止模型被诱导去查mysql.user这种敏感表。3. 实操实现一个能从关系数据库查询数据的 Skill3.1 测试环境准备Docker 起一个 MySQL我不建议拿生产库来练手。实际开发 Skill 时我习惯先用 Docker 在本机起一个干净的 MySQL 实例在一张测试表上反复调试等稳定了再换真实数据源验证。docker run --name copaw-mysql -e MYSQL_ROOT_PASSWORDroot123 -e MYSQL_DATABASEdemo -p 3306:3306 -d mysql:8接着建一张orders表模拟电商订单场景CREATE TABLE orders ( id INT AUTO_INCREMENT PRIMARY KEY, order_no VARCHAR(32) NOT NULL, customer_name VARCHAR(64) NOT NULL, amount DECIMAL(10,2) NOT NULL, status TINYINT NOT NULL DEFAULT 0, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ); INSERT INTO orders (order_no, customer_name, amount, status, created_at) VALUES (A0001, 张三, 150.00, 1, 2025-01-05 10:00:00), (A0002, 李四, 2300.50, 2, 2025-01-08 14:30:00), (A0003, 王五, 89.90, 1, 2025-01-12 09:15:00), (A0004, 张三, 420.00, 3, 2025-01-20 20:45:00);之所以选 MySQL 而不选 SQLite是为了更贴近真实业务。SQLite 驱动简单但很多生产环境用 MySQL/PostgreSQL连接参数、驱动差异、事务隔离级别这些坑你迟早要趟不如一开始就按真实场景来。3.2 数据库连接与查询脚本的完整实现核心脚本query_db.py是整个 Skill 的关键它负责接收参数、连接数据库、执行查询、格式化返回结果。我用的驱动是PyMySQL因为它在纯 Python 环境下安装最省事不需要编译。#!/usr/bin/env python3 import argparse import json import os import sys import pymysql from dotenv import load_dotenv load_dotenv() # 表名白名单防止模型传入任意表名 ALLOWED_TABLES {orders, customers, products} # 聚合函数白名单防止模型注入非法 SQL 片段 ALLOWED_AGGREGATES {COUNT, SUM, AVG, MAX, MIN} def parse_args(): parser argparse.ArgumentParser(descriptionCoPaw Skill: query relational database) parser.add_argument(--table, requiredTrue, help目标表名) parser.add_argument(--columns, default*, help要查询的列逗号分隔) parser.add_argument(--where, default, helpWHERE 条件例如 amount 100) parser.add_argument(--order_by, default, help排序字段例如 created_at DESC) parser.add_argument(--limit, typeint, default100, help限制返回行数默认100) parser.add_argument(--aggregate, default, help聚合函数例如 SUM(amount)) args parser.parse_args() if args.table not in ALLOWED_TABLES: raise ValueError(f表 {args.table} 不在白名单中允许的表: {ALLOWED_TABLES}) if args.limit 500: raise ValueError(limit 不能超过 500防止返回数据量过大) return args def build_sql(args): 根据参数构造 SQL这里只做简单拼接真正值绑定交给 execute 的参数化机制。 sql SELECT if args.aggregate: # 聚合函数也要校验不能直接拼接模型传来的值 func args.aggregate.split(()[0].strip().upper() if func not in ALLOWED_AGGREGATES: raise ValueError(f不支持的聚合函数: {func}) sql f{args.aggregate} AS agg_result else: # 列名做基础校验只允许字母数字下划线 columns args.columns.split(,) cleaned [c.strip() for c in columns if c.strip().replace(_, ).isalnum()] if not cleaned: raise ValueError(columns 参数不合法) sql , .join(cleaned) sql f FROM {args.table} if args.where: # 这里允许模型传入 where 表达式但是只塞进参数化占位符值部分安全 # 注意我们要拦截明显是 SQL 注入尝试的输入 unsafe_keywords [;, --, /*, */, DROP, DELETE, UPDATE, INSERT, ALTER] upper_where args.where.upper() for kw in unsafe_keywords: if kw in upper_where: raise ValueError(fWHERE 条件包含非法关键字: {kw}) sql f WHERE {args.where} if args.order_by: # 排序字段只允许 字段名或字段名 ASC/DESC parts args.order_by.split() if len(parts) not in (1, 2): raise ValueError(order_by 参数不合法) sql f ORDER BY {parts[0]} {parts[1] if len(parts) 2 else ASC} # LIMIT 参数已做 int 校验可以安全拼接 sql f LIMIT {args.limit} return sql def main(): try: args parse_args() sql build_sql(args) conn pymysql.connect( hostos.getenv(DB_HOST, 127.0.0.1), portint(os.getenv(DB_PORT, 3306)), useros.getenv(DB_USER, root), passwordos.getenv(DB_PASSWORD, ), databaseos.getenv(DB_NAME, demo), charsetutf8mb4, cursorclasspymysql.cursors.DictCursor, connect_timeout5, read_timeout10, ) try: with conn.cursor() as cursor: cursor.execute(sql) rows cursor.fetchall() if not rows: print(json.dumps({success: True, message: 查询无结果, rows: []}, ensure_asciiFalse)) return # 最多返回 20 行原文超出部分合并统计 display_rows rows[:20] result { success: True, row_count: len(rows), displayed_rows: len(display_rows), rows: display_rows, truncation_message: f共 {len(rows)} 行仅显示前 20 行 if len(rows) 20 else None, } print(json.dumps(result, ensure_asciiFalse, defaultstr)) finally: conn.close() except Exception as e: print(json.dumps({success: False, error: str(e)}, ensure_asciiFalse)) sys.exit(1) if __name__ __main__: main()这段代码有几个设计点值得解释一是 WHERE 条件的处理。我并没有做完整的 SQL 解析而是用黑名单拦截明显危险的片段。这个方案不完美因为绕过黑名单的方式很多但对于一个内部工具型的 Skill它已经能挡住模型绝大部分的“无意作恶”。如果面对不可信输入你就得换成 SQL 白名单解析器把 WHERE 限定为“列名 操作符 值”这种固定模板。我在 4.2 节会详细说这个坑。二是返回结果做了“20 行展示 总数统计”的双层结构。这是为了让 Agent 既能回答具体数据又不会因为几千行结果直接撑爆上下文。模型拿到row_count和displayed_rows后如果发现结果太多还可以进一步追问聚合信息这比一股脑输出所有数据要优雅得多。三是连接超时设置。connect_timeout5和read_timeout10是我在踩过几次坑之后加上的。没有超时的时候数据库一卡Skill 脚本就挂在那里Agent 也会跟着等半天体验极差。加上超时后至少能快速失败并返回错误信息。3.3 将结果格式化为 Agent 友好的输出脚本最终通过 stdout 输出 JSON 字符串。我试过直接输出 Markdown 表格但效果不如 JSON。原因是 Agent 对 JSON 的结构化解析更稳定后续要转成自然语言回答时JSON 可以原样塞给模型而 Markdown 表格当你需要提取某个字段时会很别扭。JSON 是 Agent 和脚本之间最通用的“接口协议”人类看不友好无所谓模型看着舒服就行。我在assets/sample_output.md里存了一份示例输出方便调试时对照检查{ success: true, row_count: 3, displayed_rows: 3, rows: [ {id: 1, order_no: A0001, customer_name: 张三, amount: 150.00, status: 1, created_at: 2025-01-05 10:00:00}, {id: 2, order_no: A0002, customer_name: 李四, amount: 2300.50, status: 2, created_at: 2025-01-08 14:30:00} ], truncation_message: null }3.4 注册到 CoPaw 并测试调用流程编写完脚本后接下来是将这个 Skill 挂到 CoPaw 上。CoPaw 不同版本对 Skill 的注册方式略有差异但大方向一致把db-query目录放到 CoPaw 能扫描到的 skills 目录下然后在 CoPaw 的配置里声明这个技能。启动 CoPaw 后进入调试对话界面我一般会用下面几个问题做冒烟测试“查一下 orders 表里所有的订单按金额降序排取前 5 条”“统计一下总共有多少订单总金额是多少”“查一下张三的订单”如果配置正确CoPaw 会自动把用户的自然语言映射成 Skill 的参数生成类似下面这样的调用python scripts/query_db.py --table orders --order_by amount DESC --limit 5我这里特别关注一点模型是否正确地把“按金额降序排”翻译成了order_byamount DESC。刚开始开发时模型可能会生成whereamount DESC这种明显的语义错位后面我会讲怎么在 description 里做引导。3.5 参数计算过程示例处理一次真实查询请求假设用户提出“统计订单表中已支付订单的总金额”。注意“已支付”是个业务语义不是直接的数据库字段这里 status1 代表已支付。模型拿到这个请求后需要决策如何安排参数。如果 Skill 支持聚合函数模型会生成python scripts/query_db.py --table orders --where status 1 --aggregate SUM(amount)脚本内部会先校验SUM(amount)里的函数名SUM是否在白名单校验通过后构造 SQLSELECT SUM(amount) AS agg_result FROM orders WHERE status 1 LIMIT 100为什么要带上LIMIT 100因为聚合查询即使带 LIMIT 也不会改变结果但万一模型把聚合参数用错了LIMIT 仍然能兜底。这是我在实际项目中养成的习惯能多一层保护就多一层因为模型总会在你想不到的地方出错。如果模型没有意识到要用聚合函数而是直接查了所有订单再在自然语言里算总额那也没问题只是输出会包含一些无用数据。这种情况我建议在SKILL.md里加一行注释“当用户提到总计、平均、最大值等词汇时优先使用 aggregate 参数。”模型读到这个提示后会表现好很多。4. 踩坑实录与常见问题排查4.1 连接数据库失败但命令行能连上这是我被问过最多的问题。Skill 脚本在 CoPaw 调用时连不上数据库但在终端手动执行同样的命令却完全正常。排查后发现CoPaw 在调用 Skill 时子进程的环境变量和终端环境并不完全一致。我在终端里设置了DB_PASSWORD但 CoPaw 启动时没有继承这个环境变量脚本里os.getenv(DB_PASSWORD)取到的就是空字符串连接自然失败。解决方案是把数据库连接信息写进一个.env文件放在 Skill 目录下脚本启动时显式加载DB_HOST127.0.0.1 DB_PORT3306 DB_USERroot DB_PASSWORDroot123 DB_NAMEdemo然后在 CoPaw 的配置里允许 Skill 读取该文件。用.env的好处是它不依赖外部环境变量无论 CoPaw 以什么方式拉起子进程脚本都能找到配置。4.2 模型把参数传成了字符串拼接导致 SQL 注入或语法错模型不一定懂 SQL 注入但它可能把用户输入里的单引号原样带入 WHERE 子句比如用户说“查名字叫 张三 的客户”模型生成了wherecustomer_name 张三这本身没问题但如果用户说的是张三; DROP TABLE orders;--模型可能不加处理地拼进 SQL。我的应对办法是双层防御第一层脚本层黑名单。前面代码里已经写了拦截;、--、DROP、DELETE等关键词一旦命中直接报错不让 SQL 到达数据库。第二层如果你需要更安全就把 WHERE 参数改成结构化对象。比如参数改为--where_column customer_name --where_op --where_value 张三脚本内部再用参数化查询去执行cursor.execute( fSELECT {columns} FROM {table} WHERE {col} {op} %s LIMIT %s, (value, limit), )这才是正儿八经的防注入方案。前面那个黑名单方案只能防住模型无意造成的语法错误没法抵抗刻意攻击。如果你这个 Skill 要暴露给外部用户使用我强烈建议改成结构化参数别再让模型自由发挥 WHERE 表达式了。4.3 查询结果太大把上下文撑爆模型上下文不是无限大的如果你允许返回 1000 行每行 10 个字段一次性塞回对话里几千个 token 就没了。用户后续对话可用的空间会变得非常小Agent 也会越答越蠢。我的做法已经体现在脚本里默认最多返回 20 行原文同时返回总行数。如果用户真的需要看所有数据我建议另外做一个“导出 CSV”的 Skill把大数据量结果写到文件里再给用户下载链接而不是硬塞给模型。实际测试中这个限制非常有效。有一次用户问“查一下所有订单”当时测试表里只有几百行但脚本只返回 20 行总计信息模型自然回复“共有 345 条订单以下展示前 20 条”整个过程又流畅又省 token。4.4 安全边界只读账户、最小权限、禁止危险操作这是所有数据库类 Skill 里最重要的一条请务必放在心里。即使你的脚本已经做了参数校验也强烈建议在数据库层面再设一道防线。我的实践是为 Skill 单独创建一个只读数据库账号只授予SELECT权限CREATE USER copaw_read% IDENTIFIED BY StrongPass123; GRANT SELECT ON demo.* TO copaw_read%;这样即使脚本出现漏洞最坏的情况也只是被读取数据无法执行写入或删除操作。另外还要避免使用root账号特别是在联调环境。我犯过这个错误测试时图方便用了 root结果有一次模型误传了UPDATE语句好在脚本里黑名单拦住了吓得我赶紧把所有 Skill 切到了只读账号。安全这件事不能只靠一层防线。4.5 常见问题速查表现象可能原因解决方法Agent 不调用 Skilldescription 描述与用户意图不匹配检查 SKILL.md让 description 涵盖更多查询说法脚本报“表不在白名单”数据表未加入 ALLOWED_TABLES在脚本白名单中添加表名中文乱码连接时未指定 charsetutf8mb4连接参数加 charsetutf8mb4查询超时数据量大或索引缺失设置 read_timeout并检查 SQL 是否走索引Agent 答非所问参数映射错误打印 Skill 实际接收到的参数对照修正 description返回值太多把上下文撑爆没有限制行数确保 limit 默认值存在display 行数不要超过 50数据库连接被拒环境变量或 .env 缺失检查 .env 是否被加载密码是否匹配排查问题的思路我建议从“输入-执行-输出”三段式入手先确认 Agent 传给 Skill 的参数对不对再看脚本在数据库上执行的 SQL 是否合理最后检查返回 JSON 是否完整。大多数问题都能在这个链路里定位。5. 后续扩展让 Skill 更聪明一点基础的“从关系数据库查询数据”功能跑通后你会发现这个 Skill 的可玩性比想象中高。我目前在这个框架上加了几个方向每个都能明显提升 Agent 的实际战斗力。5.1 支持多个数据库和动态 Schema一上来就写死连一个库当然简单但真实场景里一个 Agent 可能要同时面对业务库、日志库、数据仓库等多个数据源。我的做法是在SKILL.md里增加一个--database参数CoPaw 会根据用户问题的上下文自动填入对应的库名。例如“查一下昨天日志库里的错误数量”和“查一下订单库中今天新增的订单”这两个请求会分别触发对日志库和订单库的查询。脚本里用database参数去映射不同的连接配置而不是在环境变量里只写一份。需要注意的是“数据库名”也是一个高危参数同样要放在白名单里不能让模型随便传一个名字。5.2 增加查询意图校验有时候用户的问题边界模糊比如“帮我分析一下销售数据”Agent 可能拿不定主意该查哪张表、该不该聚合。我建议在脚本里加一个“预检”步骤让模型先通过SHOW TABLES或查询information_schema拿到表的字段信息再决定最终 SQL。当然这个预检结果不会展示给用户只是作为内部推理依据。社区里管这个模式叫“先看表结构再写 SQL”。它可以大幅减少因为模型猜测字段名而导致的查询失败。我测试过的效果是加了预检之后查询成功率从 70% 左右提升到了 95% 以上代价是多了一次数据库元数据查询这个成本对绝大多数场景来说可以忽略。5.3 从只读查询扩展为“可解释查询”与“报表生成”既然已经有了一个受控的查询入口下一步很自然的想法就是把它升级成“数据洞察工具”。比如用户问“上月订单金额为什么降了”Agent 可以先查月度趋势、再按客户维度拆解、再对比退货率把多步查询结果汇总成一份可读的分析报告。这一步已经超出单纯“查数据”的范畴了但基础还是同一个 Skill。换句话说数据库查询 Skill 是地基往上可以堆分析、报表、告警、数据导出等能力。这也是我把第一个 Skill 选在数据库查询上的原因——它几乎能撬动一切数据相关的 Agent 应用。我在实际使用这个 Skill 时还有一个体会CoPaw 的 Skill 机制真正把“写死流程”和“智能决策”之间的缝隙填上了。以前给 Agent 接数据库要么写一串固定的 SQL 让它执行要么直接放弃让 Agent 碰数据库。现在把查询逻辑、安全校验、结果格式化全部封装进 SkillAgent 只需要负责“听懂人话”和“传对参数”两边各司其职整个系统稳了很多。后续如果你也想做数据相关的 Agent不用急着写复杂的调度逻辑先从一个能查库的 Skill 开始跑通之后你再回头看我这句话应该会有同感。