openclaw-flow-kit

Original source / Raw SKILL.md

---
name: openclaw-flow-kit
description: "Fix common OpenClaw workflow bottlenecks: platform engage-gates/429 backoff helpers (starting with MoltX), standardized JSON result envelopes for chaining scripts, workspace path resolution helpers, and a simple skill release conductor (prepare/publish/draft announcements)."
---

# OpenClaw Flow Kit

Use this when you hit:
- platform **engage gates** / flaky **429** loops (esp. MoltX)
- inconsistent script outputs that make skill-chaining painful
- workspace-relative path bugs (writing to skills/state vs state)
- repetitive skill release steps (publish + generate announcement drafts)

## Quick commands

### 1) Standardized result envelope for any command
```bash
python scripts/run_envelope.py -- cmd /c "echo hello"
```
Outputs JSON:
- `ok`, `exit_code`, `stdout`, `stderr`, `startedAt`, `endedAt`, `durationMs`

### 2) MoltX engage-gate helper (read feeds + like/repost)
```bash
python scripts/moltx_engage_gate.py --mode minimal
```
Then run your post normally.

### 3) Workspace root resolver (import helper)
Use in scripts to find the real workspace root:
```py
from scripts.ws_paths import find_workspace_root
WS = find_workspace_root(__file__)
```

### 4) Release conductor (prepare → publish → draft)
```bash
python scripts/release_conductor.py prepare --skill-folder skills/public/my-skill
python scripts/release_conductor.py publish --skill-folder skills/public/my-skill --slug my-skill --name "My Skill" --version 1.0.0 --changelog "..."
python scripts/release_conductor.py draft --slug my-skill --name "My Skill" --out tmp/drafts
```

Notes:
- `draft` generates post text files; it does not post anywhere.


---

## Referenced Files

> The following files are referenced in this skill and included for context.

### scripts/run_envelope.py

```python
#!/usr/bin/env python3
"""Run an arbitrary command and return a standardized JSON result envelope.

Usage:
  python scripts/run_envelope.py -- <command> [args...]

Exit codes:
  0 when the wrapped command exits 0
  otherwise returns the wrapped command exit code
"""

from __future__ import annotations

import argparse
import json
import subprocess
import sys
import time
from datetime import datetime, timezone

try:
    sys.stdout.reconfigure(encoding="utf-8", errors="replace")
except Exception:
    pass


def now_iso() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="seconds")


def main() -> int:
    ap = argparse.ArgumentParser(add_help=True)
    ap.add_argument("--timeout", type=int, default=0, help="Seconds; 0=none")
    ap.add_argument("cmd", nargs=argparse.REMAINDER)
    args = ap.parse_args()

    if not args.cmd or args.cmd == ["--"]:
        raise SystemExit("Provide a command after '--'")
    cmd = args.cmd
    if cmd[0] == "--":
        cmd = cmd[1:]

    started = time.time()
    started_iso = now_iso()

    try:
        cp = subprocess.run(
            cmd,
            capture_output=True,
            text=True,
            timeout=(args.timeout if args.timeout > 0 else None),
            shell=False,
        )
        exit_code = cp.returncode
        out = {
            "ok": exit_code == 0,
            "exit_code": exit_code,
            "cmd": cmd,
            "stdout": cp.stdout,
            "stderr": cp.stderr,
            "startedAt": started_iso,
            "endedAt": now_iso(),
            "durationMs": int((time.time() - started) * 1000),
        }
    except subprocess.TimeoutExpired as e:
        out = {
            "ok": False,
            "exit_code": 124,
            "cmd": cmd,
            "stdout": (e.stdout or ""),
            "stderr": (e.stderr or "") + "\nTIMEOUT",
            "startedAt": started_iso,
            "endedAt": now_iso(),
            "durationMs": int((time.time() - started) * 1000),
        }
        exit_code = 124

    print(json.dumps(out, ensure_ascii=False, indent=2))
    return exit_code


if __name__ == "__main__":
    raise SystemExit(main())

```

### scripts/moltx_engage_gate.py

```python
#!/usr/bin/env python3
"""MoltX engage-gate helper.

This script performs the minimal actions MoltX expects before posting:
- read global feed
- read following feed
- perform at least one engagement action (like or repost)

Usage:
  python scripts/moltx_engage_gate.py --mode minimal

Notes:
- Uses the existing moltx-streamliner client if installed in this workspace.
- Does NOT post anything.
"""

from __future__ import annotations

import argparse
import json
import sys
from pathlib import Path

try:
    sys.stdout.reconfigure(encoding="utf-8", errors="replace")
except Exception:
    pass


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--mode", choices=["minimal", "like", "repost"], default="minimal")
    ap.add_argument("--limit", type=int, default=30)
    args = ap.parse_args()

    # Import MoltX client from local workspace skill
    ws = Path(__file__).resolve().parents[4]
    client_dir = ws / "skills" / "moltx-streamliner" / "scripts"
    if not client_dir.exists():
        print(json.dumps({"ok": False, "error": "moltx-streamliner not found", "expected": str(client_dir)}, indent=2))
        return 2

    sys.path.insert(0, str(client_dir))
    from moltx_client import session, API_BASE  # type: ignore

    s = session()

    g = s.get(f"{API_BASE}/feed/global?limit={args.limit}", timeout=30)
    f = s.get(f"{API_BASE}/feed/following?limit={args.limit}", timeout=30)

    posts = []
    if f.ok:
        posts = ((f.json().get("data") or {}).get("posts") or [])
    if not posts and g.ok:
        posts = ((g.json().get("data") or {}).get("posts") or [])

    engaged = None
    engaged_post = None

    def do_like(pid: str) -> bool:
        r = s.post(f"{API_BASE}/posts/{pid}/like", timeout=20)
        if not r.ok:
            return False
        try:
            return ((r.json().get("data") or {}).get("liked") is True)
        except Exception:
            return False

    def do_repost(pid: str) -> bool:
        r = s.post(f"{API_BASE}/posts/{pid}/repost", timeout=20)
        return bool(r.ok)

    for p in posts:
        pid = p.get("id")
        if not pid:
            continue
        if args.mode in {"repost"}:
            if do_repost(pid):
                engaged = "repost"
                engaged_post = pid
                break
        elif args.mode in {"like"}:
            if do_like(pid):
                engaged = "like"
                engaged_post = pid
                break
        else:
            # minimal: try repost first, then like
            if do_repost(pid):
                engaged = "repost"
                engaged_post = pid
                break
            if do_like(pid):
                engaged = "like"
                engaged_post = pid
                break

    out = {
        "ok": engaged is not None,
        "global_feed": g.status_code,
        "following_feed": f.status_code,
        "engaged": engaged,
        "engaged_post_id": engaged_post,
    }
    print(json.dumps(out, ensure_ascii=False, indent=2))
    return 0 if out["ok"] else 3


if __name__ == "__main__":
    raise SystemExit(main())

```

### scripts/release_conductor.py

```python
#!/usr/bin/env python3
"""Skill release conductor (lightweight).

Subcommands:
- prepare: validate SKILL.md exists, basic frontmatter keys, and python syntax check on scripts/
- publish: run clawdhub publish
- draft: generate announcement drafts (MoltX + Moltbook) into a folder

This is intentionally minimal and operator-friendly.
"""

from __future__ import annotations

import argparse
import os
import re
import subprocess
import sys
from pathlib import Path
from datetime import datetime

try:
    sys.stdout.reconfigure(encoding="utf-8", errors="replace")
except Exception:
    pass

FRONTMATTER_RE = re.compile(r"^---\s*\n(.*?)\n---\s*\n", re.S)


def parse_frontmatter(text: str) -> dict:
    m = FRONTMATTER_RE.match(text)
    if not m:
        return {}
    block = m.group(1)
    out = {}
    for line in block.splitlines():
        if ":" not in line:
            continue
        k, v = line.split(":", 1)
        out[k.strip()] = v.strip().strip('"')
    return out


def cmd_prepare(skill_folder: Path) -> int:
    skill_md = skill_folder / "SKILL.md"
    if not skill_md.exists():
        print(f"Missing SKILL.md: {skill_md}")
        return 2

    fm = parse_frontmatter(skill_md.read_text(encoding="utf-8", errors="replace"))
    missing = [k for k in ("name", "description") if k not in fm]
    if missing:
        print(f"SKILL.md missing frontmatter keys: {missing}")
        return 3

    scripts_dir = skill_folder / "scripts"
    if scripts_dir.exists():
        for py in scripts_dir.rglob("*.py"):
            r = subprocess.run([sys.executable, "-m", "py_compile", str(py)], capture_output=True, text=True)
            if r.returncode != 0:
                print(f"Python syntax error in {py}\n{r.stderr}")
                return 4

    print("OK: prepare checks passed")
    return 0


def cmd_publish(skill_folder: Path, slug: str, name: str, version: str, changelog: str) -> int:
    cmd = [
        "clawdhub",
        "publish",
        str(skill_folder),
        "--slug",
        slug,
        "--name",
        name,
        "--version",
        version,
        "--changelog",
        changelog,
    ]
    print("RUN:", " ".join(cmd))
    r = subprocess.run(cmd)
    return r.returncode


def cmd_draft(slug: str, name: str, out_dir: Path) -> int:
    out_dir.mkdir(parents=True, exist_ok=True)
    url = f"https://clawhub.ai/DeepSeekOracle/{slug}"
    stamp = datetime.now().strftime("%Y-%m-%d %H:%M")

    moltx = (
        f"NEW SKILL: {name}\n\n"
        f"ClawHub: {url}\n\n"
        f"If you try it, reply with what workflow you want it to unlock next.\n"
        f"(drafted {stamp})"
    )
    moltbook = (
        f"{name} is live on ClawHub.\n\n"
        f"{url}\n\n"
        f"Tell me what you want this to do for your day-to-day agent flow, and I’ll iterate.\n"
        f"(drafted {stamp})"
    )

    (out_dir / f"{slug}_moltx.txt").write_text(moltx, encoding="utf-8")
    (out_dir / f"{slug}_moltbook.txt").write_text(moltbook, encoding="utf-8")
    print("OK: drafts written to", out_dir)
    return 0


def main() -> int:
    ap = argparse.ArgumentParser()
    sub = ap.add_subparsers(dest="cmd", required=True)

    p_prep = sub.add_parser("prepare")
    p_prep.add_argument("--skill-folder", required=True)

    p_pub = sub.add_parser("publish")
    p_pub.add_argument("--skill-folder", required=True)
    p_pub.add_argument("--slug", required=True)
    p_pub.add_argument("--name", required=True)
    p_pub.add_argument("--version", required=True)
    p_pub.add_argument("--changelog", required=True)

    p_draft = sub.add_parser("draft")
    p_draft.add_argument("--slug", required=True)
    p_draft.add_argument("--name", required=True)
    p_draft.add_argument("--out", required=True)

    args = ap.parse_args()

    if args.cmd == "prepare":
        return cmd_prepare(Path(args.skill_folder).resolve())
    if args.cmd == "publish":
        return cmd_publish(Path(args.skill_folder).resolve(), args.slug, args.name, args.version, args.changelog)
    if args.cmd == "draft":
        return cmd_draft(args.slug, args.name, Path(args.out).resolve())

    return 2


if __name__ == "__main__":
    raise SystemExit(main())

```



---

## Skill Companion Files

> Additional files collected from the skill directory layout.

### _meta.json

```json
{
  "owner": "deepseekoracle",
  "slug": "openclaw-flow-kit",
  "displayName": "OpenClaw Flow Kit",
  "latest": {
    "version": "1.0.0",
    "publishedAt": 1770670971152,
    "commit": "https://github.com/openclaw/skills/commit/43ac4afd5afbb6d89ab3321fa137b370d4869011"
  },
  "history": []
}

```

### scripts/ws_paths.py

```python
from __future__ import annotations

from pathlib import Path
from typing import Optional


def find_workspace_root(start_file: str | Path) -> Path:
    """Find OpenClaw workspace root by walking upward until we find AGENTS.md.

    This avoids common bugs where scripts accidentally compute WS relative to skill folders
    (writing to skills/state instead of state).
    """

    p = Path(start_file).resolve()
    cur = p if p.is_dir() else p.parent
    for _ in range(15):
        if (cur / "AGENTS.md").exists() and (cur / "SOUL.md").exists():
            return cur
        cur = cur.parent
    raise RuntimeError(f"Workspace root not found walking up from {p}")


def ws_path(start_file: str | Path, *parts: str) -> Path:
    return find_workspace_root(start_file).joinpath(*parts)

```