模板与检查脚本

技能目录骨架，复制后按注释修改。技能包里的原文件在 templates/ 与 scripts/ 下。

SKILL.md.tmpl

templates/SKILL.md.tmpl

---
name: <skill_name>                 # 必须 ^[a-z0-9][a-z0-9_-]*$，且与目录名一致（安装预检会拦）
description: <一句话说清「什么时候该调它」：用户说「…」「…」时；做什么；不做什么。模型靠这句选工具，写得偏「推」一点。值里不要出现「英文冒号+空格」，否则 YAML 解析失败，平台退回逐行解析、嵌套块全丢；必须出现时整句加双引号>
version: 0.1.0                     # 每次发布 bump；市场按它判断「有更新」
source: builtin                    # builtin / shared / market
category: general                  # 只认：office finance ecommerce marketing promotion trading data crm channel media integration compliance general
trigger-type: explicit             # explicit 才进模型工具清单
cost-tier: low                     # zero / low / medium / high
tags: <a, b, c>
async: false                       # true = 长耗时技能，非 web 渠道转后台
force_raw_prompt: false            # true = 用用户原文覆盖第一个必填字段（生成类）
# skill_type: handler              # 可省；有 handler.py 即判执行型。knowledge / skill_md 才需要显式写
# skill_md: SKILL.md               # 文档型主文档不叫 SKILL.md 时指定
# ui: ui.html                      # 本地面板入口（.html）；写 https://… 即外部面板。见 references/ui-panel.md
# exec:                            # 非进程内 Python 时声明
#   config: request.toml           # HTTP 型
#   primary: {type: node}          # SHELL / NODE / RUBY / GO
#
# ── 配置项声明（控制台「技能参数」表单 / 安装表单读这里）──
# 无配置写 env: []（显式零配置，平台不再按正文嗅探疑似密钥）。
# 有配置写结构化数组；required 缺省为 true，secret 缺省为 false：
env:
  - name: <XXX_API_KEY>
    label: <XXX 的 API Key>
    description: <去哪申请、格式>
    required: true
    secret: true
# ── 运行前校验 / 子代理鉴权提示读这里（执行器侧）──
metadata:
  openclaw:
    primaryEnv: <XXX_API_KEY>      # 主凭证变量名，无则删整段 metadata
    requires:
      env:  [<XXX_API_KEY>]
      bins: []
---

# <skill_name>

## 作用

<两三行：它做什么、返回什么>

## 触发方式

<用户的口语说法，至少三个。动作型技能写「action 必须传中文原词」并给三个调用示例>

| 用户这样说 | 它回答什么 / 做什么 |
|---|---|
| 「…」 | … |

## 输入

<关键参数及取值范围。完整 schema 见 schema.json>

## 输出

<成功返回的关键字段；失败时的 error_type 分类>

## 注意

<该技能特有的限制；哪些 reference 要先读（文档型必写）>

handler_async.py

templates/handler_async.py

"""
进程内 async Python handler 骨架（动作型）。
- 入口签名照抄 `async def execute(input_data, runtime_context)`：create_skill / edit_skill
  的安装预检按函数名 execute、参数名 input_data / runtime_context、async 三项硬拦截
- 第二参数不设默认值
- 不要在本文件里写 argparse + __main__（会被判成 CLI 型）
- 身份信息只读 runtime_context；input_data 里的同名字段忽略
- 直连库只走 _db()（每次从 db_factory 现开现还），不读 runtime_context["db"]
"""
import json
import logging
import os

logger = logging.getLogger(__name__)

SKILL_NAME = "<skill_name>"
VERSION = "0.1.0"

# ---------- action 归一 ----------
ACTIONS = {"初始化", "拉数据", "查询", "更新"}
WRITE_ACTIONS = {"初始化", "更新"}
ALIASES = {
    "initialize": "初始化", "init": "初始化", "setup": "初始化",
    "fetch": "拉数据", "pull": "拉数据", "sync": "拉数据",
    "query": "查询", "search": "查询", "list": "查询",
    "update": "更新", "upsert": "更新",
}
assert set(ALIASES.values()) <= ACTIONS, "别名表的值必须是规范 action"

# 注意：常量名避开 key/token/secret/password/credential/auth 子串
PARAM_NAME_MAP = {
    "record": "data", "row": "data", "values": "data",
    "filter": "filters", "order_by": "sort",
    "confirm": "确认", "confirmed": "确认", "approve": "确认", "确定": "确认",
}


def _unwrap(input_data: dict) -> dict:
    """入口先拆包：payload / input / args 包了一层也要读到 action。"""
    d = dict(input_data or {})
    for k in ("payload", "input", "args"):
        inner = d.get(k)
        if isinstance(inner, str):
            try:
                inner = json.loads(inner)
            except Exception:
                inner = None
        if isinstance(inner, dict):
            d = {**inner, **{kk: vv for kk, vv in d.items() if kk != k}}
    return d


def _norm_action(raw):
    s = str(raw or "").strip()
    low = s.lower().replace("-", "_").replace(" ", "")
    if s in ACTIONS:
        return s, False, None
    if low in ALIASES:
        return ALIASES[low], False, f"{raw} → {ALIASES[low]}"
    for a in WRITE_ACTIONS:
        if s.startswith("确认" + a) or s.startswith("确认 " + a):
            return a, True, f"{raw} → {a}（含确认前缀）"
    return None, False, None


def _norm_param_names(d: dict):
    out, notes = {}, []
    for k, v in d.items():
        nk = PARAM_NAME_MAP.get(k, k)
        if nk != k:
            notes.append(f"{k} → {nk}")
        out[nk] = v
    return out, notes


def _maybe_json(v):
    """数组/对象参数可能以 JSON 字符串形式到达。"""
    if isinstance(v, str) and v.strip()[:1] in "[{":
        try:
            return json.loads(v)
        except Exception:
            return v
    return v


# ---------- 工作区（executor 已注入 workspace_dir；role_id 为空等边角情况再自推） ----------
from pathlib import Path

def _workspace(ctx: dict) -> Path:
    ws = ctx.get("workspace_dir")
    if ws:
        return Path(ws)
    root = os.getenv("AGENT_WORKSPACE_DIR", "data/workspaces")
    p = Path(root) / (ctx.get("role_id") or "") / (ctx.get("session_id") or "shared")
    p.mkdir(parents=True, exist_ok=True)
    return p


# ---------- 配置取用（四层顺序） ----------
def _cfg(ctx: dict, name: str) -> str:
    for src in (ctx.get("skill_configs", {}).get(SKILL_NAME) or {},
                ctx.get("skill_configs") or {},
                ctx.get("skill_args") or {},
                ctx):
        v = src.get(name)
        if v is not None and str(v).strip():
            return str(v)
    return os.getenv(name, "")


# ---------- datatable 技能名探测 ----------
DT_CANDIDATES = ["datatable", "nocodb"]
_dt_name = None


async def _resolve_dt(ctx: dict):
    """规范 §7：用 list_tables 逐个探测，回「未注册」换下一个，命中即缓存。"""
    global _dt_name
    if _dt_name:
        return _dt_name, None
    last = None
    for n in DT_CANDIDATES:
        r = await ctx["call_skill"](n, {"action": "list_tables"})
        _e = str(r.get("error", ""))
        if r.get("ok") is False and ("未注册" in _e or "未知工具" in _e):
            last = r
            continue
        _dt_name = n
        return n, None
    return None, {"ok": False, "error": f"数据表技能未注册（已探测 {DT_CANDIDATES}）",
                  "error_type": "config", "non_retriable": True, "upstream": last}


async def dt(ctx: dict, action: str, **kw) -> dict:
    """调 datatable。子技能是独立 session，调用前自己的写入必须已提交——
    用下方 _db(commit=True) 现开现还时每次写即已提交，这里无需再 commit。"""
    name, err = await _resolve_dt(ctx)
    if err:
        return err
    return await ctx["call_skill"](name, {"action": action, **kw})


# ---------- 直连库（仅当 datatable 表达不了时用；唯一 seam，可 monkeypatch 自测） ----------
async def _db(ctx: dict, sql: str, params=None, *, fetch: bool = False, commit: bool = False):
    """一次调用 = 一条短会话 = 一个事务。异常交给 async with 回滚并关闭。"""
    factory = (ctx or {}).get("db_factory")
    if factory is None:
        raise RuntimeError("缺少 db_factory（子进程 / 非 async 入口拿不到库）")
    from sqlalchemy import text
    async with factory() as session:
        result = await session.execute(text(sql), params or {})
        rows = [dict(m) for m in result.mappings().all()] if fetch else None
        if commit:
            await session.commit()
        return rows


def _failed(r: dict) -> bool:
    return r.get("ok") is False or r.get("success") is False


# ---------- 入口 ----------
async def execute(input_data: dict, runtime_context: dict) -> dict:
    ctx = runtime_context or {}
    d = _unwrap(input_data)
    d, name_notes = _norm_param_names(d)
    action, confirmed_by_prefix, act_note = _norm_action(d.get("action"))
    notes = ([act_note] if act_note else []) + name_notes

    if action is None:
        return {
            "success": False, "need_param": "action", "error_type": "recoverable",
            "error": f"action 缺失或无法识别：{d.get('action')!r}。必须传中文原词",
            "user_message": "请指定操作：" + " / ".join(sorted(ACTIONS)),
            "received": {k: v for k, v in d.items() if k != "action"},
        }

    if not (ctx.get("org_id") or "").strip():
        return {"success": False, "error_type": "config", "non_retriable": True,
                "error": "org_id 为空，角色未绑定组织",
                "user_message": "当前角色未绑定组织，无法操作数据表，请联系管理员配置。"}

    confirmed = bool(d.get("确认")) or confirmed_by_prefix

    try:
        if action == "查询":
            result = await _do_query(d, ctx)
        elif action == "拉数据":
            result = await _do_fetch(d, ctx)
        elif action in WRITE_ACTIONS:
            result = await _do_write(action, d, ctx, confirmed)
        else:
            result = {"success": False, "error": f"未实现：{action}",
                      "error_type": "config", "non_retriable": True}
    except Exception as e:  # 兜底：必须给 error，不得吞（_db 的回滚已由 async with 完成）
        logger.exception("%s failed", action)
        result = {"success": False, "error": f"{type(e).__name__}: {e}", "error_type": "transient"}

    if notes:
        result["action_note"] = "; ".join(notes)
    result.setdefault("version", VERSION)
    return result


# ---------- 各动作（按需实现） ----------
async def _do_query(d: dict, ctx: dict) -> dict:
    limit = int(d.get("limit") or 100)
    filters = _maybe_json(d.get("filters")) or []
    r = await dt(ctx, "query", table_name="<表名>", filters=filters, limit=limit)
    if _failed(r):
        up = r.get("output") if isinstance(r.get("output"), dict) else r   # call_skill 失败把原信封放在 output
        return {"success": False, "error": f"datatable 查询失败：{up.get('user_message') or r.get('error')}",
                "need_param": up.get("need_param"),
                "error_type": up.get("error_type", "recoverable"), "upstream": up}
    # datatable 成功信封 = {"success": True, "data": {list, pageInfo}}（已按源码核实）
    rows = (r.get("data") or {}).get("list", [])
    # rows 可被结果卸载；count 放顶层
    return {"success": True, "count": len(rows), "rows": rows}


async def _do_fetch(d: dict, ctx: dict) -> dict:
    api_value = _cfg(ctx, "XXX_API_KEY")
    if not api_value:
        return {"success": False, "error_type": "config", "non_retriable": True,
                "error": "未配置 XXX_API_KEY",
                "user_message": "请在对话中直接提供 XXX 的 API Key，平台会加密保存。"}
    # ... 取数；落库：1 条 insert / ≤100 insert_batch / >100 写工作区子目录 JSON 后 insert_from_file
    return {"success": True, "inserted": 0, "failed": 0, "verified_total": 0}


async def _do_write(action: str, d: dict, ctx: dict, confirmed: bool) -> dict:
    plan = {"action": action,
            "params": {k: v for k, v in d.items() if k not in ("action", "确认")}}
    if not confirmed:
        return {
            "success": True, "dry_run": True, "plan": plan,
            "display_md": (f"**预演 · {action}**\n\n```json\n"
                           f"{json.dumps(plan, ensure_ascii=False, indent=2)}\n```\n\n"
                           "缺 `确认=true`，所以只预演未执行。"),
            "for_user": f"可继续的操作：回复「确认{action}」执行；或调整参数后再预演。",
        }
    # ... 真正执行；写后读回核验
    return {"success": True, "updated": 0, "verified_total": 0}

handler_cli.py

templates/handler_cli.py

#!/usr/bin/env python3
"""
子进程 CLI 型 handler 骨架。
- 含 argparse + __main__ 即被平台判为 CLI 型
- 上下文从 SKILL_CONTEXT 读（无 db / call_skill / abort_signal）
- stdout 只打最终 JSON；日志走 stderr；业务失败退出码仍为 0
- 子命令到达时在 argv[0]
- 默认超时 60 秒
- 工作区：SKILL_CONTEXT 里的 workspace_dir 或环境变量 SKILL_WORKSPACE（同值）；技能目录读 SKILL_DIR
"""
import argparse
import json
import os
import sys


def log(*a):
    print(*a, file=sys.stderr)


def emit(obj: dict, code: int = 0):
    print(json.dumps(obj, ensure_ascii=False))
    sys.exit(code)


def main():
    input_data = json.loads(os.environ.get("SKILL_INPUT") or "{}")
    ctx = json.loads(os.environ.get("SKILL_CONTEXT") or "{}")

    p = argparse.ArgumentParser()
    p.add_argument("subcommand", nargs="?", default=input_data.get("_subcommand", ""))
    p.add_argument("--symbol", default=input_data.get("symbol", ""))
    args = p.parse_args()

    if not args.symbol:
        emit({"success": False, "need_param": "symbol", "error_type": "recoverable",
              "error": "symbol 为空", "user_message": "请给出标的代码，如 BTC"})

    api_value = os.environ.get("XXX_API_KEY", "")      # 凭证走环境变量，不打印
    if not api_value:
        emit({"success": False, "error_type": "config", "non_retriable": True,
              "error": "未配置 XXX_API_KEY"})

    log(f"actor={ctx.get('actor_name')!r} channel={ctx.get('channel')!r}")

    try:
        # ... 业务逻辑；产物写到会话工作区（ctx["workspace_dir"] 或 SKILL_WORKSPACE），
        #     返回 filename / local_path；中间文件放工作区子目录
        ws = ctx.get("workspace_dir") or os.environ.get("SKILL_WORKSPACE") or os.getcwd()
        emit({"success": True, "data": {"symbol": args.symbol.upper()}})
    except Exception as e:
        log("FATAL", repr(e))
        emit({"success": False, "error": f"{type(e).__name__}: {e}",
              "error_type": "transient"}, code=1)


if __name__ == "__main__":
    main()

schema.json

templates/schema.json

{
  "type": "object",
  "properties": {
    "action": {
      "type": "string",
      "description": "操作名，必须传中文原词：初始化 / 拉数据 / 查询 / 更新。以「确认」开头表示用户已确认写操作，如「确认更新」。",
      "enum": ["初始化", "拉数据", "查询", "更新", "确认更新", "确认初始化"]
    },
    "symbol": {"type": "string", "description": "标的代码，大写，如 BTC / AAPL"},
    "limit": {"type": "integer", "default": 100, "description": "返回条数，1~1000"},
    "确认": {"type": "boolean", "description": "仅当用户明确说了「确认」时才传 true。预演不需要。"},
    "file": {"type": "string", "description": "输入文件。可直接传素材清单里的引用 ID（如 file_1）"}
  },
  "required": ["action"]
}

panel/ui.html

templates/panel/ui.html

<!DOCTYPE html>
<!-- 本地面板入口模板。文件名可改，但要在 SKILL.md 的 ui: 里声明。
     只用相对路径引资源，平台会自动改写成 asset 端点。 -->
<html lang="zh"><head>
  <meta charset="utf-8">
  <meta name="viewport" content="width=device-width, initial-scale=1">
  <link rel="stylesheet" href="style.css">
  <title>面板</title>
</head><body>
  <div class="card">
    <h1>面板标题</h1>
    <input id="name" placeholder="输入点什么" />
    <button id="go">执行</button>
    <div id="out"></div>
  </div>
  <script src="app.js"></script>
</body></html>

panel/app.js

templates/panel/app.js

/* 本地面板逻辑模板。
 * 握手四步（ready → init → invoke → invoke_result）照抄，频道名勿改。
 * 只改最底部「按技能改」那一段：点按钮调哪个动作、怎么显示结果。 */
(function () {
  var CH = 'skill_panel';            // 固定频道名，勿改
  var pending = {}, seq = 0;

  // 1) 必发 ready，否则平台不下发 init、不接受 invoke
  parent.postMessage({ channel: CH, type: 'ready' }, '*');

  window.addEventListener('message', function (e) {
    var d = e.data || {};
    if (d.channel !== CH) return;
    // 2) 收 init（可套主题变量）
    if (d.type === 'init' && d.theme) {
      for (var k in d.theme) document.documentElement.style.setProperty(k, d.theme[k]);
    }
    // 4) 收结果，回填到对应 invoke 的 Promise
    if (d.type === 'invoke_result') {
      var fn = pending[d.reqId];
      if (fn) { delete pending[d.reqId]; fn(d); }
    }
  });

  // 3) 调后端动作：invoke(action, args) → Promise<{ok, data, error}>
  function invoke(action, args) {
    return new Promise(function (resolve) {
      var reqId = 'r' + (++seq);
      pending[reqId] = resolve;
      parent.postMessage({ channel: CH, type: 'invoke', reqId: reqId,
                           action: action, args: args || {} }, '*');
    });
  }

  // ↓↓↓ 按技能改这一段 ↓↓↓
  document.getElementById('go').addEventListener('click', async function () {
    var out = document.getElementById('out');
    out.textContent = '调用中…';
    var res = await invoke('greet', { name: document.getElementById('name').value.trim() });
    out.textContent = res.ok
      ? (res.data && res.data.greeting || '(无返回)')
      : '失败：' + (res.error || '未知错误');
  });
  // ↑↑↑ 按技能改这一段 ↑↑↑
})();

panel/style.css

templates/panel/style.css

/* 本地面板样式模板。
 * 宽高自适应 iframe，别写死像素。
 * var(--xxx) 是平台 init 时可能下发的主题变量，给了兜底值。 */
* { box-sizing: border-box; }
html, body { margin: 0; height: 100%; }
body {
  font-family: var(--font, system-ui, -apple-system, "PingFang SC", sans-serif);
  color: var(--fg, #1a1a1a);
  background: var(--bg, transparent);
  padding: 16px;
}
.card {
  max-width: 480px;
  margin: 0 auto;
  padding: 20px;
  border: 1px solid var(--border, #e2e2e2);
  border-radius: 12px;
  background: var(--card-bg, #fff);
}
.card h1 { margin: 0 0 12px; font-size: 18px; }
input {
  width: 100%;
  padding: 8px 10px;
  margin-bottom: 10px;
  border: 1px solid var(--border, #d0d0d0);
  border-radius: 8px;
  font-size: 14px;
}
button {
  padding: 8px 16px;
  border: none;
  border-radius: 8px;
  background: var(--accent, #3b6ef5);
  color: #fff;
  font-size: 14px;
  cursor: pointer;
}
button:active { opacity: .85; }
#out { margin-top: 12px; min-height: 20px; white-space: pre-wrap; }

检查脚本

交付前运行：

python3 scripts/check_skill.py <技能目录>

脚本按本规范逐项检查 SKILL.md、handler、schema.json，输出 ERROR / WARN 清单。