add the function of script

AMOS144 · AMOS144 · commit cd382f24749c · 2026-03-10T17:20:58.000+08:00
diff --git a/mcp_server.py b/mcp_server.py
@@ -382,6 +382,46 @@ async def list_tools() -> list[Tool]:
                 "required": ["session_id"]
             }
         ),
+        Tool(
+            name="script_binding_set",
+            description="Bind an external job_id (binding_key) to a script task_id, with optional default vars and description.",
+            inputSchema={
+                "type": "object",
+                "properties": {
+                    "binding_key": {"type": "string", "description": "External job identifier (e.g. OpenClaw job_id)"},
+                    "script_task_id": {"type": "string", "description": "ZeroToken script task_id"},
+                    "description": {"type": "string", "description": "Optional human-readable description"},
+                    "default_vars": {"type": "object", "description": "Optional default vars for this binding"}
+                },
+                "required": ["binding_key", "script_task_id"]
+            }
+        ),
+        Tool(
+            name="script_binding_get",
+            description="Get a script binding by binding_key (job_id).",
+            inputSchema={
+                "type": "object",
+                "properties": {"binding_key": {"type": "string"}},
+                "required": ["binding_key"]
+            }
+        ),
+        Tool(
+            name="script_binding_list",
+            description="List script bindings.",
+            inputSchema={
+                "type": "object",
+                "properties": {"limit": {"type": "integer", "default": 100}}
+            }
+        ),
+        Tool(
+            name="script_binding_delete",
+            description="Delete a script binding by binding_key.",
+            inputSchema={
+                "type": "object",
+                "properties": {"binding_key": {"type": "string"}},
+                "required": ["binding_key"]
+            }
+        ),
         Tool(
             name="browser_init",
             description="Initialize the browser (call once before using other browser tools). Use stealth=True to reduce automation detection (launch args + fingerprint masking).",
@@ -803,6 +843,40 @@ async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
             steps = get_storage().session_get(session_id)
             return [TextContent(type="text", text=json.dumps({"success": True, "steps": steps}, indent=2, ensure_ascii=False))]
 
+        elif name == "script_binding_set":
+            binding_key = arguments.get("binding_key")
+            script_task_id = arguments.get("script_task_id")
+            if not binding_key or not script_task_id:
+                return [TextContent(type="text", text=_error_response("binding_key and script_task_id are required", code="INVALID_PARAMS", retryable=False))]
+            get_storage().script_binding_set(
+                binding_key,
+                script_task_id=script_task_id,
+                description=arguments.get("description", "") or "",
+                default_vars=arguments.get("default_vars") or {},
+            )
+            return [TextContent(type="text", text=json.dumps({"success": True, "binding_key": binding_key, "script_task_id": script_task_id}, indent=2, ensure_ascii=False))]
+
+        elif name == "script_binding_get":
+            binding_key = arguments.get("binding_key")
+            if not binding_key:
+                return [TextContent(type="text", text=_error_response("binding_key is required", code="INVALID_PARAMS", retryable=False))]
+            binding = get_storage().script_binding_get(binding_key)
+            if binding is None:
+                return [TextContent(type="text", text=_error_response(f"No binding for key: {binding_key}", code="SCRIPT_BINDING_NOT_FOUND", retryable=False))]
+            return [TextContent(type="text", text=json.dumps({"success": True, "binding": binding}, indent=2, ensure_ascii=False))]
+
+        elif name == "script_binding_list":
+            limit = arguments.get("limit", 100)
+            items = get_storage().script_binding_list(limit=limit)
+            return [TextContent(type="text", text=json.dumps({"bindings": items}, indent=2, ensure_ascii=False))]
+
+        elif name == "script_binding_delete":
+            binding_key = arguments.get("binding_key")
+            if not binding_key:
+                return [TextContent(type="text", text=_error_response("binding_key is required", code="INVALID_PARAMS", retryable=False))]
+            ok = get_storage().script_binding_delete(binding_key)
+            return [TextContent(type="text", text=json.dumps({"success": True, "deleted": ok}, indent=2, ensure_ascii=False))]
+
         else:
             return [TextContent(
                 type="text",
diff --git a/skills/zerotoken-openclaw/SKILL.md b/skills/zerotoken-openclaw/SKILL.md
@@ -85,6 +85,26 @@ Agent 应：
 
 **不主动生成**：未提复用、未提定时/重复时，只做轨迹录制与保存。若用户后续要脚本再生成。
 
+## 定时任务如何找到对应脚本（基于 job_id 绑定）
+
+当 **OpenClaw 以定时任务触发本 Skill** 时，事件参数中会携带该任务的 `job_id`。ZeroToken 使用 `job_id` 作为绑定键（`binding_key`），并在 MCP 数据库的 `script_bindings` 表中维护「job_id ↔ 脚本」关系。
+
+Agent 必须遵守以下约定：
+
+1. **始终使用 `job_id` 作为 binding_key 查询绑定**：
+   - 调用 `script_binding_get(binding_key=job_id)`。
+2. 若返回 binding：
+   - 读取 `binding.script_task_id` 与 `binding.default_vars`；
+   - 合并本次任务特有的参数（如用户给定的视频 ID、时间范围）到 vars；
+   - 调用 `run_script(binding.script_task_id, vars=merged_vars)` 执行脚本。
+3. 若 `script_binding_get(job_id)` 返回「未找到」：
+   - 提示用户「当前 job_id 尚未绑定 ZeroToken 脚本」；
+   - 不要随意尝试其他脚本或自动新建脚本。
+4. 对于没有 `job_id` 或未标记为定时任务的场景：
+   - 视为「一次性任务」，只使用 `browser_*` + `trajectory_*` 完成当前需求，不主动查找/执行脚本。
+
+开发者应在 ZeroToken 侧或 OpenClaw 的集成层中，使用 `script_binding_set(binding_key=job_id, script_task_id=..., default_vars?, description?)` 预先将定时任务 job_id 与脚本 `task_id` 明确绑定。本 Skill 仅通过 `job_id` 查询绑定，**不对映射关系做额外推断**。
+
 ## 脚本格式与执行方式
 
 ### 格式（存于 MCP 数据库）
@@ -111,16 +131,23 @@ Agent 应：
 - 可选 `fuzzy_point`：记录该步「需要 AI/人介入」的语义信息（`reason`、`hint`），**本身不会让 ScriptEngine 自动暂停**；只有当为该步配置了匹配的 DFU / 执行点时，`run_script` 执行到该步才会返回 `status="paused"`。
 - 可选参数化：`params` 中可用 `{{varname}}`，执行前由 Agent 或配置替换（如环境变量、用户输入），或在 ExecutionPoint/DFU 暂停时由上层生成 `resolution.vars` 合并进运行时变量环境。
 
-### 执行脚本（由 ScriptEngine 自动化顺序执行）
+### 执行脚本（仅在定时 / 重复任务场景）
+
+**只有在以下两种情况下，才去查找并执行脚本：**
 
-当用户或 cron 消息为「执行 ZeroToken 脚本 &lt;task_id&gt;」或「跑一下 &lt;task_id&gt; 的脚本」时：
+- 上下文/cron 明确表明是「定时 / 周期性 / 重复执行」的任务（如每日评论、每小时抓取报表）。
+- 用户明确说「执行 ZeroToken 脚本 &lt;task_id&gt;」「跑一下 &lt;task_id&gt; 的脚本」等。
 
-1. 调用 `script_load(task_id)` 从 MCP 数据库读取脚本；若无则提示先根据轨迹生成并 `script_save`。
+在这些情况下：
+
+1. 调用 `script_load(task_id)` 从 MCP 数据库读取脚本；若无则提示先根据对应轨迹生成并 `script_save`（否则不要擅自造脚本）。
 2. 调用 `run_script(task_id, vars?)` 由 **MCP 内的 ScriptEngine 自动按 `steps` 顺序执行脚本**，无需 LLM，执行过程写入 session；返回形如 `{"success": ..., "status": "success|paused|failed", "session_id": ...}`。
 3. 若返回 `status="paused"`（例如命中 DFU / 执行点 / 失败重试上限）：
    - 上层 Agent 阅读 `pause_event`（包含 step_index、dfu_id、提示文案与需要生成的 vars），做一次决策或生成 vars；
    - 再调用 `run_script(session_id=..., resolution={...})` 恢复执行，由 ScriptEngine 继续顺序执行后续 steps。
 
+非定时/一次性任务：**优先只用 browser_* + trajectory_* 录制与完成当前任务，不主动查找/执行脚本。**
+
 脚本是「数据驱动的 MCP 调用序列」，**存于 MCP 数据库，由 ScriptEngine 自动化回放**，Token 消耗低且可通过 session 追踪每次执行。
 
 ### 模糊点 / DFU 执行约定
diff --git a/tests/test_script_bindings.py b/tests/test_script_bindings.py
@@ -0,0 +1,48 @@
+"""Tests for script bindings (job_id -> script_task_id) in SQLiteStorage."""
+
+import os
+import tempfile
+
+import pytest
+
+from zerotoken.storage_sqlite import SQLiteStorage
+
+
+@pytest.fixture
+def db_path():
+    fd, path = tempfile.mkstemp(suffix=".db")
+    os.close(fd)
+    yield path
+    try:
+        os.unlink(path)
+    except OSError:
+        pass
+
+
+@pytest.fixture
+def storage(db_path):
+    return SQLiteStorage(db_path)
+
+
+def test_script_binding_set_get_list_delete(storage: SQLiteStorage):
+    key = "job_daily_comment"
+    storage.script_binding_set(
+        key,
+        script_task_id="task_script_1",
+        description="每日评论脚本",
+        default_vars={"channel": "bilibili", "video_id": "abc"},
+    )
+
+    binding = storage.script_binding_get(key)
+    assert binding is not None
+    assert binding["binding_key"] == key
+    assert binding["script_task_id"] == "task_script_1"
+    assert binding["description"] == "每日评论脚本"
+    assert binding["default_vars"]["channel"] == "bilibili"
+
+    items = storage.script_binding_list()
+    assert any(it["binding_key"] == key for it in items)
+
+    assert storage.script_binding_delete(key) is True
+    assert storage.script_binding_get(key) is None
+
diff --git a/zerotoken/storage.py b/zerotoken/storage.py
@@ -177,3 +177,34 @@ def runtime_update(
     ) -> None:
         """Update runtime state fields for a session."""
         ...
+
+
+class ScriptBindingStore(ABC):
+    """Abstract store for binding external job_ids to script task_ids."""
+
+    @abstractmethod
+    def script_binding_set(
+        self,
+        binding_key: str,
+        *,
+        script_task_id: str,
+        description: str = "",
+        default_vars: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        """Create or update a binding from binding_key (e.g. OpenClaw job_id) to script_task_id."""
+        ...
+
+    @abstractmethod
+    def script_binding_get(self, binding_key: str) -> Optional[Dict[str, Any]]:
+        """Get binding by binding_key. Returns None if not found."""
+        ...
+
+    @abstractmethod
+    def script_binding_list(self, limit: int = 100) -> List[Dict[str, Any]]:
+        """List bindings (binding_key, script_task_id, description)."""
+        ...
+
+    @abstractmethod
+    def script_binding_delete(self, binding_key: str) -> bool:
+        """Delete binding. Returns True if deleted."""
+        ...
diff --git a/zerotoken/storage_sqlite.py b/zerotoken/storage_sqlite.py
@@ -8,7 +8,7 @@
 from pathlib import Path
 from typing import Any, Dict, List, Optional
 
-from .storage import ScriptStore, TrajectoryStore, SessionStore, DFUStore, SessionRuntimeStore
+from .storage import ScriptStore, TrajectoryStore, SessionStore, DFUStore, SessionRuntimeStore, ScriptBindingStore
 
 _RUNTIME_UNSET = object()
 
@@ -23,7 +23,7 @@ def _json_deserializer(s: Optional[str]) -> Any:
     return json.loads(s)
 
 
-class SQLiteStorage(ScriptStore, TrajectoryStore, SessionStore, DFUStore, SessionRuntimeStore):
+class SQLiteStorage(ScriptStore, TrajectoryStore, SessionStore, DFUStore, SessionRuntimeStore, ScriptBindingStore):
     """Single SQLite-backed storage for scripts, trajectories, and sessions."""
 
     def __init__(self, db_path: str = "zerotoken.db"):
@@ -110,6 +110,16 @@ def _ensure_tables(self) -> None:
                 updated_at TEXT NOT NULL
             )
         """)
+        cur.execute("""
+            CREATE TABLE IF NOT EXISTS script_bindings (
+                binding_key TEXT PRIMARY KEY,
+                script_task_id TEXT NOT NULL,
+                description TEXT,
+                default_vars_json TEXT,
+                created_at TEXT NOT NULL,
+                updated_at TEXT NOT NULL
+            )
+        """)
         self.conn.commit()
 
     # --- ScriptStore ---
@@ -147,6 +157,78 @@ def script_save(
         )
         self.conn.commit()
 
+    # --- ScriptBindingStore ---
+    def script_binding_set(
+        self,
+        binding_key: str,
+        *,
+        script_task_id: str,
+        description: str = "",
+        default_vars: Optional[Dict[str, Any]] = None,
+    ) -> None:
+        now = datetime.utcnow().isoformat() + "Z"
+        cur = self.conn.cursor()
+        cur.execute(
+            """
+            INSERT INTO script_bindings (binding_key, script_task_id, description, default_vars_json, created_at, updated_at)
+            VALUES (?, ?, ?, ?, ?, ?)
+            ON CONFLICT(binding_key) DO UPDATE SET
+                script_task_id = excluded.script_task_id,
+                description = excluded.description,
+                default_vars_json = excluded.default_vars_json,
+                updated_at = excluded.updated_at
+            """,
+            (
+                binding_key,
+                script_task_id,
+                description,
+                _json_serializer(default_vars or {}),
+                now,
+                now,
+            ),
+        )
+        self.conn.commit()
+
+    def script_binding_get(self, binding_key: str) -> Optional[Dict[str, Any]]:
+        cur = self.conn.cursor()
+        cur.execute(
+            "SELECT binding_key, script_task_id, description, default_vars_json, created_at, updated_at FROM script_bindings WHERE binding_key = ?",
+            (binding_key,),
+        )
+        row = cur.fetchone()
+        if row is None:
+            return None
+        return {
+            "binding_key": row["binding_key"],
+            "script_task_id": row["script_task_id"],
+            "description": row["description"] or "",
+            "default_vars": _json_deserializer(row["default_vars_json"]) or {},
+            "created_at": row["created_at"],
+            "updated_at": row["updated_at"],
+        }
+
+    def script_binding_list(self, limit: int = 100) -> List[Dict[str, Any]]:
+        cur = self.conn.cursor()
+        cur.execute(
+            "SELECT binding_key, script_task_id, description, updated_at FROM script_bindings ORDER BY updated_at DESC LIMIT ?",
+            (limit,),
+        )
+        return [
+            {
+                "binding_key": r["binding_key"],
+                "script_task_id": r["script_task_id"],
+                "description": r["description"] or "",
+                "updated_at": r["updated_at"],
+            }
+            for r in cur.fetchall()
+        ]
+
+    def script_binding_delete(self, binding_key: str) -> bool:
+        cur = self.conn.cursor()
+        cur.execute("DELETE FROM script_bindings WHERE binding_key = ?", (binding_key,))
+        self.conn.commit()
+        return cur.rowcount > 0
+
     def script_load(self, task_id: str) -> Optional[Dict[str, Any]]:
         cur = self.conn.cursor()
         cur.execute(
