@clawhub-zhaobudaoyuema-2d083f55c2
Agent instructions: pack/apply OpenClaw user data via scripts; overwrite-by-path only. You dry-run first, read EXPORT_MANIFEST.txt, gate optional layers, res...
---
name: openclaw-user-data-pack
version: 1.0.5
description: "Agent instructions: pack/apply OpenClaw user data via scripts; overwrite-by-path only. You dry-run first, read EXPORT_MANIFEST.txt, gate optional layers, resolve merge conflicts yourself—never imply scripts merge."
trigger: "OpenClaw backup, export user data, pack workspace, migration zip, 打包 openclaw, 迁移记忆, openclaw 一键导出, 一键应用, 导入 zip, restore openclaw from zip, 新机器恢复 openclaw"
---
# OpenClaw agent: pack and apply user data
**Who reads this:** you are the **OpenClaw agent** (runtime). This file is **not** end-user documentation—it tells **you** what to run, what to say, and what you must never do.
**Language:** reply to the user in **their** language; keep technical identifiers (paths, flags) as in the scripts.
---
## Your job in one sentence
Use `scripts/pack_openclaw.py` and `scripts/apply_openclaw.py` from this skill to export or restore workspace data (and optional layers **only** if the user clearly opts in after you warn them). **You** own preview, collision handling, and consent—the scripts only write files by path.
---
## When the user asks to export (pack)
1. Run `pip install -r requirements.txt` if dependencies may be missing.
2. Run `python scripts/pack_openclaw.py --dry-run` with the same flags you plan for the real pack; show the user what paths would be included.
3. Explain: default pack is **`workspace/`** only. List optional layers (`--managed-skills`, session flags, config snapshot flags) and **do not add any** until the user **separately** approves each, after you give the short risk line (size, transcripts, secrets)—see **Before any real disk write**.
4. Run the real pack: `python scripts/pack_openclaw.py` with **only** approved flags.
5. Give the user the zip path. **Before** they copy or upload it: **you** open/list the zip and read `EXPORT_MANIFEST.txt`; confirm it matches what you promised (paths + layers).
---
## When the user asks to import (apply)
1. If the zip is not clearly from a trusted source or from this skill’s pack layout (`workspace/`, `EXPORT_MANIFEST.txt`, …), **stop** and say why you will not apply it without their confirmation.
2. Run `pip install -r requirements.txt` if needed.
3. Tell the user to **back up** `$OPENCLAW_HOME` (or `%USERPROFILE%\.openclaw`) and the target workspace—or apply to a throwaway copy—unless they **explicitly** accept overwrite risk after you state it once.
4. **You** read `EXPORT_MANIFEST.txt` inside the zip, then run
`python scripts/apply_openclaw.py --zip <path> --dry-run`
with `--openclaw-home`, `--workspace`, and `--config` as the environment needs. Treat the combined manifest + dry-run output as the write contract.
5. Walk the user through which paths would be created/overwritten. For overlaps on **memory / persona / skills**, follow **Merge and conflicts (your work; not in scripts)**—**do not** run non–dry-run apply on a live workspace until conflicts are resolved or the user **explicitly** chooses full replace for that subtree.
6. Add `--apply-managed-skills`, session flags, or `--apply-config` **only** after separate approval **and** the warnings in **Before any real disk write**.
7. Run apply **without** `--dry-run` only when the above is satisfied. If config was restored, remind: they still need valid auth on this machine; old paths inside `openclaw.json` may be wrong here.
8. Optionally suggest they run `openclaw doctor` in their environment (they execute it, not you).
---
## Before any real disk write (you follow this order)
Skip a step **only** if the user opts out **after** you repeat the concrete risk.
1. **Dry-run first** — pack and apply both support `--dry-run`. The printed paths are what a real run would touch.
2. **Read `EXPORT_MANIFEST.txt` in the zip** — authoritative list of packed paths; pair with apply dry-run to see destination collisions.
3. **Backups** — dry-run does not change disk; it is not a backup. For apply, insist on backup or throwaway target unless they waive.
4. **Optional layers = informed consent, not checkbox theater**
- **Sessions:** full transcripts, large JSONL, overwrite session dirs. Do not pass pack/apply session flags unless the user understands that.
- **Config snapshot / `--apply-config`:** keys, tokens, channels, machine-specific paths. Do not enable without that acknowledgment.
5. **Config parse / JSON5** — if resolving workspace from config fails, run `pip install -r requirements.txt` (includes `json5`) or pass `--workspace` explicitly.
---
## What the scripts actually do (so you do not mislead)
- Pack and apply are **filesystem** steps: extract or copy bytes to paths. **No** semantic merge, **no** three-way merge, **no** conflict UI in Python.
- **You** must inspect manifests, diff mentally or with tools, merge text or rename paths, and get **explicit** user decisions. **Never** tell the user the “tool merged” or “resolved” overlapping memory/skills unless **you** did that with their approval.
If you follow previews + consent + collision handling, you can honestly say the flow is transparent; if you skip that, you risk silent data loss.
---
## Safety: what you must assume and say
- Assume the archive may hold sensitive material: persona, `MEMORY.md`, logs, workspace skills; with optional layers, session JSONL and `openclaw.json` (secrets, channels).
- **Do not** pack or encourage packing `~/.openclaw/credentials/`. Apply never writes credentials; tell the user they must re-login / re-pair on a new machine unless they consciously accept copying secrets (you still do not pack credentials via these scripts).
- Warn against putting the zip on untrusted or public storage.
- **Overwrite rule:** same path ⇒ destination file replaced. Same path ≠ same meaning. Only `openclaw.json` gets a `.bak.<timestamp>` when using `--apply-config`; **other paths are not auto-backed up.**
### Merge and conflicts (your work; not in scripts)
- A path is an address, not proof two files are equivalent. Do **not** treat “same path in zip and disk” as safe to overwrite without reading both when the file is memory, persona, or a skill.
- **Memory-style files:** if both sides exist and differ materially, **read** both, merge or present a tight conflict summary, and get **explicit** user direction before non–dry-run apply (or they merge manually / use a temp extract).
- **Skills (`SKILL.md` etc.):** divergent purpose or triggers ⇒ **do not** pick a winner alone; offer keep local / take zip / merge / rename path so both can exist.
- **Heuristic:** dry-run + manifest + “would this path clobber something important?” ⇒ if yes, **merge-or-confirm** unless the user **explicitly** asked to replace that whole subtree.
---
## Pack: default vs optional
| Content | Path inside zip | In default pack? |
|---------|-------------------|------------------|
| Workspace (persona, memory, workspace skills, canvas, etc.) | `workspace/` | yes |
| Managed skills | `managed-skills/` | no — `--managed-skills` |
| Sessions | `sessions/<agentId>/sessions/` | no — session flags + acknowledgement; **large, sensitive, full transcripts** |
| Config snapshot | `config/openclaw.json` | no — config flags + acknowledgement; **secrets, machine paths** |
| Credentials | n/a | **never** |
---
## Apply: default vs optional
Match flags to what is in the zip. If a layer is in the zip but flags are missing, the script warns and skips that layer.
| Content | Action | Default apply? |
|---------|--------|----------------|
| Workspace | Extract `workspace/*` → target workspace | yes, unless `--no-apply-workspace` |
| Managed skills | → `<openclaw-home>/skills/` | no — `--apply-managed-skills` |
| Sessions | → `<openclaw-home>/agents/<id>/sessions/` | no — `--apply-sessions` + `--i-know-restoring-sessions-overwrites` |
| Config | → `<openclaw-home>/openclaw.json` (existing → `.bak.<timestamp>`) | no — `--apply-config` + `--i-know-config-overwrites-secrets` |
---
## Paths (how you resolve them)
- OpenClaw home: `$OPENCLAW_HOME` or `~/.openclaw`; Windows: `%USERPROFILE%\.openclaw`.
- Pack: if `--workspace` omitted, script reads config. Apply: `--workspace` may create the dir; if omitted, config must parse. On a fresh machine, prefer `openclaw onboard` or pass `--workspace` explicitly.
- Run pack and apply in the **same** environment family (e.g. both WSL) so paths mean the same thing.
---
## Examples (you adapt paths for the user’s OS)
Workspace-only apply, dry-run then real:
```bash
python scripts/apply_openclaw.py --zip ./openclaw-user-export-xxx.zip \
--openclaw-home ~/.openclaw \
--workspace ~/.openclaw/workspace \
--dry-run
python scripts/apply_openclaw.py --zip ./openclaw-user-export-xxx.zip \
--openclaw-home ~/.openclaw \
--workspace ~/.openclaw/workspace
```
All optional apply layers — **only** after the user approved **each** flag’s risk:
```bash
python scripts/apply_openclaw.py --zip ./export.zip \
--openclaw-home ~/.openclaw --workspace ~/.openclaw/workspace \
--apply-managed-skills \
--apply-sessions --i-know-restoring-sessions-overwrites \
--apply-config --i-know-config-overwrites-secrets
```
---
## CLI reference (copy-paste skeletons)
**Pack:**
```text
python scripts/pack_openclaw.py [--workspace PATH] [--openclaw-home PATH] [--config PATH]
[-o FILE.zip] [--exclude-git | --no-exclude-git] [--managed-skills]
[--sessions --i-know-sessions-are-large-and-sensitive]
[--config-snapshot --i-know-config-may-contain-secrets]
[--dry-run] [--manifest-sha256] [--sha256-max-mb N]
```
**Apply:**
```text
python scripts/apply_openclaw.py --zip FILE.zip [--openclaw-home PATH] [--workspace PATH] [--config PATH]
[--no-apply-workspace] [--apply-managed-skills]
[--apply-sessions --i-know-restoring-sessions-overwrites]
[--apply-config --i-know-config-overwrites-secrets]
[--dry-run]
```
---
## When to activate this skill (trigger hints)
| Intent | Example user phrases |
|--------|----------------------|
| Export | backup workspace, export memory, pack openclaw |
| Import | new PC restore, import zip, apply backup, restore openclaw |
| Chinese | 一键打包, 一键应用, 导入 zip, 迁移 |
---
## Quick commands (you run from skill root)
| Goal | Command |
|------|---------|
| Pack preview | `python scripts/pack_openclaw.py --dry-run` |
| Pack | `python scripts/pack_openclaw.py` |
| Apply preview | `python scripts/apply_openclaw.py --zip x.zip --dry-run` |
| Apply | `python scripts/apply_openclaw.py --zip x.zip` |
| Dependencies | `pip install -r requirements.txt` |
FILE:package.json
{
"name": "openclaw-user-data-pack",
"version": "1.0.5",
"description": "OpenClaw skill: Pack user data to zip with manifest; apply zip to a new OpenClaw home/workspace (gated optional layers).",
"keywords": [
"openclaw",
"skill",
"backup",
"export",
"import",
"restore",
"migration"
],
"license": "MIT",
"files": [
"SKILL.md",
"README.md",
"README_CN.md",
"scripts",
"requirements.txt",
"publish_npm.py"
]
}
FILE:publish_npm.py
#!/usr/bin/env python3
"""
Publish this skill to npm (and optionally ClawHub).
Before publish:
1. Aligns local package.json to npm if registry is ahead
2. Bumps patch version (x.y.z -> x.y.(z+1))
3. Syncs version to package.json, SKILL.md frontmatter, scripts/__init__.py
4. Runs npm publish
5. If `clawhub` is on PATH and SKIP_CLAWHUB is not set: clawhub publish
6. Optional hook: skill.ps1 / skill.cmd / skill.bat (Windows) or skill.sh (Unix)
Usage:
python publish_npm.py
python publish_npm.py "changelog text"
Prerequisites: npm login; clawhub login (for step 5). On Windows, use plain python from repo root.
"""
from __future__ import annotations
import json
import os
import re
import shutil
import subprocess
import sys
from typing import Optional
ROOT = os.path.dirname(os.path.abspath(__file__))
PACKAGE_JSON = os.path.join(ROOT, "package.json")
SKILL_MD = os.path.join(ROOT, "SKILL.md")
INIT_PY = os.path.join(ROOT, "scripts", "__init__.py")
def bump_patch_version(version: str) -> str:
parts = version.strip().split(".")
if len(parts) < 3:
return f"{version}.1" if len(parts) == 1 else f"{version}.0"
try:
parts[2] = str(int(parts[2]) + 1)
return ".".join(parts)
except ValueError:
return version
def get_npm_version(package_name: str) -> Optional[str]:
try:
result = subprocess.run(
["npm", "view", package_name, "version"],
capture_output=True,
text=True,
cwd=ROOT,
shell=(os.name == "nt"),
timeout=15,
)
if result.returncode == 0 and result.stdout.strip():
return result.stdout.strip()
except (subprocess.TimeoutExpired, FileNotFoundError):
pass
return None
def sync_skill_md(new_version: str) -> bool:
if not os.path.isfile(SKILL_MD):
return False
with open(SKILL_MD, encoding="utf-8") as f:
content = f.read()
new_content, n = re.subn(r"(version:\s*)[\d.]+", r"\g<1>" + new_version, content, count=1)
if n == 0:
new_content, m = re.subn(
r"(name:\s*\S+\n)",
r"\1version: " + new_version + "\n",
content,
count=1,
)
if m == 0:
return False
with open(SKILL_MD, "w", encoding="utf-8") as f:
f.write(new_content)
return True
def sync_init_py(new_version: str) -> bool:
if not os.path.isfile(INIT_PY):
return False
with open(INIT_PY, encoding="utf-8") as f:
content = f.read()
new_content, n = re.subn(
r'(__version__\s*=\s*")[\d.]+(")',
r"\g<1>" + new_version + r"\2",
content,
count=1,
)
if n == 0:
return False
with open(INIT_PY, "w", encoding="utf-8") as f:
f.write(new_content)
return True
def run_optional_hook(version: str, changelog: str) -> None:
if os.name == "nt":
for name in ("skill.ps1", "skill.cmd", "skill.bat"):
p = os.path.join(ROOT, name)
if os.path.isfile(p):
if name.endswith(".ps1"):
subprocess.run(
["powershell", "-NoProfile", "-ExecutionPolicy", "Bypass", "-File", p, version, changelog],
cwd=ROOT,
shell=False,
)
else:
subprocess.run([p, version, changelog], cwd=ROOT, shell=True)
print(f"ran hook: {name}")
return
else:
p = os.path.join(ROOT, "skill.sh")
if os.path.isfile(p) and os.access(p, os.X_OK):
subprocess.run([p, version, changelog], cwd=ROOT)
print("ran hook: skill.sh")
def clawhub_publish(version: str, changelog: str) -> int:
exe = shutil.which("clawhub")
if not exe:
print("clawhub not on PATH — skip (install: npm i -g clawhub)")
return 0
slug = os.environ.get("CLAWHUB_SLUG", "openclaw-user-data-pack")
name = os.environ.get("CLAWHUB_NAME", "OpenClaw User Data Pack")
cmd = [
exe,
"publish",
".",
"--slug",
slug,
"--name",
name,
"--version",
version,
"--changelog",
changelog,
"--tags",
"latest",
]
return subprocess.run(cmd, cwd=ROOT, shell=(os.name == "nt")).returncode
def main() -> None:
changelog = sys.argv[1] if len(sys.argv) > 1 else None
if not os.path.isfile(SKILL_MD):
print("SKILL.md not found in project root", file=sys.stderr)
sys.exit(1)
if not os.path.isfile(PACKAGE_JSON):
print("package.json not found in project root", file=sys.stderr)
sys.exit(1)
with open(PACKAGE_JSON, encoding="utf-8") as f:
pkg = json.load(f)
pkg_name = pkg.get("name", "")
local_version = pkg.get("version", "1.0.0")
npm_version = get_npm_version(pkg_name) if pkg_name else None
if npm_version:
print(f"npm current version: {npm_version}")
base = npm_version if npm_version else local_version
if npm_version and local_version != npm_version:
pkg["version"] = npm_version
with open(PACKAGE_JSON, "w", encoding="utf-8") as f:
json.dump(pkg, f, indent=2, ensure_ascii=False)
f.write("\n")
print(f"Local synced to npm version: {npm_version}")
new_version = bump_patch_version(base)
pkg["version"] = new_version
with open(PACKAGE_JSON, "w", encoding="utf-8") as f:
json.dump(pkg, f, indent=2, ensure_ascii=False)
f.write("\n")
print(f"Version bumped: {base} -> {new_version}")
if sync_skill_md(new_version):
print(f"SKILL.md version synced: {new_version}")
else:
print("Warning: SKILL.md version not updated", file=sys.stderr)
if sync_init_py(new_version):
print(f"scripts/__init__.py version synced: {new_version}")
else:
print("Warning: scripts/__init__.py version not updated", file=sys.stderr)
msg = changelog if changelog else f"Release {new_version}"
if os.environ.get("SKIP_NPM") == "1":
print("[SKIP_NPM=1] skip npm publish")
else:
result = subprocess.run(["npm", "publish", "--access", "public"], cwd=ROOT, shell=(os.name == "nt"))
if result.returncode != 0:
print(
"npm publish failed. Try: npm login. 403: enable 2FA or granular token.",
file=sys.stderr,
)
sys.exit(result.returncode)
print(f"Published to npm: {new_version}")
if os.environ.get("SKIP_CLAWHUB") == "1":
print("[SKIP_CLAWHUB=1] skip clawhub")
else:
rc = clawhub_publish(new_version, msg)
if rc != 0:
print("clawhub publish failed", file=sys.stderr)
sys.exit(rc)
print(f"Published to ClawHub: {new_version}")
run_optional_hook(new_version, msg)
print(f"Done: {new_version}")
if __name__ == "__main__":
main()
FILE:README.md
# openclaw-user-data-pack
An OpenClaw skill that exports user-owned data (workspace memory, persona, skills, and more) into a single zip with a manifest, and can apply that zip onto a new OpenClaw home or workspace. Optional layers such as managed skills, session transcripts, and config snapshots are opt-in on both export and import so secrets and large logs are not moved by accident.
**Behavior:** pack/apply are **overwrite-by-path** filesystem operations; they do **not** merge conflicting memory or skills. Use `--dry-run`, inspect `EXPORT_MANIFEST.txt` inside the zip, back up the destination before apply, and only enable session/config flags if you understand transcripts and secrets—see `SKILL.md` for the full operator checklist.
Usage and agent instructions live in `SKILL.md`. Chinese readme: [README_CN.md](README_CN.md).
## Copy and send to OpenClaw
**ClawHub (recommended, international)**
```text
Please run clawhub install openclaw-user-data-pack to install this skill, and help me export or apply my OpenClaw user data (zip with manifest, optional managed skills / sessions / config).
```
**npm**
```text
Please run npm i openclaw-user-data-pack to install this skill, and help me export or apply my OpenClaw user data (zip with manifest, optional managed skills / sessions / config).
```
## Bai Xia Da Zhan
A tongue-in-cheek roster of “lobster-adjacent” names in the wild (for fun; no affiliation with this project):
1. 腾讯 WorkBuddy
2. 腾讯 QClaw
3. 腾讯龙虾管家
4. 腾讯云保安
5. 腾讯乐享知识库 · 龙虾版
6. 字节 ArkClaw
7. 智谱 AutoClaw
8. 月之暗面 Kimi Claw
9. 阿里云 CoPaw
10. 阿里云 JVSClaw
11. 阿里云 QoderWork
12. 百度红手指 Operator
13. 百度 DuClaw
14. 科大讯飞 AstronClaw
15. MiniMax MaxClaw
16. 网易有道 LobsterAI
17. 当贝 Molili
18. 智麻 ChatClaw
19. 矽速 PicoClaw
20. 博云 BocLaw
21. ZeroClaw
22. 万得 WindClaw
23. 小米 MiClaw
24. 猎豹 EasyClaw
25. 猎豹元气AI Bot
26. 京东灵犀Claw
27. 快手 KClaw
28. 美图Claw
29. 360安全Claw
30. 商汤 SenseClaw
31. 华为小艺Claw
32. ToDesk ToClaw
## License
MIT
FILE:README_CN.md
# openclaw-user-data-pack（中文版）
面向 OpenClaw 的技能：把用户自有数据（工作区记忆、人设、技能等）打成带清单的 zip，并可在新环境一键应用回 `~/.openclaw` 与工作区。托管技能、会话、配置等可选层在导出与导入时均需显式选择，避免误迁移密钥与大体积会话。
**行为说明：** 打包/应用是按路径**覆盖写入**的文件操作，脚本**不会**对冲突的记忆或技能做语义合并。请先用 `--dry-run`、打开 zip 核对 `EXPORT_MANIFEST.txt`、应用前备份目标目录；仅在理解会话全文与配置中的密钥后再开启会话/配置相关开关——完整清单见 `SKILL.md`。
具体用法与代理说明见 `SKILL.md`。英文说明：[README.md](README.md)。
## 复制发给 OpenClaw
**ClawHub（推荐，国际）**
```text
请执行 clawhub install openclaw-user-data-pack 安装本 skill，并帮我打包或恢复 OpenClaw 用户数据（带清单的 zip，可选托管技能/会话/配置）。
```
**npm**
```text
请执行 npm i openclaw-user-data-pack 安装本 skill，并帮我打包或恢复 OpenClaw 用户数据（带清单的 zip，可选托管技能/会话/配置）。
```
## 百虾大战
江湖传闻中的「龙虾系」友商名录（娱乐向排序，与本品无隶属关系）：
1. 腾讯 WorkBuddy
2. 腾讯 QClaw
3. 腾讯龙虾管家
4. 腾讯云保安
5. 腾讯乐享知识库 · 龙虾版
6. 字节 ArkClaw
7. 智谱 AutoClaw
8. 月之暗面 Kimi Claw
9. 阿里云 CoPaw
10. 阿里云 JVSClaw
11. 阿里云 QoderWork
12. 百度红手指 Operator
13. 百度 DuClaw
14. 科大讯飞 AstronClaw
15. MiniMax MaxClaw
16. 网易有道 LobsterAI
17. 当贝 Molili
18. 智麻 ChatClaw
19. 矽速 PicoClaw
20. 博云 BocLaw
21. ZeroClaw
22. 万得 WindClaw
23. 小米 MiClaw
24. 猎豹 EasyClaw
25. 猎豹元气AI Bot
26. 京东灵犀Claw
27. 快手 KClaw
28. 美图Claw
29. 360安全Claw
30. 商汤 SenseClaw
31. 华为小艺Claw
32. ToDesk ToClaw
## 许可
MIT
FILE:requirements.txt
# Optional: parse ~/.openclaw/openclaw.json (JSON5 with comments)
json5>=0.9.0
FILE:scripts/apply_openclaw.py
#!/usr/bin/env python3
"""
Apply a zip produced by pack_openclaw.py into a local OpenClaw home + workspace.
Never writes to credentials/. Rejects path-traversal archive names.
"""
from __future__ import annotations
import argparse
import shutil
import sys
from datetime import datetime
from pathlib import Path
from zipfile import ZipFile, ZipInfo
from openclaw_paths import openclaw_home, resolve_workspace
def _arc_parts(name: str) -> list[str]:
return [p for p in name.replace("\\", "/").split("/") if p != ""]
def _is_safe_arcname(name: str) -> bool:
if not name or name.endswith("/"):
return False
parts = _arc_parts(name)
return ".." not in parts
def _classify_arcname(arcname: str) -> tuple[str, Path] | None:
"""
Returns (layer, relative Path inside that layer) for file members.
layer is workspace | managed | sessions | config
For sessions, relative path is agent_id / rest_under_sessions_dir
"""
if not _is_safe_arcname(arcname):
return None
parts = _arc_parts(arcname)
if not parts:
return None
if parts[0] == "EXPORT_MANIFEST.txt":
return None
if parts[0] == "workspace":
if len(parts) < 2:
return None
return ("workspace", Path(*parts[1:]))
if parts[0] == "managed-skills":
if len(parts) < 2:
return None
return ("managed", Path(*parts[1:]))
if parts[0] == "sessions":
# sessions/<agentId>/sessions/<...>
if len(parts) < 4 or parts[2] != "sessions":
return None
agent_id = parts[1]
rest = Path(*parts[3:]) if len(parts) > 3 else Path()
return ("sessions", Path(agent_id) / rest)
if parts[0] == "config" and len(parts) == 2 and parts[1] == "openclaw.json":
return ("config", Path("openclaw.json"))
return None
def _resolve_dest(
layer: str,
rel: Path,
*,
workspace: Path,
home: Path,
) -> Path | None:
ws = workspace.resolve()
hm = home.resolve()
if layer == "workspace":
dest = (ws / rel).resolve()
try:
dest.relative_to(ws)
except ValueError:
return None
return dest
if layer == "managed":
base = (hm / "skills").resolve()
dest = (base / rel).resolve()
try:
dest.relative_to(base)
except ValueError:
return None
return dest
if layer == "sessions":
# rel = agent_id / file...
parts = rel.parts
if len(parts) < 1:
return None
agent_id = parts[0]
tail = Path(*parts[1:]) if len(parts) > 1 else Path()
base = (hm / "agents" / agent_id / "sessions").resolve()
dest = (base / tail).resolve()
try:
dest.relative_to((hm / "agents").resolve())
except ValueError:
return None
if agent_id == ".." or ".." in agent_id:
return None
return dest
if layer == "config":
return (hm / "openclaw.json").resolve()
return None
def _gather_plan(
zf: ZipFile,
*,
workspace: Path,
home: Path,
apply_workspace: bool,
apply_managed: bool,
apply_sessions: bool,
apply_config: bool,
) -> tuple[list[tuple[str, str, Path]], list[str]]:
"""Returns (ops: arcname, layer, dest), skips)."""
ops: list[tuple[str, str, Path]] = []
skips: list[str] = []
names = {zi.filename.replace("\\", "/"): zi for zi in zf.infolist()}
for arcname, zi in sorted(names.items()):
if zi.is_dir():
continue
cl = _classify_arcname(arcname)
if cl is None:
if arcname != "EXPORT_MANIFEST.txt" and not arcname.endswith("EXPORT_MANIFEST.txt"):
skips.append(f"unrecognized (skipped): {arcname}")
continue
layer, rel = cl
if layer == "workspace" and not apply_workspace:
skips.append(f"skipped (--no-apply-workspace): {arcname}")
continue
if layer == "managed" and not apply_managed:
skips.append(f"skipped (need --apply-managed-skills): {arcname}")
continue
if layer == "sessions" and not apply_sessions:
skips.append(f"skipped (need --apply-sessions + ack): {arcname}")
continue
if layer == "config" and not apply_config:
skips.append(f"skipped (need --apply-config + ack): {arcname}")
continue
dest = _resolve_dest(layer, rel, workspace=workspace, home=home)
if dest is None:
skips.append(f"unsafe path (skipped): {arcname}")
continue
ops.append((arcname, layer, dest))
return ops, skips
def _write_member(zf: ZipFile, zi: ZipInfo, dest: Path, *, dry_run: bool) -> None:
dest.parent.mkdir(parents=True, exist_ok=True)
if dry_run:
return
with zf.open(zi, "r") as src:
data = src.read()
dest.write_bytes(data)
def main() -> None:
parser = argparse.ArgumentParser(
description="Apply openclaw-user-export zip (from pack_openclaw.py) into local OpenClaw dirs."
)
parser.add_argument("--zip", "-z", type=Path, required=True, help="Path to export .zip")
parser.add_argument(
"--workspace",
type=Path,
default=None,
help="Target workspace root (created if missing when passed; else resolve from config)",
)
parser.add_argument(
"--openclaw-home",
type=Path,
default=None,
help="OpenClaw data directory (default: $OPENCLAW_HOME or ~/.openclaw)",
)
parser.add_argument(
"--config",
type=Path,
default=None,
help="Path to openclaw.json for workspace resolution (default: <openclaw-home>/openclaw.json)",
)
parser.add_argument(
"--no-apply-workspace",
action="store_true",
help="Do not extract workspace/ from zip",
)
parser.add_argument(
"--apply-managed-skills",
action="store_true",
help="Extract managed-skills/ into ~/.openclaw/skills/",
)
parser.add_argument(
"--apply-sessions",
action="store_true",
help="Extract sessions/ into ~/.openclaw/agents/<id>/sessions/",
)
parser.add_argument(
"--i-know-restoring-sessions-overwrites",
action="store_true",
help="Required with --apply-sessions",
)
parser.add_argument(
"--apply-config",
action="store_true",
help="Write config/openclaw.json from zip to ~/.openclaw/openclaw.json",
)
parser.add_argument(
"--i-know-config-overwrites-secrets",
action="store_true",
help="Required with --apply-config",
)
parser.add_argument(
"--dry-run",
action="store_true",
help="Print planned writes only",
)
args = parser.parse_args()
if args.apply_sessions and not args.i_know_restoring_sessions_overwrites:
raise SystemExit("Refusing --apply-sessions without --i-know-restoring-sessions-overwrites")
if args.apply_config and not args.i_know_config_overwrites_secrets:
raise SystemExit("Refusing --apply-config without --i-know-config-overwrites-secrets")
home = (args.openclaw_home or openclaw_home()).expanduser().resolve()
cfg_path = args.config.expanduser().resolve() if args.config else None
if args.workspace is not None:
ws = args.workspace.expanduser().resolve()
ws.mkdir(parents=True, exist_ok=True)
else:
ws = resolve_workspace(workspace=None, openclaw_home_dir=home, config_path=cfg_path)
zip_path = args.zip.expanduser().resolve()
if not zip_path.is_file():
raise SystemExit(f"zip not found: {zip_path}")
apply_workspace = not args.no_apply_workspace
with ZipFile(zip_path, "r") as zf:
ops, skips = _gather_plan(
zf,
workspace=ws,
home=home,
apply_workspace=apply_workspace,
apply_managed=args.apply_managed_skills,
apply_sessions=args.apply_sessions,
apply_config=args.apply_config,
)
for s in skips:
print(f"warning: {s}", file=sys.stderr)
if not ops:
raise SystemExit("Nothing to apply (check flags and zip contents).")
# Config: backup existing
config_ops = [(a, l, d) for a, l, d in ops if l == "config"]
other_ops = [(a, l, d) for a, l, d in ops if l != "config"]
print(f"openclaw_home: {home}")
print(f"workspace: {ws}")
print(f"operations: {len(ops)}")
if args.dry_run:
for arc, layer, dest in sorted(ops, key=lambda x: x[2].as_posix()):
print(f" [{layer}] {arc} -> {dest}")
return
name_map = {zi.filename.replace("\\", "/"): zi for zi in zf.infolist()}
stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
# Non-config first so files land before openclaw.json is replaced.
for arc, layer, dest in sorted(other_ops, key=lambda x: x[2].as_posix()):
zi = name_map.get(arc)
if zi is None:
continue
_write_member(zf, zi, dest, dry_run=False)
print(f"wrote: {dest}")
for arc, layer, dest in config_ops:
cfg_dest = dest
if cfg_dest.is_file():
bak = cfg_dest.with_name(f"openclaw.json.bak.{stamp}")
shutil.copy2(cfg_dest, bak)
print(f"backup: {bak}", file=sys.stderr)
zi = name_map.get(arc)
if zi is None:
continue
_write_member(zf, zi, cfg_dest, dry_run=False)
print(f"wrote: {cfg_dest}")
print("done.")
if __name__ == "__main__":
main()
FILE:scripts/openclaw_paths.py
"""Shared path resolution for pack/apply scripts."""
from __future__ import annotations
import json
import os
import sys
from pathlib import Path
def openclaw_home() -> Path:
env = os.environ.get("OPENCLAW_HOME", "")
return Path(env).expanduser() if env else Path.home() / ".openclaw"
def load_openclaw_config(path: Path) -> dict | None:
if not path.is_file():
return None
raw = path.read_text(encoding="utf-8", errors="replace")
try:
return json.loads(raw)
except json.JSONDecodeError:
try:
import json5 # type: ignore
return json5.loads(raw)
except ImportError:
print(
"error: openclaw.json is not strict JSON (likely JSON5). "
"Install deps: pip install -r requirements.txt",
file=sys.stderr,
)
return None
except Exception as e: # noqa: BLE001
print(f"error: failed to parse {path}: {e}", file=sys.stderr)
return None
def workspace_from_config(data: dict) -> str | None:
agent = data.get("agent")
if isinstance(agent, dict):
w = agent.get("workspace")
if isinstance(w, str) and w.strip():
return w
agents = data.get("agents")
if isinstance(agents, dict):
defaults = agents.get("defaults")
if isinstance(defaults, dict):
w = defaults.get("workspace")
if isinstance(w, str) and w.strip():
return w
return None
def resolve_workspace(
*,
workspace: Path | None,
openclaw_home_dir: Path,
config_path: Path | None,
) -> Path:
if workspace is not None:
p = workspace.expanduser().resolve()
if not p.is_dir():
raise SystemExit(f"workspace is not a directory: {p}")
return p
cfg_file = config_path if config_path is not None else openclaw_home_dir / "openclaw.json"
data = load_openclaw_config(cfg_file)
if data:
ws = workspace_from_config(data)
if ws:
p = Path(ws).expanduser().resolve()
if p.is_dir():
return p
profile = os.environ.get("OPENCLAW_PROFILE", "default")
if profile and profile != "default":
candidate = (openclaw_home_dir / f"workspace-{profile}").resolve()
if candidate.is_dir():
return candidate
default_ws = (openclaw_home_dir / "workspace").resolve()
if default_ws.is_dir():
return default_ws
raise SystemExit(
"Could not resolve workspace. Pass --workspace PATH or fix openclaw.json / install json5 for JSON5 configs."
)
FILE:scripts/pack_openclaw.py
#!/usr/bin/env python3
"""
Pack OpenClaw user-owned data into a zip with EXPORT_MANIFEST.txt inside the archive.
Never includes ~/.openclaw/credentials/ (not implemented — do not add).
"""
from __future__ import annotations
import argparse
import hashlib
import os
import sys
import zipfile
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
from typing import Iterable
from openclaw_paths import openclaw_home, resolve_workspace
def iter_files(root: Path, *, exclude_git: bool) -> Iterable[Path]:
root = root.resolve()
always_skip = {"__pycache__", ".venv", "node_modules"}
for dirpath, dirnames, filenames in os.walk(root):
remove = [d for d in dirnames if d in always_skip or (d == ".git" and exclude_git)]
for d in remove:
dirnames.remove(d)
for name in filenames:
yield Path(dirpath) / name
@dataclass
class ManifestEntry:
arcname: str
size: int
sha256: str | None = None
@dataclass
class PackPlan:
entries: list[tuple[Path, str]] = field(default_factory=list) # (abs_path, arcname)
warnings: list[str] = field(default_factory=list)
def total_bytes(self) -> int:
return sum(p.stat().st_size for p, _ in self.entries if p.is_file())
def plan_workspace(root: Path, *, exclude_git: bool) -> PackPlan:
plan = PackPlan()
root = root.resolve()
if not root.is_dir():
plan.warnings.append(f"missing workspace: {root}")
return plan
prefix = "workspace/"
for f in iter_files(root, exclude_git=exclude_git):
if not f.is_file():
continue
try:
rel = f.relative_to(root)
except ValueError:
continue
arc = prefix + rel.as_posix()
plan.entries.append((f, arc))
return plan
def plan_managed_skills(home: Path) -> PackPlan:
plan = PackPlan()
skills = (home / "skills").resolve()
if not skills.is_dir():
plan.warnings.append(f"managed skills directory missing: {skills}")
return plan
prefix = "managed-skills/"
for f in iter_files(skills, exclude_git=True):
if not f.is_file():
continue
rel = f.relative_to(skills)
plan.entries.append((f, prefix + rel.as_posix()))
return plan
def plan_sessions(home: Path) -> PackPlan:
plan = PackPlan()
agents = home / "agents"
if not agents.is_dir():
plan.warnings.append(f"agents directory missing: {agents}")
return plan
for agent_dir in sorted(agents.iterdir()):
if not agent_dir.is_dir():
continue
sess = agent_dir / "sessions"
if not sess.is_dir():
continue
agent_id = agent_dir.name
base = f"sessions/{agent_id}/sessions/"
for f in iter_files(sess, exclude_git=True):
if not f.is_file():
continue
rel = f.relative_to(sess)
plan.entries.append((f, base + rel.as_posix()))
if not any(e[1].startswith("sessions/") for e in plan.entries):
plan.warnings.append("no session directories found under ~/.openclaw/agents/*/sessions/")
return plan
def plan_config(home: Path) -> PackPlan:
plan = PackPlan()
cfg = home / "openclaw.json"
if not cfg.is_file():
plan.warnings.append(f"config missing: {cfg}")
return plan
plan.entries.append((cfg.resolve(), "config/openclaw.json"))
return plan
def file_sha256(path: Path, chunk: int = 1024 * 1024) -> str:
h = hashlib.sha256()
with path.open("rb") as fp:
while True:
b = fp.read(chunk)
if not b:
break
h.update(b)
return h.hexdigest()
def build_manifest(
entries: list[tuple[Path, str]],
*,
with_sha256: bool,
sha256_max_mb: float,
) -> list[ManifestEntry]:
out: list[ManifestEntry] = []
max_b = int(sha256_max_mb * 1024 * 1024)
for abs_path, arcname in sorted(entries, key=lambda x: x[1]):
if not abs_path.is_file():
continue
size = abs_path.stat().st_size
digest = None
if with_sha256 and size <= max_b:
digest = file_sha256(abs_path)
out.append(ManifestEntry(arcname=arcname, size=size, sha256=digest))
return out
def write_zip(
output: Path,
entries: list[tuple[Path, str]],
*,
with_sha256: bool,
sha256_max_mb: float,
) -> str:
manifest_lines = [
"# OpenClaw user data export",
f"# created_utc: {datetime.now(timezone.utc).isoformat()}",
"# format: path<TAB>size_bytes<TAB>sha256_or_empty",
"",
]
m_entries = build_manifest(entries, with_sha256=with_sha256, sha256_max_mb=sha256_max_mb)
for m in m_entries:
sh = m.sha256 or ""
manifest_lines.append(f"{m.arcname}\t{m.size}\t{sh}")
manifest_body = "\n".join(manifest_lines) + "\n"
output.parent.mkdir(parents=True, exist_ok=True)
with zipfile.ZipFile(output, "w", compression=zipfile.ZIP_DEFLATED) as zf:
zf.writestr("EXPORT_MANIFEST.txt", manifest_body.encode("utf-8"))
seen: set[str] = {"EXPORT_MANIFEST.txt"}
for abs_path, arcname in entries:
if not abs_path.is_file():
continue
if arcname in seen:
continue
seen.add(arcname)
zf.write(abs_path, arcname)
return manifest_body
def main() -> None:
parser = argparse.ArgumentParser(description="Pack OpenClaw user data into a zip (workspace + optional layers).")
parser.add_argument(
"--workspace",
type=Path,
default=None,
help="Agent workspace root (default: resolve from openclaw.json or ~/.openclaw/workspace[-profile])",
)
parser.add_argument(
"--openclaw-home",
type=Path,
default=None,
help="OpenClaw data directory (default: $OPENCLAW_HOME or ~/.openclaw)",
)
parser.add_argument(
"--config",
type=Path,
default=None,
help="Path to openclaw.json for workspace resolution only (default: <openclaw-home>/openclaw.json)",
)
parser.add_argument(
"--output",
"-o",
type=Path,
default=None,
help="Output zip path (default: ./openclaw-user-export-YYYYMMDD-HHMMSS.zip in cwd)",
)
parser.add_argument(
"--exclude-git",
action=argparse.BooleanOptionalAction,
default=True,
help="Skip .git directories under workspace (default: true)",
)
parser.add_argument(
"--managed-skills",
action="store_true",
help="Include ~/.openclaw/skills as managed-skills/",
)
parser.add_argument(
"--sessions",
action="store_true",
help="Include ~/.openclaw/agents/*/sessions/ (large, full chat transcripts)",
)
parser.add_argument(
"--i-know-sessions-are-large-and-sensitive",
action="store_true",
help="Required with --sessions",
)
parser.add_argument(
"--config-snapshot",
action="store_true",
help="Include openclaw.json as config/openclaw.json (may contain secrets)",
)
parser.add_argument(
"--i-know-config-may-contain-secrets",
action="store_true",
help="Required with --config-snapshot",
)
parser.add_argument(
"--dry-run",
action="store_true",
help="Print planned files and totals; do not write zip",
)
parser.add_argument(
"--manifest-sha256",
action="store_true",
help="Add SHA256 per file in EXPORT_MANIFEST.txt (skipped for files larger than --sha256-max-mb)",
)
parser.add_argument(
"--sha256-max-mb",
type=float,
default=32.0,
help="Max file size (MB) to checksum when --manifest-sha256 (default: 32)",
)
args = parser.parse_args()
home = (args.openclaw_home or openclaw_home()).expanduser().resolve()
if args.sessions and not args.i_know_sessions_are_large_and_sensitive:
raise SystemExit("Refusing --sessions without --i-know-sessions-are-large-and-sensitive")
if args.config_snapshot and not args.i_know_config_may_contain_secrets:
raise SystemExit("Refusing --config-snapshot without --i-know-config-may-contain-secrets")
cfg_path = args.config.expanduser().resolve() if args.config else None
ws = resolve_workspace(workspace=args.workspace, openclaw_home_dir=home, config_path=cfg_path)
plans: list[PackPlan] = [plan_workspace(ws, exclude_git=args.exclude_git)]
if args.managed_skills:
plans.append(plan_managed_skills(home))
if args.sessions:
plans.append(plan_sessions(home))
if args.config_snapshot:
plans.append(plan_config(home))
merged: list[tuple[Path, str]] = []
for p in plans:
merged.extend(p.entries)
if not merged:
raise SystemExit("Nothing to pack.")
for p in plans:
for w in p.warnings:
print(f"warning: {w}", file=sys.stderr)
total_files = len(merged)
total_bytes = sum(Path(a).stat().st_size for a, _ in merged if Path(a).is_file())
print(f"workspace: {ws}")
print(f"openclaw_home: {home}")
print(f"files: {total_files} bytes: {total_bytes}")
if args.dry_run:
for abs_path, arc in sorted(merged, key=lambda x: x[1])[:500]:
print(f" {arc} <= {abs_path}")
if total_files > 500:
print(f" ... ({total_files - 500} more)")
return
out = args.output
if out is None:
stamp = datetime.now().strftime("%Y%m%d-%H%M%S")
out = Path.cwd() / f"openclaw-user-export-{stamp}.zip"
else:
out = out.expanduser().resolve()
write_zip(
out,
merged,
with_sha256=args.manifest_sha256,
sha256_max_mb=args.sha256_max_mb,
)
print(f"wrote: {out}")
if __name__ == "__main__":
main()
Count tokens and estimate costs for 300+ LLM models. Primary use: audit OpenClaw workspace token consumption (memory, persona, skills).
---
name: prompt-token-counter
version: 1.0.10
description: "Count tokens and estimate costs for 300+ LLM models. Primary use: audit OpenClaw workspace token consumption (memory, persona, skills)."
trigger: "token count, cost estimate, prompt length, API cost, OpenClaw audit, workspace token usage, memory/persona/skills tokens, context window limit"
---
# Prompt Token Counter (toksum)
> **First load reminder:** This skill provides the `scripts` CLI (toksum). Use it when the user asks to count tokens, estimate API costs, or **audit OpenClaw component token consumption** (memory, persona, skills).
## Before Installing — Security & Privacy
- **What will be read:** The audit workflow reads files under `~/.openclaw/workspace` and `~/.openclaw/skills` (AGENTS.md, SOUL.md, MEMORY.md, SKILL.md, etc.). Those files may contain personal data or secrets. Only install if you accept that access.
- **URL fetching:** The CLI can fetch HTTP(S) URLs via `-u`. SKILL.md requires the agent to confirm each URL with the user before fetching. Insist the agent follow that rule; never allow automatic fetching of unknown URLs.
- **Source verification:** Source: [https://github.com/Zhaobudaoyuema/prompt-token-counter](https://github.com/Zhaobudaoyuema/prompt-token-counter). Review `scripts/core.py` and `scripts/cli.py` before use. The code performs local file reads and optional HTTP GETs only; no other network calls or data exfiltration.
- **Run locally first:** If unsure, run the CLI manually in an isolated environment against safe test files to verify behavior.
## Primary Use: OpenClaw Token Consumption Audit
**Goal:** Help users identify which OpenClaw components consume tokens and how much.
### 1. Memory & Persona Files
These files are injected into sessions and consume tokens. Search and count them:
| File | Purpose | Typical Location |
|------|---------|------------------|
| `AGENTS.md` | Operating instructions, workflow, priorities | `~/.openclaw/workspace/` |
| `SOUL.md` | Persona, tone, values, behavioral guidelines | `~/.openclaw/workspace/` |
| `IDENTITY.md` | Name, role, goals, visual description | `~/.openclaw/workspace/` |
| `USER.md` | User preferences, communication style | `~/.openclaw/workspace/` |
| `MEMORY.md` | Long-term memory, persistent facts | `~/.openclaw/workspace/` |
| `TOOLS.md` | Tool quirks, path conventions | `~/.openclaw/workspace/` |
| `HEARTBEAT.md` | Periodic maintenance checklist | `~/.openclaw/workspace/` |
| `BOOT.md` | Startup ritual (when hooks enabled) | `~/.openclaw/workspace/` |
| `memory/YYYY-MM-DD.md` | Daily memory logs | `~/.openclaw/workspace/memory/` |
**Workspace path:** Default `~/.openclaw/workspace`; may be overridden in `~/.openclaw/openclaw.json` via `agent.workspace`.
### 2. Skill Files (SKILL.md)
Skills are loaded per session. Count each `SKILL.md`:
| Location | Scope |
|----------|-------|
| `~/.openclaw/skills/*/SKILL.md` | OpenClaw managed skills |
| `~/.openclaw/workspace/skills/*/SKILL.md` | Workspace-specific skills (override) |
### 3. Audit Workflow
1. **Locate workspace:** Resolve `~/.openclaw/workspace` (or config override).
2. **Collect files:** List all memory/persona files and `SKILL.md` paths above.
3. **Count tokens:** Run `python -m scripts.cli <path1> <path2> ... -m <model> -c` (batch mode).
4. **Summarize:** Group by category (memory, persona, skills), report total and per-file.
**Example audit command (PowerShell):**
```powershell
$ws = "$env:USERPROFILE\.openclaw\workspace"
python -m scripts.cli -m gpt-4o -c "$ws\AGENTS.md" "$ws\SOUL.md" "$ws\USER.md" "$ws\IDENTITY.md" "$ws\MEMORY.md" "$ws\TOOLS.md"
```
**Example audit (Bash):**
```bash
WS=~/.openclaw/workspace
python -m scripts.cli -m gpt-4o -c "$WS/AGENTS.md" "$WS/SOUL.md" "$WS/USER.md" "$WS/IDENTITY.md" "$WS/MEMORY.md" "$WS/TOOLS.md"
```
---
## Project Layout
```
prompt_token_counter/
├── SKILL.md
├── package.json # npm package (OpenClaw skill)
├── publish_npm.py # Publish to npm; syncs version
└── scripts/ # Python package, CLI + examples
├── cli.py # Entry point
├── core.py # TokenCounter, estimate_cost
├── registry/
│ ├── models.py # 300+ models
│ └── pricing.py # Pricing data
└── examples/ # Script examples
├── count_prompt.py
├── estimate_cost.py
├── batch_compare.py
└── benchmark_token_ratio.py
```
Invoke: `python -m scripts.cli` from project root.
### Version Sync (publish_npm.py)
When publishing to npm, `publish_npm.py` bumps the patch version and syncs it to:
- `package.json` — `version`
- `SKILL.md` — frontmatter `version`
- `scripts/__init__.py` — `__version__`
Run: `python publish_npm.py` (after `npm login`).
---
## Runtime Dependencies
- **Python 3** — required
- **tiktoken** (optional) — `pip install tiktoken` for exact OpenAI counts
---
## Language Rule
**Respond in the user's language.** Match the user's language (e.g. Chinese if they write in Chinese, English if they write in English).
---
## URL Usage — Mandatory Agent Rule
**Before using `-u` / `--url` to fetch content from any URL, you MUST:**
1. **Explicitly warn the user** that the CLI will make an outbound HTTP/HTTPS request to the given URL.
2. **Confirm the URL is trusted** — tell the user: "Only use URLs you fully trust. Untrusted URLs may expose your IP, leak data, or be used for SSRF. Do you confirm this URL is safe?"
3. **Prefer alternatives** — if the user can provide the content via `-f` (local file) or inline text, suggest that instead of URL fetch.
4. **Never auto-fetch** — do not invoke `-u` without the user having explicitly provided the URL and acknowledged the risk.
**If the user insists on using a URL:** Proceed only after they confirm. State clearly: "I will fetch from [URL] to count tokens. Proceed?"
---
## Model Name — Mandatory Agent Rule
**Before invoking the CLI, you MUST have a concrete model name from the user.**
1. **Require explicit model** — `-m` / `--model` is required. Do not guess or assume; the user must provide the exact name (e.g. gpt-4o, claude-3-5-sonnet-20241022).
2. **If unclear, ask** — if the user says "GPT" or "Claude" or "the latest model" without a specific name, ask: "Please specify the exact model name (e.g. gpt-4o, claude-3-5-sonnet-20241022). Run `python -m scripts.cli -l` to list supported models."
3. **Do not auto-pick** — never substitute a model on behalf of the user without their confirmation.
4. **Validate when possible** — if the model name seems ambiguous, offer `-l` output or confirm: "I'll use [model]. Is that correct?"
---
## CLI Usage
**Default:** Read from local file(s). No segmentation. Supports multiple file paths for batch execution.
```bash
python -m scripts.cli [OPTIONS] [FILE ...]
```
| Option | Short | Description |
|--------|-------|-------------|
| `--model` | `-m` | Model name (required unless `--list-models`) — **Agent must obtain exact name from user; ask if unclear** |
| `--file` | `-f` | Read from file (repeatable) |
| `--url` | `-u` | Read from URL (repeatable) — **Agent must warn user before use; only trusted URLs** |
| `--list-models` | `-l` | List supported models |
| `--cost` | `-c` | Show cost estimate |
| `--output-tokens` | | Use output token pricing |
| `--currency` | | USD or INR |
| `--verbose` | `-v` | Detailed output |
### Examples
```bash
# Multiple local files (default batch mode)
python -m scripts.cli file1.txt file2.txt -m gpt-4
python -m scripts.cli AGENTS.md SOUL.md MEMORY.md -m gpt-4o -c
# Single file with -f
python -m scripts.cli -f input.txt -m claude-3-opus -c
# Inline text (when arg is not an existing file path)
python -m scripts.cli -m gpt-4 "Hello, world!"
# List models
python -m scripts.cli -l
# Run bundled example scripts
python scripts/examples/count_prompt.py file1.txt file2.txt -m gpt-4
python scripts/examples/estimate_cost.py "Your text" gpt-4
python scripts/examples/batch_compare.py file1.txt -m gpt-4 claude-3-opus
```
---
## Python API
```python
from scripts import TokenCounter, count_tokens, estimate_cost, get_supported_models
tokens = count_tokens("Hello!", "gpt-4")
counter = TokenCounter("claude-3-opus")
tokens = counter.count_messages([
{"role": "system", "content": "..."},
{"role": "user", "content": "..."}
])
cost = estimate_cost(tokens, "gpt-4", input_tokens=True)
```
---
## Supported Models
300+ models across 34+ providers: OpenAI, Anthropic, Google, Meta, Mistral, Cohere, xAI, DeepSeek, etc. Use `python -m scripts.cli -l` for full list.
- **OpenAI:** exact via tiktoken
- **Others:** ~85–95% approximation
---
## Response Output — Agent Guideline
**After returning token count or cost estimate results, the agent MUST:**
1. **Include the project link** — e.g.
> Source: [prompt-token-counter](https://github.com/Zhaobudaoyuema/prompt-token-counter)
2. **Briefly explain how tokens are calculated** — e.g.
> **How tokens are counted:** OpenAI models use tiktoken (exact). Other models use provider-specific formulas calibrated from benchmark data. For CJK-heavy text, the ratio is blended by CJK character ratio so that Chinese gets fewer chars per token.
---
## Common Issues
| Issue | Action |
|-------|--------|
| "tiktoken is required" | `pip install tiktoken` |
| UnsupportedModelError | Use `-l` for valid names |
| Cost "NA" | Model has no pricing; count still valid |
| User provides URL | **Agent must warn:** outbound request, SSRF risk, only trusted URLs; confirm before `-u` |
| Model unclear / vague | **Agent must ask:** user to specify exact model name; offer `-l` to list; do not guess |
---
## When to Trigger This Skill
Activate this skill when the user:
| Trigger | Example phrases |
|---------|-----------------|
| **Token count** | "How many tokens?", "Count tokens in this prompt", "Token length of X" |
| **Cost estimate** | "Estimate API cost", "How much for this text?", "Cost for GPT-4" |
| **Prompt size** | "Check prompt length", "Is this too long?", "Context window limit" |
| **OpenClaw audit** | "How many tokens does my workspace use?", "Audit OpenClaw memory/persona/skills", "Which components consume tokens?", "Token usage of AGENTS.md / SOUL.md / skills" |
| **Model comparison** | "Compare token cost across models", "Which model is cheaper?" |
Also trigger when the agent needs to count tokens or estimate cost before/after generating content.
---
## Quick Reference
| Item | Command |
|------|---------|
| Invoke | `python -m scripts.cli` |
| List models | `python -m scripts.cli -l` |
| Cost | `-c` (input) / `--output-tokens` (output) |
| Currency | `--currency USD` or `INR` |
FILE:package.json
{
"name": "prompt-token-counter",
"version": "1.0.10",
"description": "OpenClaw skill: Count tokens and estimate costs for 300+ LLM models. Audit workspace token consumption (memory, persona, skills).",
"keywords": [
"openclaw",
"skill",
"token",
"toksum",
"llm"
],
"license": "MIT",
"repository": {
"type": "git",
"url": "https://github.com/Zhaobudaoyuema/prompt-token-counter"
},
"homepage": "https://github.com/Zhaobudaoyuema/prompt-token-counter",
"bugs": {
"url": "https://github.com/Zhaobudaoyuema/prompt-token-counter/issues"
},
"files": [
"SKILL.md",
"publish_npm.py",
"scripts"
]
}
FILE:README.md
# prompt-token-counter
[中文](README.zh.md)
An **OpenClaw skill** for counting tokens and estimating API costs.
## How It Works
This project provides **local token counting** without API calls:
| Approach | OpenAI | Other models |
|----------|--------|--------------|
| **Method** | tiktoken (exact) | Formula-based approximation |
| **Accuracy** | Exact | ~85–95% typical |
For non-OpenAI models, we use **provider-specific formulas** calibrated from benchmark data. When text contains CJK (Chinese/Japanese/Korean), we **blend the ratio** by CJK character ratio so that Chinese-heavy text gets fewer chars-per-token (more tokens per character).
### Benchmark: Main Models (8927 chars, English/mixed)
| Model | Chars | Tokens | 1 token ≈ chars | Status |
|-------|-------|--------|-----------------|--------|
| anthropic/claude-sonnet-4-6 | 8927 | 2763 | 3.23 | ✓ |
| anthropic/claude-sonnet-4-5 | 8927 | 2763 | 3.23 | ✓ |
| anthropic/claude-opus-4.6 | 8927 | 2763 | 3.23 | ✓ |
| openai/gpt-5.2-codex | 8927 | 2459 | 3.63 | ✓ |
| google/gemini-3.1-pro-preview | 8927 | 2627 | 3.40 | ✓ |
| z-ai/glm-5 | 8927 | 2457 | 3.63 | ✓ |
| volcengine/doubao-seed-2-0-pro | 8927 | 2702 | 3.30 | ✓ |
| moonshot/kimi-k2.5 | 8927 | 2402 | 3.72 | ✓ |
| minimax/MiniMax-M2.5 | 8927 | 2428 | 3.68 | ✓ |
| deepseek-v3.2 | 8927 | 2578 | 3.46 | ✓ |
### Benchmark: Main Models (3050 chars, Chinese-mixed 混杂中文)
| Model | Chars | Tokens | 1 token ≈ chars | Status |
|-------|-------|--------|-----------------|--------|
| anthropic/claude-sonnet-4-6 | 3050 | 1913 | 1.59 | ✓ |
| anthropic/claude-sonnet-4-5 | 3050 | 1913 | 1.59 | ✓ |
| anthropic/claude-opus-4.6 | 3050 | 1913 | 1.59 | ✓ |
| openai/gpt-5.2-codex | 3050 | 1564 | 1.95 | ✓ |
| google/gemini-3.1-pro-preview | 3050 | 1473 | 2.07 | ✓ |
| z-ai/glm-5 | 3050 | 1318 | 2.31 | ✓ |
| volcengine/doubao-seed-2-0-pro | 3050 | 1494 | 2.04 | ✓ |
| moonshot/kimi-k2.5 | 3050 | 1257 | 2.43 | ✓ |
| minimax/MiniMax-M2.5 | 3050 | 1289 | 2.37 | ✓ |
| deepseek-v3.2 | 3050 | 1361 | 2.24 | ✓ |
*Benchmark data from `scripts/examples/benchmark_token_ratio.py` (API mode).*
## What This Skill Does
When loaded, the agent can:
| Capability | Use case |
|------------|----------|
| **Count tokens** | "How many tokens in this prompt?", "Token length of X" |
| **Estimate cost** | "How much for this text on GPT-4?", "API cost for Claude" |
| **Audit OpenClaw workspace** | "How many tokens does my workspace use?", "Which memory/persona/skills consume tokens?" |
| **Compare models** | "Compare token cost across models", "Which model is cheaper?" |
### OpenClaw Token Audit
The skill helps identify token consumption of workspace components:
- **Memory & persona**: AGENTS.md, SOUL.md, IDENTITY.md, USER.md, MEMORY.md, TOOLS.md, etc.
- **Skills**: Each SKILL.md under `~/.openclaw/skills/` or `workspace/skills/`
Example audit (batch mode, multiple files):
```bash
python -m scripts.cli -m gpt-4o -c AGENTS.md SOUL.md MEMORY.md
```
## When to Trigger
- User asks about token count, prompt length, API cost
- User mentions OpenClaw context size or workspace token usage
- Agent needs to audit token consumption before/after changes
## Copy and send to OpenClaw
**ClawHub (recommended, international)**
```text
Please run clawhub install prompt-token-counter to install this skill, and help me count tokens and estimate API costs.
```
**npm**
```text
Please run npm i prompt-token-counter to install this skill, and help me count tokens and estimate API costs.
```
## Quick Reference
```bash
python -m scripts.cli -m gpt-4 "Hello, world!"
python -m scripts.cli -f input.txt -m claude-3-opus -c
python -m scripts.cli -l # list 300+ models
```
## Benchmark Script
Run `scripts/examples/benchmark_token_ratio.py` to test token ratios across models:
- **API mode** (default): Uses model API to get exact `prompt_tokens`. Set `API_KEY` and `BASE_URL` in the script.
- **Local mode** (`--local`): Uses this project's TokenCounter (no API). Good for quick comparison.
```bash
python scripts/examples/benchmark_token_ratio.py # API mode
python scripts/examples/benchmark_token_ratio.py --local # local approximation
```
MIT
FILE:README.zh.md
# prompt-token-counter
[English](README.md)
一个 **OpenClaw skill**，用于统计 token 数量并估算 API 成本。
## 实现原理
本项目提供**本地 token 计数**，无需调用 API：
| 方式 | OpenAI | 其他模型 |
|------|--------|----------|
| **方法** | tiktoken（精确） | 公式近似 |
| **精度** | 精确 | 约 85–95% |
对非 OpenAI 模型，使用**按 provider 校准的公式**。当文本包含 CJK（中/日/韩）时，会根据 CJK 字符占比**混合比例**，使中文为主的文本获得更低的 chars/token（每字符更多 token）。
### 基准测试：主力模型（8927 字符，英文/混合）
| 模型 | 字符数 | Token 数 | 1 token ≈ 多少字符 | 状态 |
|------|--------|----------|-------------------|------|
| anthropic/claude-sonnet-4-6 | 8927 | 2763 | 3.23 | ✓ |
| anthropic/claude-sonnet-4-5 | 8927 | 2763 | 3.23 | ✓ |
| anthropic/claude-opus-4.6 | 8927 | 2763 | 3.23 | ✓ |
| openai/gpt-5.2-codex | 8927 | 2459 | 3.63 | ✓ |
| google/gemini-3.1-pro-preview | 8927 | 2627 | 3.40 | ✓ |
| z-ai/glm-5 | 8927 | 2457 | 3.63 | ✓ |
| volcengine/doubao-seed-2-0-pro | 8927 | 2702 | 3.30 | ✓ |
| moonshot/kimi-k2.5 | 8927 | 2402 | 3.72 | ✓ |
| minimax/MiniMax-M2.5 | 8927 | 2428 | 3.68 | ✓ |
| deepseek-v3.2 | 8927 | 2578 | 3.46 | ✓ |
### 基准测试：主力模型（3050 字符，混杂中文）
| 模型 | 字符数 | Token 数 | 1 token ≈ 多少字符 | 状态 |
|------|--------|----------|-------------------|------|
| anthropic/claude-sonnet-4-6 | 3050 | 1913 | 1.59 | ✓ |
| anthropic/claude-sonnet-4-5 | 3050 | 1913 | 1.59 | ✓ |
| anthropic/claude-opus-4.6 | 3050 | 1913 | 1.59 | ✓ |
| openai/gpt-5.2-codex | 3050 | 1564 | 1.95 | ✓ |
| google/gemini-3.1-pro-preview | 3050 | 1473 | 2.07 | ✓ |
| z-ai/glm-5 | 3050 | 1318 | 2.31 | ✓ |
| volcengine/doubao-seed-2-0-pro | 3050 | 1494 | 2.04 | ✓ |
| moonshot/kimi-k2.5 | 3050 | 1257 | 2.43 | ✓ |
| minimax/MiniMax-M2.5 | 3050 | 1289 | 2.37 | ✓ |
| deepseek-v3.2 | 3050 | 1361 | 2.24 | ✓ |
*基准数据来自 `scripts/examples/benchmark_token_ratio.py`（API 模式）。*
## 这个 Skill 能做什么
加载后，Agent 可以：
| 能力 | 使用场景 |
|------|----------|
| **统计 token** | 「这段 prompt 有多少 token？」、「X 的 token 长度」 |
| **估算成本** | 「这段文字用 GPT-4 要多少钱？」、「Claude 的 API 成本」 |
| **审计 OpenClaw 工作区** | 「我的工作区用了多少 token？」、「哪些 memory/persona/skills 消耗 token？」 |
| **对比模型** | 「对比不同模型的 token 成本」、「哪个模型更便宜？」 |
### OpenClaw Token 审计
该 skill 帮助识别工作区各组件的 token 消耗：
- **Memory 与 persona**：AGENTS.md、SOUL.md、IDENTITY.md、USER.md、MEMORY.md、TOOLS.md 等
- **Skills**：`~/.openclaw/skills/` 或 `workspace/skills/` 下的每个 SKILL.md
审计示例（批量多文件）：
```bash
python -m scripts.cli -m gpt-4o -c AGENTS.md SOUL.md MEMORY.md
```
## 何时触发
- 用户询问 token 数量、prompt 长度、API 成本
- 用户提到 OpenClaw 上下文大小或工作区 token 使用
- Agent 需要在变更前后审计 token 消耗
## 复制发送给 OpenClaw
**ClawHub（推荐，国际）**
```text
Please run clawhub install prompt-token-counter to install this skill, and help me count tokens and estimate API costs.
```
**npm**
```text
Please run npm i prompt-token-counter to install this skill, and help me count tokens and estimate API costs.
```
## 快速参考
```bash
python -m scripts.cli -m gpt-4 "Hello, world!"
python -m scripts.cli -f input.txt -m claude-3-opus -c
python -m scripts.cli -l # 列出 300+ 模型
```
## 基准测试脚本
运行 `scripts/examples/benchmark_token_ratio.py` 可批量测试各模型的 token 比例：
- **API 模式**（默认）：调用模型 API 获取精确 `prompt_tokens`。需在脚本中设置 `API_KEY` 和 `BASE_URL`。
- **本地模式**（`--local`）：使用本项目的 TokenCounter 近似计算，无需 API。
```bash
python scripts/examples/benchmark_token_ratio.py # API 模式
python scripts/examples/benchmark_token_ratio.py --local # 本地近似
```
MIT
FILE:scripts/cli.py
"""
Command-line interface for scripts.
This module provides a comprehensive command-line interface for the scripts library,
allowing users to count tokens and estimate costs for various LLM models directly
from the terminal.
The CLI supports:
- Token counting for text input or files
- Cost estimation with detailed breakdowns
- Listing all supported models by provider
- Verbose output with detailed information
- Support for both input and output token pricing
Examples:
Basic token counting:
.. code-block:: bash
scripts "Hello, world!" gpt-4
scripts --file input.txt claude-3-opus-20240229
Cost estimation:
.. code-block:: bash
scripts --cost "Your text here" gpt-4
scripts --cost --output-tokens "Response text" gpt-4
List supported models:
.. code-block:: bash
scripts --list-models
Verbose output:
.. code-block:: bash
scripts --verbose --cost --file large_document.txt gpt-4
Functions:
main: Main CLI entry point that handles argument parsing and execution
list_models: Display all supported models organized by provider
The CLI provides comprehensive error handling and user-friendly output formatting
for both simple token counting and detailed cost analysis workflows.
"""
import argparse
import sys
import urllib.parse
import urllib.request
from pathlib import Path
from typing import List, Optional, Tuple
from .core import TokenCounter, get_supported_models, estimate_cost
from .exceptions import UnsupportedModelError, TokenizationError
def _read_file(path: str) -> str:
with open(path, "r", encoding="utf-8") as file_handle:
return file_handle.read()
def _read_url(url: str) -> str:
parsed = urllib.parse.urlparse(url)
if parsed.scheme not in {"http", "https"}:
raise ValueError("URL must start with http:// or https://")
with urllib.request.urlopen(url, timeout=10) as response:
return response.read().decode("utf-8", errors="replace")
def _collect_inputs(argv: List[str]) -> List[Tuple[str, str, Optional[str]]]:
"""Collect inputs. Returns list of (source, value, label) where label is path for display."""
items: List[Tuple[str, str, Optional[str]]] = []
i = 0
while i < len(argv):
arg = argv[i]
if arg in {"--file", "-f"}:
if i + 1 >= len(argv):
raise ValueError("--file requires a path")
path = argv[i + 1]
items.append(("file", path, path))
i += 2
continue
if arg in {"--url", "-u"}:
if i + 1 >= len(argv):
raise ValueError("--url requires a URL")
items.append(("url", argv[i + 1], argv[i + 1]))
i += 2
continue
if arg in {"--model", "-m", "--currency"}:
i += 2
continue
if arg in {"--cost", "-c", "--output-tokens", "--verbose", "-v", "--list-models", "-l"}:
i += 1
continue
if arg.startswith("-"):
i += 1
continue
# Positional: treat as file path if it exists, else inline text
p = Path(arg)
if p.is_file():
items.append(("file", arg, arg))
else:
items.append(("text", arg, None))
i += 1
return items
def _format_cost(cost: float, currency: str) -> str:
if cost <= 0:
return "NA"
symbol = "₹" if currency.upper() == "INR" else "$"
precision = 2 if currency.upper() == "INR" else 6
return f"{symbol}{cost:.{precision}f}"
def main() -> None:
"""
Main CLI entry point.
Parses command-line arguments and executes the appropriate scripts functionality.
Supports token counting, cost estimation, model listing, and file input processing.
The function handles:
- Argument parsing and validation
- Text input from command line or file
- Token counting for specified models
- Cost estimation with input/output token differentiation
- Model listing with provider organization
- Comprehensive error handling and user feedback
- Verbose output formatting
Command-line Arguments:
text (str, optional): Text to count tokens for
model (str, optional): Model name (required unless using --list-models)
--file, -f (str): Read text from file instead of command line
--list-models, -l: List all supported models by provider
--cost, -c: Show cost estimation along with token count
--output-tokens: Calculate cost for output tokens instead of input
--verbose, -v: Show detailed output with additional information
Exit Codes:
0: Success
1: Error (unsupported model, file not found, tokenization failure, etc.)
Raises:
SystemExit: On error conditions or user interruption
Examples:
Basic usage:
.. code-block:: bash
scripts "Hello, world!" gpt-4
scripts --file document.txt claude-3-opus-20240229
With cost estimation:
.. code-block:: bash
scripts --cost --verbose "Long text content" gpt-4
scripts --cost --output-tokens "Response text" gpt-4
List models:
.. code-block:: bash
scripts --list-models
"""
parser = argparse.ArgumentParser(
description="Count tokens for various LLM models",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""
Examples:
scripts file1.txt file2.txt --model gpt-4
scripts --file input.txt --model claude-3-opus-20240229
scripts --cost --model gpt-4 AGENTS.md SOUL.md MEMORY.md
scripts --list-models
"""
)
parser.add_argument(
"paths",
nargs="*",
metavar="FILE",
help="Local file path(s) to count tokens (default input mode). Multiple paths supported for batch."
)
parser.add_argument(
"--model", "-m",
help="Model name (e.g., gpt-4, claude-3-opus-20240229)"
)
parser.add_argument(
"--file", "-f",
action="append",
help="Read text from file (can be used multiple times)"
)
parser.add_argument(
"--url", "-u",
action="append",
help="Read text from URL (http/https, can be used multiple times)"
)
parser.add_argument(
"--list-models", "-l",
action="store_true",
help="List all supported models"
)
parser.add_argument(
"--cost", "-c",
action="store_true",
help="Show cost estimation along with token count"
)
parser.add_argument(
"--currency",
default="USD",
help="Currency for cost output (USD or INR)"
)
parser.add_argument(
"--output-tokens",
action="store_true",
help="Calculate cost for output tokens instead of input tokens"
)
parser.add_argument(
"--verbose", "-v",
action="store_true",
help="Show verbose output"
)
args = None
try:
args = parser.parse_args()
if args.list_models:
list_models()
return
if not args.model:
parser.error("Model name is required unless using --list-models")
inputs = _collect_inputs(sys.argv[1:])
# Also add positional paths from argparse (in case -m was before paths)
for p in (args.paths or []):
if Path(p).is_file():
if not any(src == "file" and val == p for src, val, _ in inputs):
inputs.append(("file", p, p))
else:
if not any(src == "text" and val == p for src, val, _ in inputs):
inputs.append(("text", p, None))
if not inputs:
parser.error("Provide local file path(s), --file, or --url inputs")
items: List[Tuple[str, str, Optional[str]]] = []
for source, value, label in inputs:
if source == "text":
items.append((source, value, label))
continue
try:
if source == "file":
content = _read_file(value)
items.append((source, content, value))
elif source == "url":
content = _read_url(value)
items.append((source, content, value))
else:
continue
except FileNotFoundError:
print(f"Error: File '{value}' not found", file=sys.stderr)
sys.exit(1)
except Exception as e:
print(f"Error reading {source} '{value}': {e}", file=sys.stderr)
sys.exit(1)
try:
counter = TokenCounter(args.model)
for index, (source, content, label) in enumerate(items, start=1):
token_count = counter.count(content)
prefix = f"{label}: " if label else f"{index}) "
if args.cost:
cost = estimate_cost(
token_count,
args.model,
input_tokens=not args.output_tokens,
currency=args.currency,
)
cost_display = _format_cost(cost, args.currency)
print(f"{prefix}tokens={token_count} cost={cost_display}")
else:
print(f"{prefix}tokens={token_count}")
except UnsupportedModelError as e:
print(f"Error: {e}", file=sys.stderr)
if args.verbose:
print("\nUse --list-models to see supported models", file=sys.stderr)
sys.exit(1)
except TokenizationError as e:
print(f"Error: {e}", file=sys.stderr)
sys.exit(1)
except KeyboardInterrupt:
print("\nInterrupted by user", file=sys.stderr)
sys.exit(1)
except Exception as e:
print(f"Unexpected error: {e}", file=sys.stderr)
if args is not None and args.verbose:
import traceback
traceback.print_exc()
sys.exit(1)
def list_models() -> None:
"""
List all supported models organized by provider.
Displays a comprehensive list of all supported models grouped by their
respective providers (OpenAI, Anthropic, Google, Meta, etc.). The output
includes model counts per provider and a total count across all providers.
The function:
- Retrieves all supported models using get_supported_models()
- Groups models by provider with clear section headers
- Sorts models alphabetically within each provider
- Shows model counts for each provider and overall total
- Formats output for easy readability
Output Format:
.. code-block:: text
Supported models:
==================================================
OPENAI (25 models):
------------------------------
gpt-3.5-turbo
gpt-4
gpt-4o
...
ANTHROPIC (12 models):
------------------------------
claude-3-haiku-20240307
claude-3-opus-20240229
...
Total: 200+ models
Note:
This function is typically called when the --list-models CLI flag is used.
It provides users with a complete overview of available models for token
counting and cost estimation.
"""
models = get_supported_models()
print("Supported models:")
print("=" * 50)
for provider, model_list in models.items():
print(f"\n{provider.upper()} ({len(model_list)} models):")
print("-" * 30)
for model in sorted(model_list):
print(f" {model}")
print(f"\nTotal: {sum(len(models) for models in models.values())} models")
if __name__ == "__main__":
main()
FILE:scripts/core.py
"""
Core functionality for token counting across different LLM providers.
This module contains the main TokenCounter class and supporting functions that provide
token counting capabilities for 200+ Large Language Models from 25+ providers.
The module implements:
- **Precise tokenization** for OpenAI models using the tiktoken library
- **Intelligent approximation algorithms** for all other providers
- **Provider detection** with case-insensitive model name matching
- **Message format support** for chat-based interactions
- **Comprehensive error handling** with detailed error messages
- **Cost estimation** for supported models with USD/INR currency support
Key Components:
- :class:`TokenCounter`: Main class for token counting operations
- :func:`count_tokens`: Convenience function for quick token counting
- :func:`get_supported_models`: Returns all supported models by provider
- :func:`estimate_cost`: Calculates estimated costs for token usage
Provider Support:
The module supports models from major providers including:
- **OpenAI**: GPT-4, GPT-3.5, GPT-4o, O1 models, embeddings
- **Anthropic**: Claude 3/3.5 (Opus, Sonnet, Haiku), Claude 2, Instant
- **Google**: Gemini Pro/Flash, Gemini 1.5/2.0, PaLM models
- **Meta**: LLaMA 2/3/3.1/3.2/3.3 in various sizes
- **Mistral**: Mistral 7B, Mixtral, Mistral Large variants
- **Cohere**: Command, Command-R, Command-R+ models
- **xAI**: Grok 1/1.5/2 and beta models
- **Chinese providers**: Alibaba Qwen, Baidu ERNIE, Huawei PanGu, Tsinghua ChatGLM
- **Code-specialized**: DeepSeek Coder, Replit Code, BigCode StarCoder
- **Open source**: EleutherAI, Stability AI, TII Falcon, RWKV
- **Enterprise**: Databricks DBRX, Microsoft Phi, Amazon Titan, IBM Granite
Tokenization Approach:
- **OpenAI models**: Uses official tiktoken encodings (cl100k_base, p50k_base, r50k_base)
- **Other providers**: Intelligent approximation based on:
- Character count analysis
- Whitespace and punctuation detection
- Provider-specific adjustment factors
- Language-optimized calculations (Chinese, Russian, etc.)
The approximation algorithms are calibrated to provide reasonable accuracy for:
- Cost estimation and budgeting
- Rate limit planning
- Content length assessment
- Comparative analysis across providers
Note:
For production applications requiring exact token counts, use OpenAI models
with tiktoken. For other providers, the approximations are suitable for
planning and estimation purposes.
"""
import re
from typing import Dict, List, Optional, Union, TYPE_CHECKING, Any
if TYPE_CHECKING:
import tiktoken
from anthropic import Anthropic
else:
try:
import tiktoken
except ImportError:
tiktoken = None
try:
from anthropic import Anthropic
except ImportError:
Anthropic = None
from .exceptions import UnsupportedModelError, TokenizationError
from .registry.models import (
ModelInfo,
get_all_supported_models,
get_model_info,
get_supported_models_by_provider,
normalize_model_name,
)
from .registry.pricing import USD_TO_INR, get_pricing_rate
DEFAULT_FORMULA = (4.0, 0.3)
FORMULAS = {
"anthropic": (4.0, 0.3),
"google": (3.8, 0.25),
"meta": (3.5, 0.2),
"mistral": (3.7, 0.25),
"cohere": (4.2, 0.3),
"perplexity": (3.6, 0.2),
"huggingface": (4.0, 0.25),
"ai21": (3.8, 0.25),
"together": (3.9, 0.25),
"xai": (3.8, 0.25),
"alibaba": (3.2, 0.2),
"baidu": (3.3, 0.2),
"huawei": (3.4, 0.2),
"yandex": (3.6, 0.2),
"stability": (3.8, 0.25),
"tii": (3.7, 0.25),
"eleutherai": (3.6, 0.2),
"mosaicml": (3.7, 0.25),
"replit": (3.5, 0.2),
"minimax": (3.3, 0.2),
"aleph_alpha": (3.9, 0.25),
"deepseek": (3.6, 0.2),
"tsinghua": (3.2, 0.2),
"rwkv": (3.8, 0.25),
"community": (3.6, 0.2),
"microsoft": (3.7, 0.25),
"amazon": (3.9, 0.25),
"nvidia": (3.6, 0.2),
"ibm": (3.8, 0.25),
"salesforce": (3.5, 0.2),
"bigcode": (3.4, 0.2),
"databricks": (4.0, 0.25),
"voyage": (3.8, 0.25),
"volcengine": (3.3, 0.2),
"moonshot": (3.7, 0.2),
"glm5": (3.63, 0.15),
"minimax_m25": (3.68, 0.15),
}
# Chars per token for CJK-heavy text (3050 chars benchmark). Used when cjk_ratio > 0.2.
CJK_RATIO = {
"anthropic": 1.59,
"google": 2.07,
"volcengine": 2.04,
"moonshot": 2.43,
"glm5": 2.31,
"minimax_m25": 2.37,
"minimax": 2.37,
"deepseek": 2.24,
}
DEFAULT_CJK_RATIO = 2.0
def _cjk_ratio(text: str) -> float:
"""Return fraction of CJK characters (0..1). CJK chars tokenize ~1 char/token vs ~4 for English."""
if not text:
return 0.0
cjk = sum(1 for c in text if "\u4e00" <= c <= "\u9fff" or "\u3400" <= c <= "\u4dbf" or "\u3000" <= c <= "\u303f")
return cjk / len(text)
class TokenCounter:
"""
A comprehensive token counter for various Large Language Model (LLM) providers.
This class provides functionality to count tokens for 200+ different LLMs from 25+ providers,
including OpenAI, Anthropic, Google, Meta, Mistral, and many others. It supports both
individual text strings and lists of messages (for chat-like interactions).
The token counting is precise for OpenAI models using the official tiktoken library,
and provides reasonable approximations for other providers using intelligent algorithms
calibrated for each provider's tokenization characteristics.
Attributes:
model (str): The model name (converted to lowercase)
provider (str): The detected provider name
tokenizer (Optional[Any]): The tokenizer instance (tiktoken for OpenAI, None for others)
Supported Providers:
- **OpenAI**: GPT-4, GPT-3.5, GPT-4o, O1 models, embeddings (25+ models)
- **Anthropic**: Claude 3/3.5 (Opus, Sonnet, Haiku), Claude 2, Instant (12+ models)
- **Google**: Gemini Pro/Flash, Gemini 1.5/2.0, PaLM (10+ models)
- **Meta**: LLaMA 2/3/3.1/3.2/3.3 in various sizes (15+ models)
- **Mistral**: Mistral 7B, Mixtral, Mistral Large variants (10+ models)
- **Cohere**: Command, Command-R, Command-R+ (8+ models)
- **xAI**: Grok 1/1.5/2 and beta models (4+ models)
- **Alibaba**: Qwen 1.5/2.0/2.5 and vision models (20+ models)
- **Baidu**: ERNIE 3.0/3.5/4.0 and variants (8+ models)
- **Huawei**: PanGu Alpha and Coder models (5+ models)
- **Yandex**: YaLM and YaGPT models (4+ models)
- **DeepSeek**: Coder, VL, and LLM models (8+ models)
- **Tsinghua**: ChatGLM and GLM models (5+ models)
- **And 15+ more providers with specialized models**
Examples:
Basic usage:
.. code-block:: python
# Count tokens for a single text string
counter = TokenCounter("gpt-4")
token_count = counter.count("This is a test string.")
print(f"Token count: {token_count}")
Chat message format:
.. code-block:: python
# Count tokens for a list of messages (chat format)
messages = [
{"role": "user", "content": "Hello!"},
{"role": "assistant", "content": "How can I help you?"},
]
token_count = counter.count_messages(messages)
print(f"Token count (messages): {token_count}")
Different providers:
.. code-block:: python
# Compare token counts across providers
models = ["gpt-4", "claude-3-opus", "gemini-pro", "llama-3-70b"]
text = "Compare tokenization across different models."
for model in models:
counter = TokenCounter(model)
tokens = counter.count(text)
print(f"{model}: {tokens} tokens")
Cost estimation:
.. code-block:: python
from scripts.core import estimate_cost
counter = TokenCounter("gpt-4")
tokens = counter.count("Your text here")
cost = estimate_cost(tokens, "gpt-4", input_tokens=True)
print(f"Estimated cost: .4f")
Tokenization Accuracy:
- **OpenAI models**: Exact token counts using official tiktoken encodings
- **Other providers**: Approximations with typical accuracy of ±10-20%
- **Approximation factors**: Calibrated per provider based on tokenization patterns
- **Language optimization**: Adjusted for Chinese, Russian, and other languages
Note:
For production applications requiring exact token counts, use OpenAI models.
For other providers, approximations are suitable for cost estimation,
rate limit planning, and comparative analysis.
Raises:
UnsupportedModelError: If the specified model is not supported
TokenizationError: If tokenization fails or required dependencies are missing
"""
def __init__(self, model: str):
"""
Initialize the TokenCounter with a specific model.
Sets up the appropriate tokenizer based on the model's provider. For OpenAI models,
initializes the tiktoken tokenizer with the correct encoding. For other providers,
sets up approximation-based token counting.
Args:
model (str): The model name (e.g., 'gpt-4', 'claude-3-opus-20240229', 'gemini-pro').
Model names are case-insensitive and will be converted to lowercase.
Raises:
UnsupportedModelError: If the model is not supported. The exception includes
a list of all supported models for reference.
TokenizationError: If required dependencies are missing (e.g., tiktoken for OpenAI models)
or if tokenizer initialization fails.
Examples:
.. code-block:: python
# OpenAI model (requires tiktoken)
counter = TokenCounter("gpt-4")
# Anthropic model (uses approximation)
counter = TokenCounter("claude-3-opus-20240229")
# Case-insensitive model names
counter = TokenCounter("GPT-4") # Same as "gpt-4"
# Google model
counter = TokenCounter("gemini-pro")
# Meta model
counter = TokenCounter("llama-3-70b")
Note:
The constructor automatically detects the provider based on the model name
and sets up the appropriate tokenization method. OpenAI models use precise
tiktoken-based counting, while other providers use calibrated approximations.
"""
self.tokenizer: Optional[Any] = None
self.model = normalize_model_name(model)
self.model_info = self._get_model_info()
self.provider = self.model_info.provider
self._setup_tokenizer()
def _get_model_info(self) -> ModelInfo:
model_info = get_model_info(self.model)
if not model_info:
raise UnsupportedModelError(self.model, get_all_supported_models())
return model_info
def _detect_provider(self) -> str:
return self.model_info.provider
def _setup_tokenizer(self) -> None:
"""
Setup the appropriate tokenizer for the model based on its provider.
For OpenAI models, initializes the tiktoken tokenizer with the correct encoding
(cl100k_base, p50k_base, or r50k_base). For all other providers, sets the
tokenizer to None to indicate approximation-based counting will be used.
OpenAI Encodings:
- **cl100k_base**: GPT-4, GPT-3.5-turbo, GPT-4o, embeddings (most models)
- **p50k_base**: text-davinci-003, text-davinci-002 (legacy completion models)
- **r50k_base**: GPT-3, davinci, curie, babbage, ada (oldest models)
Raises:
TokenizationError: If tiktoken is not installed for OpenAI models, or if
the tokenizer fails to initialize for any reason.
Examples:
The tokenizer setup is automatic and transparent:
.. code-block:: python
# OpenAI model - sets up tiktoken with cl100k_base encoding
counter = TokenCounter("gpt-4")
# Anthropic model - sets tokenizer to None for approximation
counter = TokenCounter("claude-3-opus")
Note:
This method is called automatically during initialization. Users should not
call this method directly. The tokenizer instance is stored in self.tokenizer
and used by the count() method.
Dependencies:
- **tiktoken**: Required for OpenAI models. Install with: ``pip install tiktoken``
- **No dependencies**: Required for other providers (uses built-in approximation)
"""
if self.provider == "openai":
if tiktoken is None:
raise TokenizationError(
"tiktoken is required for OpenAI models. Install with: pip install tiktoken",
model=self.model
)
encoding_name = self.model_info.encoding or "cl100k_base"
try:
self.tokenizer = tiktoken.get_encoding(encoding_name)
except Exception as e:
raise TokenizationError(f"Failed to load tokenizer: {str(e)}", model=self.model)
else:
# For all other providers, we'll use approximation since they don't provide public tokenizers
self.tokenizer = None
def count(self, text: str) -> int:
"""
Count tokens in the given text.
Performs token counting using the appropriate method for the model's provider.
For OpenAI models, uses precise tiktoken-based counting. For other providers,
uses intelligent approximation algorithms calibrated for each provider.
Args:
text (str): The text to count tokens for. Must be a string.
Returns:
int: The number of tokens in the text. Returns 0 for empty strings.
Raises:
TokenizationError: If tokenization fails, input is invalid, or required
dependencies are missing. Includes detailed error context
with model name and text preview.
Input Validation:
The method performs comprehensive input validation:
- **None check**: Rejects None input with clear error message
- **Type check**: Ensures input is a string, not int/float/list/dict/etc.
- **Empty string**: Returns 0 for empty strings (valid case)
Tokenization Methods:
- **OpenAI models**: Uses tiktoken.encode() for exact token counts
- **Other providers**: Uses _approximate_tokens() with provider-specific calibration
Provider-Specific Accuracy:
- **OpenAI**: 100% accurate (official tokenizer)
- **Anthropic**: ~90-95% accurate (well-calibrated approximation)
- **Google**: ~85-90% accurate (Gemini-optimized approximation)
- **Meta**: ~85-90% accurate (LLaMA-optimized approximation)
- **Chinese models**: ~80-90% accurate (character-optimized for Chinese)
- **Code models**: ~85-95% accurate (code-pattern optimized)
- **Other providers**: ~80-90% accurate (general approximation)
Examples:
Basic usage:
.. code-block:: python
counter = TokenCounter("gpt-4")
# Simple text
tokens = counter.count("Hello, world!")
print(f"Tokens: {tokens}") # Exact count for OpenAI
# Empty string
tokens = counter.count("")
print(f"Tokens: {tokens}") # Always returns 0
# Longer text
text = "This is a longer text that will be tokenized."
tokens = counter.count(text)
print(f"Tokens: {tokens}")
Comparing providers:
.. code-block:: python
text = "Compare tokenization across different models."
models = ["gpt-4", "claude-3-opus", "gemini-pro"]
for model in models:
counter = TokenCounter(model)
tokens = counter.count(text)
print(f"{model}: {tokens} tokens")
Error handling:
.. code-block:: python
try:
counter = TokenCounter("gpt-4")
tokens = counter.count("Valid text")
except TokenizationError as e:
print(f"Tokenization failed: {e}")
Performance:
- **OpenAI models**: Fast (native tiktoken performance)
- **Other providers**: Very fast (lightweight approximation algorithms)
- **Typical speed**: 10,000+ texts per second for approximation methods
Note:
For production applications requiring exact token counts, use OpenAI models.
For cost estimation, rate limiting, and comparative analysis, approximations
provide sufficient accuracy with much better performance.
"""
# Comprehensive input validation
if text is None:
raise TokenizationError("Input cannot be None", model=self.model)
if not isinstance(text, str):
# Handle common invalid types explicitly
if isinstance(text, (int, float, bool)):
raise TokenizationError(f"Input must be a string, got {type(text).__name__}", model=self.model)
elif isinstance(text, (list, tuple, dict, set)):
raise TokenizationError(f"Input must be a string, got {type(text).__name__}", model=self.model)
else:
raise TokenizationError("Input must be a string", model=self.model)
# Handle empty string case
if not text:
return 0
try:
if self.provider == "openai":
if self.tokenizer is None:
raise TokenizationError("Tokenizer not initialized", model=self.model)
if not text:
return 0
return len(self.tokenizer.encode(text))
else:
# Use approximation for all other providers
return self._approximate_tokens(text)
except TokenizationError:
# Re-raise TokenizationError as-is
raise
except Exception as e:
raise TokenizationError(str(e), model=self.model, text_preview=text)
def _approximate_tokens(self, text: str) -> int:
"""
Approximate token count for non-OpenAI models.
Uses intelligent approximation algorithms calibrated for each provider's
tokenization characteristics. The approximation considers character count,
whitespace patterns, punctuation density, and provider-specific factors.
Args:
text (str): The text to approximate tokens for. Must be a string.
Returns:
int: Approximated number of tokens. Minimum return value is 1 for non-empty text.
Raises:
TokenizationError: If text processing fails or input is invalid.
Algorithm Components:
The approximation algorithm analyzes several text characteristics:
1. **Character count**: Base measurement for token estimation
2. **Whitespace analysis**: Spaces and newlines often become separate tokens
3. **Punctuation analysis**: Special characters frequently tokenize separately
4. **Provider calibration**: Adjustment factors based on tokenizer characteristics
Provider-Specific Calibrations:
Each provider has calibrated ratios based on empirical analysis:
**Major Providers:**
- **Anthropic**: ~4 chars/token (Claude guidance), +30% punctuation adjustment
- **Google**: ~3.8 chars/token (Gemini-optimized), +25% adjustment
- **Meta**: ~3.5 chars/token (LLaMA-optimized), +20% adjustment
- **Mistral**: ~3.7 chars/token (GPT-similar), +25% adjustment
- **Cohere**: ~4.2 chars/token (conservative), +30% adjustment
**Regional/Language-Optimized:**
- **Alibaba/Qwen**: ~3.2 chars/token (Chinese-optimized)
- **Baidu/ERNIE**: ~3.3 chars/token (Chinese-optimized)
- **Huawei/PanGu**: ~3.4 chars/token (Chinese-optimized)
- **Yandex/YaLM**: ~3.6 chars/token (Russian-optimized)
- **Tsinghua/ChatGLM**: ~3.2 chars/token (Chinese-optimized)
**Code-Specialized:**
- **DeepSeek Coder**: ~3.6 chars/token (code-optimized)
- **Replit Code**: ~3.5 chars/token (code-optimized)
- **BigCode StarCoder**: ~3.4 chars/token (code-optimized)
- **Salesforce CodeGen**: ~3.5 chars/token (code-optimized)
Accuracy Expectations:
- **Well-calibrated providers**: ±10-15% accuracy
- **Language-optimized**: ±15-20% for target languages
- **General approximation**: ±20-25% accuracy
- **Code models**: ±10-20% for code content
Examples:
This method is called automatically by count() for non-OpenAI models:
.. code-block:: python
# Automatic approximation for Anthropic
counter = TokenCounter("claude-3-opus")
tokens = counter.count("Hello, world!") # Uses _approximate_tokens()
# Different providers give different approximations
text = "This is a test sentence with punctuation!"
anthropic_counter = TokenCounter("claude-3-opus")
google_counter = TokenCounter("gemini-pro")
meta_counter = TokenCounter("llama-3-70b")
print(f"Anthropic: {anthropic_counter.count(text)} tokens")
print(f"Google: {google_counter.count(text)} tokens")
print(f"Meta: {meta_counter.count(text)} tokens")
Performance:
- **Speed**: Very fast, 10,000+ approximations per second
- **Memory**: Minimal memory usage, no model loading required
- **Dependencies**: No external dependencies required
Note:
This method should not be called directly. Use the count() method instead,
which automatically selects the appropriate tokenization method based on
the model's provider.
"""
if not isinstance(text, str):
raise TokenizationError(f"Input must be a string, got {type(text).__name__}", model=self.model)
if not text:
return 0
try:
# Basic character-based approximation
char_count = len(text)
# Adjust for whitespace (spaces and newlines are often separate tokens)
whitespace_count = len(re.findall(r'\s+', text))
# Adjust for punctuation (often separate tokens)
punctuation_count = len(re.findall(r'[^\w\s]', text))
except Exception as e:
raise TokenizationError(f"Failed to process text: {str(e)}", model=self.model, text_preview=text)
ratio, adjustment_factor = FORMULAS.get(self.model_info.formula or self.provider, DEFAULT_FORMULA)
formula_key = self.model_info.formula or self.provider
cjk = _cjk_ratio(text)
if cjk > 0.2:
cjk_chars_per_token = CJK_RATIO.get(formula_key) or CJK_RATIO.get(self.provider) or DEFAULT_CJK_RATIO
ratio = ratio * (1 - cjk) + cjk_chars_per_token * cjk
base_tokens = char_count / ratio
adjustment = (whitespace_count + punctuation_count) * adjustment_factor
return max(1, int(base_tokens + adjustment))
def count_messages(self, messages: List[Dict[str, str]]) -> int:
"""
Count tokens for a list of messages in chat format.
Processes a list of message dictionaries (typical chat/conversation format)
and returns the total token count including any formatting overhead. This
method is essential for chat-based applications and conversation analysis.
Args:
messages (List[Dict[str, str]]): List of message dictionaries. Each message
must contain 'role' and 'content' keys.
Expected format:
.. code-block:: python
[
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Hello!"},
{"role": "assistant", "content": "Hi there!"}
]
Returns:
int: Total token count for all messages including formatting overhead.
Raises:
TokenizationError: If messages format is invalid, contains non-string content,
or if tokenization of individual messages fails. Includes
detailed error context with message index and content preview.
Message Format Validation:
The method performs comprehensive validation:
- **Input type**: Must be a list, not string/dict/int/etc.
- **Message structure**: Each message must be a dictionary
- **Required keys**: Each message must have 'role' and 'content' keys
- **Content type**: Message content must be a string, not None/int/list/etc.
- **Role type**: Message role must be a string if present
Formatting Overhead:
Different providers handle message formatting differently:
- **OpenAI**: Minimal overhead (~1 token per role)
- **Anthropic**: No additional formatting overhead
- **Other providers**: No additional overhead assumed
Common Message Roles:
- **system**: System instructions or context
- **user**: User input or questions
- **assistant**: AI assistant responses
- **function**: Function call results (some providers)
Examples:
Basic chat conversation:
.. code-block:: python
counter = TokenCounter("gpt-4")
messages = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "What is the capital of France?"},
{"role": "assistant", "content": "The capital of France is Paris."}
]
total_tokens = counter.count_messages(messages)
print(f"Total conversation tokens: {total_tokens}")
Comparing individual vs. message counting:
.. code-block:: python
counter = TokenCounter("gpt-4")
# Count individual messages
individual_total = 0
for msg in messages:
tokens = counter.count(msg["content"])
individual_total += tokens
print(f"{msg['role']}: {tokens} tokens")
# Count as message format (includes formatting overhead)
message_total = counter.count_messages(messages)
print(f"Individual sum: {individual_total}")
print(f"Message format: {message_total}")
print(f"Formatting overhead: {message_total - individual_total}")
Error handling:
.. code-block:: python
try:
counter = TokenCounter("gpt-4")
# Invalid format - missing content
invalid_messages = [{"role": "user"}]
tokens = counter.count_messages(invalid_messages)
except TokenizationError as e:
print(f"Message format error: {e}")
Multi-provider comparison:
.. code-block:: python
messages = [
{"role": "user", "content": "Hello!"},
{"role": "assistant", "content": "Hi there! How can I help?"}
]
models = ["gpt-4", "claude-3-opus", "gemini-pro"]
for model in models:
counter = TokenCounter(model)
tokens = counter.count_messages(messages)
print(f"{model}: {tokens} tokens")
Performance:
- **Speed**: Processes thousands of message lists per second
- **Memory**: Minimal additional memory overhead
- **Scalability**: Handles conversations with hundreds of messages
Use Cases:
- **Chat applications**: Calculate conversation costs
- **API rate limiting**: Plan request sizes for chat endpoints
- **Conversation analysis**: Analyze dialogue token patterns
- **Cost estimation**: Budget for chat-based AI applications
- **Content moderation**: Assess conversation length and complexity
Note:
This method is specifically designed for chat/conversation formats.
For simple text token counting, use the count() method instead.
"""
# Comprehensive input validation
if messages is None:
raise TokenizationError("Messages cannot be None", model=self.model)
if not isinstance(messages, list):
# Handle common invalid types explicitly
if isinstance(messages, str):
raise TokenizationError("Messages must be a list, got string", model=self.model)
elif isinstance(messages, (int, float, bool)):
raise TokenizationError(f"Messages must be a list, got {type(messages).__name__}", model=self.model)
elif isinstance(messages, (dict, tuple, set)):
raise TokenizationError(f"Messages must be a list, got {type(messages).__name__}", model=self.model)
else:
raise TokenizationError("Messages must be a list", model=self.model)
total_tokens = 0
for i, message in enumerate(messages):
if not isinstance(message, dict):
raise TokenizationError(f"Message at index {i} must be a dict, got {type(message).__name__}", model=self.model)
if 'role' not in message:
raise TokenizationError(f"Message at index {i} must have 'role' key", model=self.model)
if 'content' not in message:
raise TokenizationError(f"Message at index {i} must have 'content' key", model=self.model)
# Validate content is a string
if not isinstance(message['content'], str):
if message['content'] is None:
raise TokenizationError(f"Message content at index {i} cannot be None", model=self.model)
else:
raise TokenizationError(f"Message content at index {i} must be a string, got {type(message['content']).__name__}", model=self.model)
# Validate role if present
if 'role' in message and not isinstance(message['role'], str):
if message['role'] is None:
raise TokenizationError(f"Message role at index {i} cannot be None", model=self.model)
else:
raise TokenizationError(f"Message role at index {i} must be a string, got {type(message['role']).__name__}", model=self.model)
# Count content tokens
try:
content_tokens = self.count(message['content'])
total_tokens += content_tokens
except TokenizationError:
# Re-raise TokenizationError as-is
raise
except Exception as e:
raise TokenizationError(f"Failed to count tokens for message at index {i}: {str(e)}", model=self.model)
# Add overhead for message formatting (extremely minimal overhead)
if self.provider == "openai":
# OpenAI adds minimal formatting overhead
if 'role' in message:
total_tokens += 1 # Role is typically 1 token
elif self.provider == "anthropic":
# Claude has no additional formatting overhead beyond content
pass
else:
# Other providers have no additional overhead
pass
# No additional final message overhead
return total_tokens
def count_tokens(text: str, model: str) -> int:
"""
Convenience function to count tokens for a given text and model.
This is a simplified interface that creates a TokenCounter instance and
performs token counting in a single function call. Ideal for one-off
token counting operations without needing to manage TokenCounter instances.
Args:
text (str): The text to count tokens for. Must be a string.
model (str): The model name (e.g., 'gpt-4', 'claude-3-opus-20240229').
Model names are case-insensitive.
Returns:
int: The number of tokens in the text.
Raises:
UnsupportedModelError: If the specified model is not supported.
TokenizationError: If tokenization fails or input is invalid.
Examples:
Basic usage:
.. code-block:: python
from scripts import count_tokens
# OpenAI model
tokens = count_tokens("Hello, world!", "gpt-4")
print(f"GPT-4 tokens: {tokens}")
# Anthropic model
tokens = count_tokens("Hello, world!", "claude-3-opus")
print(f"Claude tokens: {tokens}")
# Case-insensitive model names
tokens = count_tokens("Hello, world!", "GPT-4") # Same as "gpt-4"
Comparing models:
.. code-block:: python
text = "This is a sample text for comparison."
models = ["gpt-4", "gpt-3.5-turbo", "claude-3-opus", "gemini-pro"]
for model in models:
tokens = count_tokens(text, model)
print(f"{model}: {tokens} tokens")
Error handling:
.. code-block:: python
try:
tokens = count_tokens("Hello!", "unsupported-model")
except UnsupportedModelError as e:
print(f"Model not supported: {e}")
except TokenizationError as e:
print(f"Tokenization failed: {e}")
Performance:
This function creates a new TokenCounter instance for each call.
For multiple operations with the same model, consider using
TokenCounter directly for better performance:
.. code-block:: python
# Less efficient for multiple calls
for text in texts:
tokens = count_tokens(text, "gpt-4")
# More efficient for multiple calls
counter = TokenCounter("gpt-4")
for text in texts:
tokens = counter.count(text)
Note:
This function is equivalent to:
.. code-block:: python
counter = TokenCounter(model)
return counter.count(text)
"""
counter = TokenCounter(model)
return counter.count(text)
def get_supported_models() -> Dict[str, List[str]]:
"""
Get a comprehensive dictionary of supported models organized by provider.
Returns all 200+ supported models grouped by their respective providers,
making it easy to discover available models and understand the scope
of scripts's capabilities.
Returns:
Dict[str, List[str]]: Dictionary with provider names as keys and lists
of model names as values. Providers include:
- **openai**: GPT-4, GPT-3.5, GPT-4o, O1, embeddings (25+ models)
- **anthropic**: Claude 3/3.5, Claude 2, Instant (12+ models)
- **google**: Gemini Pro/Flash, Gemini 1.5/2.0, PaLM (10+ models)
- **meta**: LLaMA 2/3/3.1/3.2/3.3 variants (15+ models)
- **mistral**: Mistral 7B, Mixtral, Large variants (10+ models)
- **cohere**: Command, Command-R, Command-R+ (8+ models)
- **xai**: Grok 1/1.5/2 and beta models (4+ models)
- **alibaba**: Qwen 1.5/2.0/2.5 and vision models (20+ models)
- **baidu**: ERNIE 3.0/3.5/4.0 variants (8+ models)
- **huawei**: PanGu Alpha and Coder models (5+ models)
- **yandex**: YaLM and YaGPT models (4+ models)
- **deepseek**: Coder, VL, and LLM models (8+ models)
- **tsinghua**: ChatGLM and GLM models (5+ models)
- **databricks**: DBRX and Dolly models (6+ models)
- **voyage**: Voyage embedding models (6+ models)
- **And 10+ more providers**
Examples:
Basic usage:
.. code-block:: python
from scripts import get_supported_models
models = get_supported_models()
# List all providers
print("Supported providers:")
for provider in models.keys():
print(f" {provider}")
Explore specific providers:
.. code-block:: python
models = get_supported_models()
# OpenAI models
print("OpenAI models:")
for model in models["openai"]:
print(f" {model}")
# Anthropic models
print("\\nAnthropic models:")
for model in models["anthropic"]:
print(f" {model}")
Count models by provider:
.. code-block:: python
models = get_supported_models()
print("Model counts by provider:")
total_models = 0
for provider, model_list in models.items():
count = len(model_list)
total_models += count
print(f" {provider}: {count} models")
print(f"\\nTotal: {total_models} models")
Find models by pattern:
.. code-block:: python
models = get_supported_models()
# Find all GPT-4 variants
gpt4_models = []
for model in models["openai"]:
if "gpt-4" in model:
gpt4_models.append(model)
print("GPT-4 variants:")
for model in gpt4_models:
print(f" {model}")
Validate model support:
.. code-block:: python
models = get_supported_models()
def is_model_supported(model_name):
model_lower = model_name.lower()
for provider_models in models.values():
if model_lower in [m.lower() for m in provider_models]:
return True
return False
# Check if models are supported
test_models = ["gpt-4", "claude-3-opus", "unknown-model"]
for model in test_models:
supported = is_model_supported(model)
print(f"{model}: {'✓' if supported else '✗'}")
Integration with TokenCounter:
.. code-block:: python
from scripts import TokenCounter, get_supported_models
models = get_supported_models()
text = "Test tokenization across providers."
# Test a few models from each major provider
test_models = {
"openai": models["openai"][0], # First OpenAI model
"anthropic": models["anthropic"][0], # First Anthropic model
"google": models["google"][0], # First Google model
"meta": models["meta"][0] # First Meta model
}
for provider, model in test_models.items():
counter = TokenCounter(model)
tokens = counter.count(text)
print(f"{provider} ({model}): {tokens} tokens")
Provider Categories:
The returned dictionary includes models from these categories:
**Major Cloud Providers:**
- OpenAI, Anthropic, Google, Microsoft, Amazon
**AI-First Companies:**
- Mistral, Cohere, xAI, Perplexity, AI21
**Regional/Language-Specific:**
- Alibaba (Chinese), Baidu (Chinese), Huawei (Chinese)
- Yandex (Russian), Tsinghua (Chinese)
**Open Source/Research:**
- EleutherAI, Stability AI, TII, RWKV, Community models
**Enterprise/Specialized:**
- Databricks, Voyage, DeepSeek, BigCode, Replit
- Nvidia, IBM, Salesforce
Note:
The model lists are comprehensive but may not include every variant
or the very latest models. The library is regularly updated to
include new models as they become available.
See Also:
- :class:`TokenCounter`: For creating token counters with specific models
- :func:`count_tokens`: For quick token counting with model validation
- :exc:`UnsupportedModelError`: Exception raised for unsupported models
"""
return get_supported_models_by_provider()
def estimate_cost(
token_count: int,
model: str,
input_tokens: bool = True,
currency: str = "USD"
) -> float:
"""
Estimate the cost for a given number of tokens and model.
Calculates estimated costs based on current pricing for supported models.
Supports both input and output token pricing, as many models have different
rates for input vs. output tokens. Provides costs in USD or INR currency.
Args:
token_count (int): Number of tokens to estimate cost for. Must be non-negative.
model (str): Model name (e.g., "gpt-4", "gpt-4o", "claude-3-opus-20240229").
Model names are case-insensitive.
input_tokens (bool, optional): True for input token pricing, False for output
token pricing. Defaults to True. Many models charge
more for output tokens than input tokens.
currency (str, optional): Currency code ("USD" or "INR"). Defaults to "USD".
Uses current conversion rate for INR.
Returns:
float: Estimated cost in the specified currency. Returns 0.0 if the model
is not in the pricing database or if pricing is not available.
Pricing Coverage:
The function includes pricing for major models:
**OpenAI Models:**
- GPT-4: $0.03/$0.06 per 1K tokens (input/output)
- GPT-4 Turbo: $0.01/$0.03 per 1K tokens
- GPT-4o: $0.005/$0.015 per 1K tokens
- GPT-4o Mini: $0.00015/$0.0006 per 1K tokens
- GPT-3.5 Turbo: $0.001/$0.002 per 1K tokens
**Anthropic Models:**
- Claude-3 Opus: $0.015/$0.075 per 1K tokens
- Claude-3 Sonnet: $0.003/$0.015 per 1K tokens
- Claude-3 Haiku: $0.00025/$0.00125 per 1K tokens
- Claude-3.5 Sonnet: $0.003/$0.015 per 1K tokens
- Claude-3.5 Haiku: $0.001/$0.005 per 1K tokens
**Databricks Models:**
- DBRX Instruct: $0.001/$0.002 per 1K tokens
- Dolly models: $0.001/$0.002 per 1K tokens
**Voyage AI Models:**
- All Voyage models: $0.0001/$0.0001 per 1K tokens
Examples:
Basic cost estimation:
.. code-block:: python
from scripts import count_tokens, estimate_cost
text = "This is a sample text for cost estimation."
model = "gpt-4"
# Count tokens and estimate cost
tokens = count_tokens(text, model)
input_cost = estimate_cost(tokens, model, input_tokens=True)
output_cost = estimate_cost(tokens, model, input_tokens=False)
print(f"Text: '{text}'")
print(f"Tokens: {tokens}")
print(f"Input cost: .4f")
print(f"Output cost: .4f")
Compare costs across models:
.. code-block:: python
text = "Compare costs across different models." * 100 # Longer text
models = ["gpt-4", "gpt-4o", "gpt-3.5-turbo", "claude-3-opus", "claude-3-haiku"]
print(f"Text length: {len(text)} characters")
print("\\nCost comparison:")
for model in models:
try:
tokens = count_tokens(text, model)
input_cost = estimate_cost(tokens, model, input_tokens=True)
output_cost = estimate_cost(tokens, model, input_tokens=False)
print(f"{model}:")
print(f" Tokens: {tokens}")
print(f" Input: .4f")
print(f" Output: .4f")
except Exception as e:
print(f"{model}: Error - {e}")
Currency conversion:
.. code-block:: python
tokens = 1000
model = "gpt-4"
# USD pricing
cost_usd = estimate_cost(tokens, model, currency="USD")
print(f"Cost in USD: .4f")
# INR pricing
cost_inr = estimate_cost(tokens, model, currency="INR")
print(f"Cost in INR: ₹{cost_inr:.2f}")
Batch cost estimation:
.. code-block:: python
texts = [
"Short text",
"Medium length text with more content",
"Much longer text that will cost more to process" * 10
]
model = "gpt-4o"
total_cost = 0
print("Individual text costs:")
for i, text in enumerate(texts, 1):
tokens = count_tokens(text, model)
cost = estimate_cost(tokens, model)
total_cost += cost
print(f"Text {i}: {tokens} tokens, .4f")
print(f"\\nTotal estimated cost: .4f")
Chat conversation costing:
.. code-block:: python
from scripts import TokenCounter
messages = [
{"role": "system", "content": "You are a helpful assistant."},
{"role": "user", "content": "Explain quantum computing."},
{"role": "assistant", "content": "Quantum computing is a revolutionary..."}
]
counter = TokenCounter("gpt-4")
total_tokens = counter.count_messages(messages)
# Estimate costs for the conversation
input_cost = estimate_cost(total_tokens, "gpt-4", input_tokens=True)
output_cost = estimate_cost(total_tokens, "gpt-4", input_tokens=False)
print(f"Conversation tokens: {total_tokens}")
print(f"If all input: .4f")
print(f"If all output: .4f")
Currency Conversion:
- **USD to INR rate**: 83.0 (as of July 2025)
- **Rate updates**: The conversion rate is periodically updated
- **Precision**: INR costs are calculated from USD base prices
Limitations:
- **Pricing accuracy**: Based on publicly available pricing, may not reflect
current rates or enterprise discounts
- **Model coverage**: Only includes models with known pricing
- **Rate changes**: Pricing may change without notice
- **Approximation**: For non-OpenAI models, token counts are approximated
Note:
This function provides cost estimates for planning and budgeting purposes.
Actual costs may vary based on current pricing, volume discounts, and
exact tokenization. Always verify current pricing with the model provider
for production applications.
See Also:
- :func:`count_tokens`: For getting token counts to use with this function
- :class:`TokenCounter`: For more complex token counting scenarios
- :func:`get_supported_models`: For checking which models are available
"""
model = normalize_model_name(model)
model_info = get_model_info(model)
pricing_key = model_info.pricing_id if model_info else model
rate = get_pricing_rate(pricing_key, input_tokens)
if rate is None:
return 0.0
cost_usd: float = (token_count / 1000) * rate
return cost_usd * USD_TO_INR if currency.upper() == "INR" else cost_usd
FILE:scripts/examples/batch_compare.py
#!/usr/bin/env python3
"""
Compare token counts across multiple models for local file(s).
Run from project root:
python scripts/examples/batch_compare.py file1.txt file2.txt
python scripts/examples/batch_compare.py -f input.txt -m gpt-4 claude-3-opus
"""
import argparse
import sys
from pathlib import Path
# Ensure project root is on path
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from scripts.core import TokenCounter
from scripts.exceptions import UnsupportedModelError, TokenizationError
def main() -> None:
parser = argparse.ArgumentParser(
description="Compare token counts across models for local file(s)"
)
parser.add_argument(
"paths",
nargs="*",
metavar="FILE",
help="Local file path(s) to compare",
)
parser.add_argument(
"-f", "--file",
action="append",
dest="files",
help="Read from file (repeatable)",
)
parser.add_argument(
"-m", "--models",
nargs="*",
default=["gpt-4", "claude-3-opus", "gemini-pro"],
help="Model names (default: gpt-4 claude-3-opus gemini-pro)",
)
args = parser.parse_args()
paths = list(args.paths or [])
if args.files:
paths.extend(args.files)
if not paths:
parser.error("Provide at least one local file path")
models = args.models or ["gpt-4", "claude-3-opus", "gemini-pro"]
print(f"Files: {paths}")
print(f"Models: {models}")
print("---")
for path_str in paths:
path = Path(path_str)
if not path.exists():
print(f"Error: File '{path}' not found", file=sys.stderr)
sys.exit(1)
text = path.read_text(encoding="utf-8")
print(f"\n{path_str}:")
for model in models:
try:
tokens = TokenCounter(model).count(text)
print(f" {model}: tokens={tokens}")
except (UnsupportedModelError, TokenizationError) as e:
print(f" {model}: Error - {e}")
if __name__ == "__main__":
main()
FILE:scripts/examples/benchmark_token_ratio.py
#!/usr/bin/env python3
"""
批量测试多个模型的 token 与文字比例。
- API 模式：通过请求模型 API 获取 token 数（需配置 API_KEY、BASE_URL）
- 本地模式（--local）：使用本项目的 TokenCounter 近似计算，无需 API
"""
import argparse
import asyncio
import sys
from datetime import datetime
from pathlib import Path
from typing import Optional
# 测试文本：包含中文、英文、符号（使用 r 前缀避免 \U \u 等被当作转义）
SAMPLE_TEXT = r"""
---
name: prompt-token-counter
description: "Count tokens and estimate costs for 300+ LLM models. Primary use: audit OpenClaw workspace token consumption (memory, persona, skills)."
trigger: "token count, cost estimate, prompt length, API cost, OpenClaw audit, workspace token usage, memory/persona/skills tokens, context window limit"
---
# Prompt Token Counter (toksum)
> **First load reminder:** This skill provides the `scripts` CLI (toksum). Use it when the user asks to count tokens, estimate API costs, or **audit OpenClaw component token consumption** (memory, persona, skills).
## Primary Use: OpenClaw Token Consumption Audit
**Goal:** Help users identify which OpenClaw components consume tokens and how much.
### 1. Memory & Persona Files
These files are injected into sessions and consume tokens. Search and count them:
| File | Purpose | Typical Location |
|------|---------|------------------|
| `AGENTS.md` | Operating instructions, workflow, priorities | `~/.openclaw/workspace/` |
| `SOUL.md` | Persona, tone, values, behavioral guidelines | `~/.openclaw/workspace/` |
| `IDENTITY.md` | Name, role, goals, visual description | `~/.openclaw/workspace/` |
| `USER.md` | User preferences, communication style | `~/.openclaw/workspace/` |
| `MEMORY.md` | Long-term memory, persistent facts | `~/.openclaw/workspace/` |
| `TOOLS.md` | Tool quirks, path conventions | `~/.openclaw/workspace/` |
| `HEARTBEAT.md` | Periodic maintenance checklist | `~/.openclaw/workspace/` |
| `BOOT.md` | Startup ritual (when hooks enabled) | `~/.openclaw/workspace/` |
| `memory/YYYY-MM-DD.md` | Daily memory logs | `~/.openclaw/workspace/memory/` |
**Workspace path:** Default `~/.openclaw/workspace`; may be overridden in `~/.openclaw/openclaw.json` via `agent.workspace`.
### 2. Skill Files (SKILL.md)
Skills are loaded per session. Count each `SKILL.md`:
| Location | Scope |
|----------|-------|
| `~/.openclaw/skills/*/SKILL.md` | OpenClaw managed skills |
| `~/.openclaw/workspace/skills/*/SKILL.md` | Workspace-specific skills (override) |
### 3. Audit Workflow
1. **Locate workspace:** Resolve `~/.openclaw/workspace` (or config override).
2. **Collect files:** List all memory/persona files and `SKILL.md` paths above.
3. **Count tokens:** For each file, run `python -m scripts.cli -f <path> -m <model> -c`.
4. **Summarize:** Group by category (memory, persona, skills), report total and per-file.
**Example audit command (PowerShell):**
```powershell
$ws = "$env:USERPROFILE\.openclaw\workspace"
python -m scripts.cli -m gpt-4o -c -f "$ws\AGENTS.md" -f "$ws\SOUL.md" -f "$ws\USER.md" -f "$ws\IDENTITY.md" -f "$ws\MEMORY.md" -f "$ws\TOOLS.md"
```
**Example audit (Bash):**
```bash
WS=~/.openclaw/workspace
python -m scripts.cli -m gpt-4o -c -f "$WS/AGENTS.md" -f "$WS/SOUL.md" -f "$WS/USER.md" -f "$WS/IDENTITY.md" -f "$WS/MEMORY.md" -f "$WS/TOOLS.md"
```
---
## Project Layout
```
prompt_token_counter/
├── SKILL.md
├── package.json # npm package (OpenClaw skill)
├── publish_npm.py # Publish to npm; syncs version
└── scripts/ # Python package, CLI + examples
├── cli.py # Entry point
├── core.py # TokenCounter, estimate_cost
├── registry/
│ ├── models.py # 300+ models
│ └── pricing.py # Pricing data
└── examples/ # Script examples
├── count_prompt.py
├── estimate_cost.py
├── batch_compare.py
└── benchmark_token_ratio.py
```
Invoke: `python -m scripts.cli` from project root.
### Version Sync (publish_npm.py)
When publishing to npm, `publish_npm.py` bumps the patch version and syncs it to:
- `package.json` — `version`
- `SKILL.md` — frontmatter `version`
- `scripts/__init__.py` — `__version__`
Run: `python publish_npm.py` (after `npm login`).
---
## Runtime Dependencies
- **Python 3** — required
- **tiktoken** (optional) — `pip install tiktoken` for exact OpenAI counts
---
## Language Rule
**Respond in the user's language.** Match the user's language (e.g. Chinese if they write in Chinese, English if they write in English).
---
## URL Usage — Mandatory Agent Rule
**Before using `-u` / `--url` to fetch content from any URL, you MUST:**
1. **Explicitly warn the user** that the CLI will make an outbound HTTP/HTTPS request to the given URL.
2. **Confirm the URL is trusted** — tell the user: "Only use URLs you fully trust. Untrusted URLs may expose your IP, leak data, or be used for SSRF. Do you confirm this URL is safe?"
3. **Prefer alternatives** — if the user can provide the content via `-f` (local file) or inline text, suggest that instead of URL fetch.
4. **Never auto-fetch** — do not invoke `-u` without the user having explicitly provided the URL and acknowledged the risk.
**If the user insists on using a URL:** Proceed only after they confirm. State clearly: "I will fetch from [URL] to count tokens. Proceed?"
---
## Model Name — Mandatory Agent Rule
**Before invoking the CLI, you MUST have a concrete model name from the user.**
1. **Require explicit model** — `-m` / `--model` is required. Do not guess or assume; the user must provide the exact name (e.g. gpt-4o, claude-3-5-sonnet-20241022).
2. **If unclear, ask** — if the user says "GPT" or "Claude" or "the latest model" without a specific name, ask: "Please specify the exact model name (e.g. gpt-4o, claude-3-5-sonnet-20241022). Run `python -m scripts.cli -l` to list supported models."
3. **Do not auto-pick** — never substitute a model on behalf of the user without their confirmation.
4. **Validate when possible** — if the model name seems ambiguous, offer `-l` output or confirm: "I'll use [model]. Is that correct?"
---
## CLI Usage
```bash
python -m scripts.cli [OPTIONS] [TEXT ...]
```
| Option | Short | Description |
|--------|-------|-------------|
| `--model` | `-m` | Model name (required unless `--list-models`) — **Agent must obtain exact name from user; ask if unclear** |
| `--file` | `-f` | Read from file (repeatable) |
| `--url` | `-u` | Read from URL (repeatable) — **Agent must warn user before use; only trusted URLs** |
| `--list-models` | `-l` | List supported models |
| `--cost` | `-c` | Show cost estimate |
| `--output-tokens` | | Use output token pricing |
| `--currency` | | USD or INR |
| `--verbose` | `-v` | Detailed output |
### Examples
```bash
# Inline text
python -m scripts.cli -m gpt-4 "Hello, world!"
# File with cost
python -m scripts.cli -f input.txt -m claude-3-opus -c
# Multiple files (OpenClaw audit)
python -m scripts.cli -v -c -f AGENTS.md -f SOUL.md -f MEMORY.md -m gpt-4o
# List models
python -m scripts.cli -l
# Run bundled example scripts
python scripts/examples/count_prompt.py "Hello, world!" gpt-4
python scripts/examples/estimate_cost.py "Your text" gpt-4
python scripts/examples/batch_compare.py "Compare text" gpt-4 claude-3-opus
```
---
## Python API
```python
from scripts import TokenCounter, count_tokens, estimate_cost, get_supported_models
tokens = count_tokens("Hello!", "gpt-4")
counter = TokenCounter("claude-3-opus")
tokens = counter.count_messages([
{"role": "system", "content": "..."},
{"role": "user", "content": "..."}
])
cost = estimate_cost(tokens, "gpt-4", input_tokens=True)
```
---
## Supported Models
300+ models across 34+ providers: OpenAI, Anthropic, Google, Meta, Mistral, Cohere, xAI, DeepSeek, etc. Use `python -m scripts.cli -l` for full list.
- **OpenAI:** exact via tiktoken
- **Others:** ~85–95% approximation
---
## Common Issues
| Issue | Action |
|-------|--------|
| "tiktoken is required" | `pip install tiktoken` |
| UnsupportedModelError | Use `-l` for valid names |
| Cost "NA" | Model has no pricing; count still valid |
| User provides URL | **Agent must warn:** outbound request, SSRF risk, only trusted URLs; confirm before `-u` |
| Model unclear / vague | **Agent must ask:** user to specify exact model name; offer `-l` to list; do not guess |
---
## When to Trigger This Skill
Activate this skill when the user:
| Trigger | Example phrases |
|---------|-----------------|
| **Token count** | "How many tokens?", "Count tokens in this prompt", "Token length of X" |
| **Cost estimate** | "Estimate API cost", "How much for this text?", "Cost for GPT-4" |
| **Prompt size** | "Check prompt length", "Is this too long?", "Context window limit" |
| **OpenClaw audit** | "How many tokens does my workspace use?", "Audit OpenClaw memory/persona/skills", "Which components consume tokens?", "Token usage of AGENTS.md / SOUL.md / skills" |
| **Model comparison** | "Compare token cost across models", "Which model is cheaper?" |
Also trigger when the agent needs to count tokens or estimate cost before/after generating content.
---
## Quick Reference
| Item | Command |
|------|---------|
| Invoke | `python -m scripts.cli` |
| List models | `python -m scripts.cli -l` |
| Cost | `-c` (input) / `--output-tokens` (output) |
| Currency | `--currency USD` or `INR` |
"""
# 待测试的模型列表（可按需修改）
MODELS = [
"anthropic/claude-sonnet-4-6",
"anthropic/claude-sonnet-4-5",
"anthropic/claude-opus-4.6",
"openai/gpt-5.2-codex",
"google/gemini-3.1-pro-preview",
"z-ai/glm-5",
"volcengine/doubao-seed-2-0-pro",
"moonshot/kimi-k2.5",
"minimax/MiniMax-M2.5",
"deepseek-v3.2",
]
# API 配置（与 skill_executor 保持一致，API 模式时使用）
API_KEY = ""
BASE_URL = ""
def get_token_count_local(model: str, text: str) -> dict:
"""使用本项目的 TokenCounter 本地近似计算 token 数。"""
try:
from scripts.core import TokenCounter
counter = TokenCounter(model)
token_count = counter.count(text)
except Exception as e:
return {
"model": model,
"success": False,
"error": str(e),
"char_count": len(text),
"token_count": None,
"ratio": None,
}
char_count = len(text)
ratio = char_count / token_count if token_count > 0 else 0
return {
"model": model,
"success": True,
"char_count": char_count,
"token_count": token_count,
"ratio": round(ratio, 4),
"ratio_desc": f"1 token ≈ {ratio:.4f} 字符",
}
async def get_token_count_api(model: str, text: str) -> dict:
"""
请求模型 API，获取输入文本的 token 数。
使用 chat completion，从 response.usage.prompt_tokens 获取。
依赖: pip install openai
"""
try:
import openai
client = openai.AsyncOpenAI(api_key=API_KEY, base_url=BASE_URL)
messages = [{"role": "user", "content": text}]
response = await client.chat.completions.create(
model=model,
messages=messages,
max_tokens=8 * 1024, # 部分 API 要求 >= 16，只关心 prompt_tokens
temperature=0,
)
usage = getattr(response, "usage", None)
if usage is None:
return {
"model": model,
"success": False,
"error": "API 响应中无 usage 字段",
"char_count": len(text),
"token_count": None,
"ratio": None,
}
prompt_tokens = getattr(usage, "prompt_tokens", None)
if prompt_tokens is None:
prompt_tokens = getattr(usage, "total_tokens", None)
if prompt_tokens is None:
return {
"model": model,
"success": False,
"error": "usage 中无 prompt_tokens/total_tokens",
"char_count": len(text),
"token_count": None,
"ratio": None,
}
char_count = len(text)
ratio = char_count / prompt_tokens if prompt_tokens > 0 else 0
return {
"model": model,
"success": True,
"char_count": char_count,
"token_count": prompt_tokens,
"ratio": round(ratio, 4),
"ratio_desc": f"1 token ≈ {ratio:.4f} 字符",
}
except Exception as e:
return {
"model": model,
"success": False,
"error": str(e),
"char_count": len(text),
"token_count": None,
"ratio": None,
}
async def batch_test_api(models: list[str], text: str) -> list[dict]:
"""批量异步请求所有模型（API 模式）。"""
tasks = [get_token_count_api(m, text) for m in models]
results = await asyncio.gather(*tasks, return_exceptions=True)
out = []
for i, r in enumerate(results):
if isinstance(r, Exception):
out.append(
{
"model": models[i],
"success": False,
"error": str(r),
"char_count": len(text),
"token_count": None,
"ratio": None,
}
)
else:
out.append(r)
return out
def batch_test_local(models: list[str], text: str) -> list[dict]:
"""批量测试所有模型（本地模式）。"""
return [get_token_count_local(m, text) for m in models]
def write_report(results: list[dict], text: str, output_path: Path, mode: str) -> None:
"""将测试结果写入 Markdown 文档。"""
mode_label = "API" if mode == "api" else "本地近似"
lines = [
"# 模型 Token 与文字比例测试报告",
"",
f"**生成时间**: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}",
f"**测试字符数**: {len(text)}",
f"**模式**: {mode_label}",
"",
"| 模型 | 字符数 | Token 数 | 1 token ≈ 多少字符 | 状态 |",
"|------|--------|----------|-------------------|------|",
]
for r in results:
model = r.get("model", "-")
char_count = r.get("char_count") or "-"
token_count = r.get("token_count") if r.get("token_count") is not None else "-"
ratio = r.get("ratio")
ratio_str = f"{ratio:.4f}" if isinstance(ratio, (int, float)) else "-"
status = "✓" if r.get("success") else f"✗ {r.get('error', '')}"
lines.append(f"| {model} | {char_count} | {token_count} | {ratio_str} | {status} |")
lines.extend(
[
"",
"## 说明",
"",
"- **1 token ≈ 多少字符**：字符数 / token 数，数值越大表示 1 个 token 能表示更多文字",
"- 中文通常比英文消耗更多 token",
"- 不同模型使用不同 tokenizer，比例会有差异",
"",
]
)
output_path.write_text("\n".join(lines), encoding="utf-8")
print(f"报告已保存: {output_path}")
def main(
text: Optional[str] = None,
text_file: Optional[Path] = None,
models: Optional[list[str]] = None,
output_path: Optional[Path] = None,
use_local: bool = False,
) -> list[dict]:
"""
主流程：批量测试并输出报告。
Args:
text: 测试文本，默认使用 SAMPLE_TEXT
text_file: 从文件读取测试文本（覆盖 text）
models: 模型列表，默认使用 MODELS
output_path: 输出文件路径，默认 token_ratio_report.md
use_local: True 使用本地 TokenCounter，False 使用 API
"""
if text_file and text_file.exists():
text = text_file.read_text(encoding="utf-8")
elif text is None:
text = SAMPLE_TEXT
models = models or MODELS
if output_path is None:
output_path = Path(__file__).parent / "token_ratio_report.md"
mode = "local" if use_local else "api"
if use_local:
print(f"本地模式：测试 {len(models)} 个模型，文本长度 {len(text)} 字符...")
results = batch_test_local(models, text)
else:
try:
import openai # noqa: F401
except ImportError:
print("API 模式需要 openai: pip install openai")
print("或使用 --local 进行本地近似测试")
return []
if not API_KEY or not BASE_URL:
print("API 模式需在脚本中设置 API_KEY 和 BASE_URL")
print("或使用 --local 进行本地近似测试")
return []
print(f"API 模式：测试 {len(models)} 个模型，文本长度 {len(text)} 字符...")
results = asyncio.run(batch_test_api(models, text))
for r in results:
if r.get("success"):
print(f" {r['model']}: {r['token_count']} tokens, 1 token ≈ {r['ratio']:.4f} 字符")
else:
print(f" {r['model']}: 失败 - {r.get('error', '')}")
write_report(results, text, output_path, mode)
return results
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="批量测试模型 token 与文字比例")
parser.add_argument("--local", action="store_true", help="使用本地 TokenCounter 近似（无需 API）")
parser.add_argument("--text-file", "-f", type=Path, help="从文件读取测试文本")
parser.add_argument("--output", "-o", type=Path, help="输出报告路径")
args = parser.parse_args()
# 确保从项目根目录运行时可导入 scripts
script_dir = Path(__file__).resolve().parent
project_root = script_dir.parent.parent
if str(project_root) not in sys.path:
sys.path.insert(0, str(project_root))
main(text_file=args.text_file, output_path=args.output, use_local=args.local)
FILE:scripts/examples/count_prompt.py
#!/usr/bin/env python3
"""
Count tokens for local file(s). Default: read from local files, batch mode.
Run from project root:
python scripts/examples/count_prompt.py file1.txt file2.txt -m gpt-4
python scripts/examples/count_prompt.py -f input.txt -m gpt-4
"""
import argparse
import sys
from pathlib import Path
# Ensure project root is on path
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from scripts.core import TokenCounter
from scripts.exceptions import UnsupportedModelError, TokenizationError
def main() -> None:
parser = argparse.ArgumentParser(
description="Count tokens for local file(s), batch mode"
)
parser.add_argument(
"paths",
nargs="*",
metavar="FILE",
help="Local file path(s) to count (default input)",
)
parser.add_argument(
"-f", "--file",
action="append",
dest="files",
help="Read from file (repeatable)",
)
parser.add_argument(
"-m", "--model",
default="gpt-4",
help="Model name (default: gpt-4)",
)
args = parser.parse_args()
paths = list(args.paths or [])
if args.files:
paths.extend(args.files)
if not paths:
parser.error("Provide at least one local file path")
model = args.model
print(f"Model: {model}")
print("---")
try:
counter = TokenCounter(model)
for p in paths:
path = Path(p)
if not path.exists():
print(f"Error: File '{path}' not found", file=sys.stderr)
sys.exit(1)
text = path.read_text(encoding="utf-8")
tokens = counter.count(text)
print(f"{p}: tokens={tokens}")
except UnsupportedModelError as e:
print(f"Error: {e}", file=sys.stderr)
sys.exit(1)
except TokenizationError as e:
print(f"Error: {e}", file=sys.stderr)
sys.exit(1)
if __name__ == "__main__":
main()
FILE:scripts/examples/estimate_cost.py
#!/usr/bin/env python3
"""
Estimate API cost for a prompt (text or file).
Run from project root: python scripts/examples/estimate_cost.py [text_or_file] [model]
Or: python scripts/examples/estimate_cost.py input.txt gpt-4
"""
import argparse
import sys
from pathlib import Path
# Ensure project root is on path
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from scripts.core import TokenCounter, estimate_cost
from scripts.exceptions import UnsupportedModelError, TokenizationError
def main() -> None:
parser = argparse.ArgumentParser(
description="Estimate API cost for a prompt (text or file)"
)
parser.add_argument(
"input",
nargs="?",
default="Your prompt text here",
help="Text to estimate, or path to file",
)
parser.add_argument(
"model",
nargs="?",
default="gpt-4",
help="Model name (default: gpt-4)",
)
parser.add_argument(
"-m", "--model",
dest="model_opt",
help="Model name (alternative to positional)",
)
args = parser.parse_args()
model = args.model_opt or args.model
if Path(args.input).is_file():
path = Path(args.input)
if not path.exists():
print(f"Error: File '{path}' not found", file=sys.stderr)
sys.exit(1)
text = path.read_text(encoding="utf-8")
print(f"Estimating cost for file: {path}")
else:
text = args.input
print(f'Estimating cost for text: "{text[:50]}{"..." if len(text) > 50 else ""}"')
try:
tokens = TokenCounter(model).count(text)
cost = estimate_cost(tokens, model, input_tokens=True)
symbol = "$"
print(f"tokens={tokens} cost={symbol}{cost:.6f}")
except UnsupportedModelError as e:
print(f"Error: {e}", file=sys.stderr)
sys.exit(1)
except TokenizationError as e:
print(f"Error: {e}", file=sys.stderr)
sys.exit(1)
if __name__ == "__main__":
main()
FILE:scripts/exceptions.py
"""
Custom exceptions for the scripts library.
This module defines a hierarchy of custom exceptions that provide detailed error
information for various failure modes in token counting operations. The exceptions
are designed to give users clear, actionable error messages with relevant context.
Exception Hierarchy:
.. code-block:: text
ToksumError (base)
├── UnsupportedModelError
├── ModelNotFoundError
├── TokenizationError
│ ├── InvalidTokenError
│ └── EmptyTextError
The exception hierarchy allows for both specific error handling and general
error catching at different levels of granularity.
Usage Examples:
Specific exception handling:
.. code-block:: python
from scripts import TokenCounter
from scripts.exceptions import UnsupportedModelError, TokenizationError
try:
counter = TokenCounter("unknown-model")
tokens = counter.count("Hello, world!")
except UnsupportedModelError as e:
print(f"Model not supported: {e.model}")
print(f"Available models: {e.supported_models}")
except TokenizationError as e:
print(f"Tokenization failed: {e}")
if e.model:
print(f"Model: {e.model}")
if e.text_preview:
print(f"Text preview: {e.text_preview}")
General error handling:
.. code-block:: python
from scripts import count_tokens
from scripts.exceptions import ToksumError
try:
tokens = count_tokens("Hello!", "gpt-4")
except ToksumError as e:
print(f"Toksum error: {e}")
except Exception as e:
print(f"Unexpected error: {e}")
Error Context:
All exceptions include relevant context information:
- **Model information**: Which model caused the error
- **Text previews**: Sample of problematic text (truncated for privacy)
- **Supported alternatives**: List of valid options when applicable
- **Detailed messages**: Clear descriptions of what went wrong
The exceptions are designed to be both human-readable and machine-parseable,
making them suitable for both debugging and automated error handling.
"""
from typing import List, Optional
class ToksumError(Exception):
"""
Base exception class for all scripts library errors.
This is the root exception class that all other scripts exceptions inherit from.
It allows for catching any scripts-related error with a single exception type
while still providing the ability to catch specific error types when needed.
This exception should generally not be raised directly. Instead, use one of
the more specific exception subclasses that provide additional context and
error-specific information.
Examples:
Catching all scripts errors:
.. code-block:: python
from scripts import count_tokens
from scripts.exceptions import ToksumError
try:
tokens = count_tokens("Hello!", "some-model")
except ToksumError as e:
print(f"Toksum error occurred: {e}")
# Handle any scripts-related error
except Exception as e:
print(f"Unexpected error: {e}")
# Handle non-scripts errors
Catching specific errors with fallback:
.. code-block:: python
from scripts import TokenCounter
from scripts.exceptions import UnsupportedModelError, TokenizationError, ToksumError
try:
counter = TokenCounter("gpt-4")
tokens = counter.count("Hello!")
except UnsupportedModelError:
print("Model not supported")
except TokenizationError:
print("Tokenization failed")
except ToksumError:
print("Other scripts error")
except Exception:
print("Non-scripts error")
Note:
This base class provides a common interface for all scripts exceptions
but does not add any additional functionality beyond the standard
Python Exception class.
"""
class UnsupportedModelError(ToksumError):
"""
Raised when an unsupported model is specified.
This exception is raised when attempting to create a TokenCounter or use
count_tokens() with a model name that is not recognized by the library.
The exception includes the invalid model name and optionally a list of
supported models for reference.
Attributes:
model (str): The unsupported model name that was specified
supported_models (List[str]): List of supported model names (optional)
The exception message automatically includes the unsupported model name
and, if provided, a formatted list of supported alternatives.
Examples:
Basic usage with model validation:
.. code-block:: python
from scripts import TokenCounter
from scripts.exceptions import UnsupportedModelError
try:
counter = TokenCounter("invalid-model-name")
except UnsupportedModelError as e:
print(f"Model '{e.model}' is not supported")
if e.supported_models:
print("Supported models include:")
for model in e.supported_models[:5]: # Show first 5
print(f" - {model}")
Handling with model suggestions:
.. code-block:: python
from scripts import get_supported_models
from scripts.exceptions import UnsupportedModelError
def safe_create_counter(model_name):
try:
return TokenCounter(model_name)
except UnsupportedModelError as e:
print(f"Error: {e}")
# Suggest similar models
all_models = get_supported_models()
suggestions = []
for provider_models in all_models.values():
for model in provider_models:
if model_name.lower() in model.lower():
suggestions.append(model)
if suggestions:
print("Did you mean one of these?")
for suggestion in suggestions[:3]:
print(f" - {suggestion}")
return None
Programmatic error handling:
.. code-block:: python
def validate_model(model_name):
try:
TokenCounter(model_name)
return True
except UnsupportedModelError:
return False
# Test multiple models
test_models = ["gpt-4", "invalid-model", "claude-3-opus"]
for model in test_models:
if validate_model(model):
print(f"✓ {model} is supported")
else:
print(f"✗ {model} is not supported")
Common Causes:
- **Typos in model names**: "gpt4" instead of "gpt-4"
- **Incorrect casing**: "GPT-4" vs "gpt-4" (though scripts is case-insensitive)
- **Outdated model names**: Using deprecated or renamed models
- **Provider confusion**: Using model names from unsupported providers
- **Version specificity**: Using overly specific version numbers
Resolution:
1. **Check spelling**: Verify the model name is spelled correctly
2. **Use get_supported_models()**: Get the complete list of supported models
3. **Check provider**: Ensure the model provider is supported
4. **Update library**: Newer models may require a library update
See Also:
- :func:`get_supported_models`: Get all supported models by provider
- :class:`TokenCounter`: Main class that raises this exception
- :func:`count_tokens`: Convenience function that may raise this exception
"""
def __init__(self, model: str, supported_models: Optional[List[str]] = None):
"""
Initialize UnsupportedModelError with model name and optional supported models list.
Args:
model (str): The unsupported model name that was specified
supported_models (Optional[List[str]]): List of supported model names
for user reference. If provided,
will be included in error message.
"""
self.model = model
self.supported_models = supported_models or []
if self.supported_models:
message = f"Model '{model}' is not supported. Supported models: {', '.join(self.supported_models)}"
else:
message = f"Model '{model}' is not supported."
super().__init__(message)
class ModelNotFoundError(ToksumError):
"""
Raised when a specified model is not found.
This exception is raised in scenarios where a model name is recognized
as potentially valid but cannot be located or accessed. This is distinct
from UnsupportedModelError, which is raised for completely unrecognized
model names.
This exception is typically used for cases where:
- A model exists but is temporarily unavailable
- A model requires special access or authentication
- A model has been deprecated or removed
- There are network or service issues preventing model access
Attributes:
model (str): The model name that could not be found
Examples:
Basic error handling:
.. code-block:: python
from scripts.exceptions import ModelNotFoundError
try:
# Some operation that might fail to find a model
result = some_model_operation("gpt-4")
except ModelNotFoundError as e:
print(f"Model not found: {e}")
# Maybe try a fallback model or retry later
Distinguishing from UnsupportedModelError:
.. code-block:: python
from scripts import TokenCounter
from scripts.exceptions import UnsupportedModelError, ModelNotFoundError
try:
counter = TokenCounter("some-model")
except UnsupportedModelError:
print("Model is not supported by scripts")
except ModelNotFoundError:
print("Model is supported but cannot be found/accessed")
Note:
This exception is less commonly used in the current scripts implementation
but is available for future extensions and specific use cases where
model availability needs to be distinguished from model support.
See Also:
- :exc:`UnsupportedModelError`: For completely unsupported models
- :class:`TokenCounter`: May raise this exception in certain scenarios
"""
def __init__(self, model: str):
"""
Initialize ModelNotFoundError with the model name.
Args:
model (str): The model name that could not be found
"""
super().__init__(f"Model '{model}' not found.")
class TokenizationError(ToksumError):
"""
Raised when tokenization fails for any reason.
This is the primary exception for tokenization-related failures. It provides
detailed context about what went wrong, including the model being used and
a preview of the problematic text (truncated for privacy and readability).
This exception covers a wide range of tokenization failures:
- Invalid input types (non-string input)
- Missing dependencies (e.g., tiktoken for OpenAI models)
- Tokenizer initialization failures
- Text processing errors
- Encoding/decoding issues
Attributes:
model (Optional[str]): The model name being used when the error occurred
text_preview (Optional[str]): A preview of the problematic text (truncated)
The exception message includes the base error description and automatically
appends model and text preview information when available.
Examples:
Basic error handling:
.. code-block:: python
from scripts import TokenCounter
from scripts.exceptions import TokenizationError
try:
counter = TokenCounter("gpt-4")
tokens = counter.count(123) # Invalid input type
except TokenizationError as e:
print(f"Tokenization failed: {e}")
if e.model:
print(f"Model: {e.model}")
if e.text_preview:
print(f"Text preview: {e.text_preview}")
Handling missing dependencies:
.. code-block:: python
try:
counter = TokenCounter("gpt-4") # Requires tiktoken
tokens = counter.count("Hello!")
except TokenizationError as e:
if "tiktoken" in str(e):
print("Please install tiktoken: pip install tiktoken")
else:
print(f"Tokenization error: {e}")
Robust error handling with fallbacks:
.. code-block:: python
def safe_count_tokens(text, model, fallback_model=None):
try:
counter = TokenCounter(model)
return counter.count(text)
except TokenizationError as e:
print(f"Primary model failed: {e}")
if fallback_model:
try:
print(f"Trying fallback model: {fallback_model}")
counter = TokenCounter(fallback_model)
return counter.count(text)
except TokenizationError:
print("Fallback model also failed")
return None
Input validation with detailed errors:
.. code-block:: python
def validate_and_count(text, model):
try:
if not isinstance(text, str):
raise TokenizationError(
f"Input must be string, got {type(text).__name__}",
model=model
)
counter = TokenCounter(model)
return counter.count(text)
except TokenizationError as e:
print(f"Validation failed: {e}")
return None
Text Preview Handling:
The text_preview attribute contains a truncated version of the problematic
text to help with debugging while protecting privacy:
- **Truncation**: Long text is truncated to ~50 characters + "..."
- **Privacy**: Sensitive information is not fully exposed in error messages
- **Debugging**: Provides enough context to identify the problematic content
Common Causes:
- **Wrong input type**: Passing int, list, dict instead of string
- **None input**: Passing None instead of a string
- **Missing dependencies**: tiktoken not installed for OpenAI models
- **Encoding issues**: Text with problematic character encodings
- **Memory issues**: Extremely large text causing processing failures
Resolution:
1. **Check input type**: Ensure text is a string
2. **Install dependencies**: Install tiktoken for OpenAI models
3. **Validate text**: Check for encoding issues or special characters
4. **Check text size**: Very large texts may cause issues
5. **Try different model**: Some models may handle edge cases better
See Also:
- :exc:`InvalidTokenError`: For specific token-related errors
- :exc:`EmptyTextError`: For empty text specific errors
- :class:`TokenCounter`: Main class that raises this exception
"""
def __init__(self, message: str, model: Optional[str] = None, text_preview: Optional[str] = None):
"""
Initialize TokenizationError with detailed context.
Args:
message (str): The base error message describing what went wrong
model (Optional[str]): The model name being used when error occurred
text_preview (Optional[str]): Preview of the problematic text.
Will be truncated if longer than 50 characters.
"""
self.model = model
self.text_preview = text_preview
full_message = f"Tokenization failed: {message}"
if model:
full_message += f" (model: {model})"
if text_preview:
preview = text_preview[:50] + "..." if len(text_preview) > 50 else text_preview
full_message += f" (text preview: '{preview}')"
super().__init__(full_message)
class InvalidTokenError(TokenizationError):
"""
Raised when an invalid token is encountered during tokenization.
This exception is a specialized form of TokenizationError that specifically
handles cases where individual tokens are invalid or problematic. It provides
additional context about the specific token that caused the issue.
This exception might be raised when:
- A token contains invalid characters or sequences
- A token exceeds maximum length limits
- A token has encoding issues
- A token conflicts with model-specific restrictions
Attributes:
token (str): The specific invalid token that caused the error
model (Optional[str]): The model name (inherited from TokenizationError)
text_preview (Optional[str]): Preview of the text (inherited from TokenizationError)
Examples:
Handling invalid token errors:
.. code-block:: python
from scripts import TokenCounter
from scripts.exceptions import InvalidTokenError, TokenizationError
try:
counter = TokenCounter("gpt-4")
tokens = counter.count("Some text with problematic content")
except InvalidTokenError as e:
print(f"Invalid token encountered: '{e.token}'")
print(f"Error details: {e}")
# Maybe try preprocessing the text to remove problematic tokens
except TokenizationError as e:
print(f"General tokenization error: {e}")
Token-specific error handling:
.. code-block:: python
def safe_tokenize_with_cleanup(text, model):
try:
counter = TokenCounter(model)
return counter.count(text)
except InvalidTokenError as e:
print(f"Removing problematic token: {e.token}")
# Remove or replace the problematic token
cleaned_text = text.replace(e.token, "")
return counter.count(cleaned_text)
except TokenizationError:
print("Could not tokenize even after cleanup")
return None
Note:
This exception is not commonly raised in the current scripts implementation
but is available for future enhancements and edge cases where specific
token validation is needed.
See Also:
- :exc:`TokenizationError`: Parent class for general tokenization errors
- :exc:`EmptyTextError`: For empty text specific cases
- :class:`TokenCounter`: May raise this exception in specific scenarios
"""
def __init__(self, token: str, message: str, model: Optional[str] = None, text_preview: Optional[str] = None):
"""
Initialize InvalidTokenError with token and context information.
Args:
token (str): The specific invalid token that caused the error
message (str): Description of why the token is invalid
model (Optional[str]): The model name being used
text_preview (Optional[str]): Preview of the problematic text
"""
full_message = f"Invalid token '{token}': {message}"
super().__init__(full_message, model, text_preview)
class EmptyTextError(TokenizationError):
"""
Raised when attempting to tokenize empty text in contexts where it's not allowed.
This exception is a specialized form of TokenizationError for cases where
empty text is specifically problematic. Note that in most scripts operations,
empty text is handled gracefully and returns 0 tokens. This exception is
reserved for contexts where empty text indicates a logical error.
This exception might be raised when:
- Empty text is passed to functions that require non-empty content
- Batch operations encounter unexpected empty strings
- Validation functions detect empty content where it shouldn't be
- Message content is empty in chat format validation
Attributes:
model (Optional[str]): The model name (inherited from TokenizationError)
Examples:
Handling empty text validation:
.. code-block:: python
from scripts.exceptions import EmptyTextError, TokenizationError
def validate_content(text, model):
try:
if not text.strip(): # Check for empty or whitespace-only
raise EmptyTextError(model=model)
counter = TokenCounter(model)
return counter.count(text)
except EmptyTextError as e:
print("Error: Content cannot be empty")
return None
except TokenizationError as e:
print(f"Tokenization error: {e}")
return None
Batch processing with empty text handling:
.. code-block:: python
def process_text_batch(texts, model):
results = []
counter = TokenCounter(model)
for i, text in enumerate(texts):
try:
if not text:
raise EmptyTextError(model=model)
tokens = counter.count(text)
results.append(tokens)
except EmptyTextError:
print(f"Skipping empty text at index {i}")
results.append(0) # or None, depending on requirements
except TokenizationError as e:
print(f"Error processing text {i}: {e}")
results.append(None)
return results
Message validation:
.. code-block:: python
def validate_messages(messages):
for i, msg in enumerate(messages):
if not msg.get("content", "").strip():
raise EmptyTextError(
f"Message {i} has empty content"
)
Normal Empty Text Handling:
In most scripts operations, empty text is handled normally:
.. code-block:: python
counter = TokenCounter("gpt-4")
tokens = counter.count("") # Returns 0, no exception
messages = [
{"role": "user", "content": ""}, # This might raise EmptyTextError
{"role": "assistant", "content": "Hello!"}
]
Use Cases:
- **Content validation**: Ensuring required fields are not empty
- **Batch processing**: Identifying problematic entries in datasets
- **API validation**: Checking inputs before expensive operations
- **Data quality**: Ensuring content meets minimum requirements
Note:
This exception is used sparingly in scripts. Most empty text cases
are handled gracefully by returning 0 tokens. Use this exception
only when empty text represents a logical error in your application.
See Also:
- :exc:`TokenizationError`: Parent class for general tokenization errors
- :exc:`InvalidTokenError`: For specific token validation issues
- :meth:`TokenCounter.count`: Handles empty text gracefully (returns 0)
- :meth:`TokenCounter.count_messages`: May raise this for empty message content
"""
def __init__(self, model: Optional[str] = None):
"""
Initialize EmptyTextError with optional model context.
Args:
model (Optional[str]): The model name being used when the error occurred
"""
super().__init__("Cannot tokenize empty text.", model)
FILE:scripts/registry/models.py
"""Model registry for scripts.
Defines a canonical model registry with metadata needed for provider detection,
OpenAI encoding selection, approximate token formulas, and pricing lookup.
"""
from __future__ import annotations
import difflib
import re
from dataclasses import dataclass
from typing import Dict, Iterable, List, Optional
@dataclass(frozen=True)
class ModelInfo:
name: str
provider: str
encoding: Optional[str] = None
formula: Optional[str] = None
pricing_id: Optional[str] = None
aliases: Optional[List[str]] = None
def normalize_model_name(model: str) -> str:
return model.strip().lower()
MODELS: Dict[str, ModelInfo] = {
# OpenAI
"gpt-4": ModelInfo("gpt-4", "openai", encoding="cl100k_base", pricing_id="gpt-4"),
"gpt-4-0314": ModelInfo("gpt-4-0314", "openai", encoding="cl100k_base", pricing_id="gpt-4"),
"gpt-4-0613": ModelInfo("gpt-4-0613", "openai", encoding="cl100k_base", pricing_id="gpt-4"),
"gpt-4-32k": ModelInfo("gpt-4-32k", "openai", encoding="cl100k_base", pricing_id="gpt-4-32k"),
"gpt-4-32k-0314": ModelInfo("gpt-4-32k-0314", "openai", encoding="cl100k_base", pricing_id="gpt-4-32k"),
"gpt-4-32k-0613": ModelInfo("gpt-4-32k-0613", "openai", encoding="cl100k_base", pricing_id="gpt-4-32k"),
"gpt-4-1106-preview": ModelInfo("gpt-4-1106-preview", "openai", encoding="cl100k_base"),
"gpt-4-0125-preview": ModelInfo("gpt-4-0125-preview", "openai", encoding="cl100k_base"),
"gpt-4-turbo-preview": ModelInfo("gpt-4-turbo-preview", "openai", encoding="cl100k_base"),
"gpt-4-vision-preview": ModelInfo("gpt-4-vision-preview", "openai", encoding="cl100k_base"),
"gpt-4-turbo": ModelInfo("gpt-4-turbo", "openai", encoding="cl100k_base", pricing_id="gpt-4-turbo"),
"gpt-4-turbo-2024-04-09": ModelInfo("gpt-4-turbo-2024-04-09", "openai", encoding="cl100k_base", pricing_id="gpt-4-turbo"),
"gpt-4o": ModelInfo("gpt-4o", "openai", encoding="cl100k_base", pricing_id="gpt-4o"),
"gpt-4o-2024-05-13": ModelInfo("gpt-4o-2024-05-13", "openai", encoding="cl100k_base", pricing_id="gpt-4o"),
"gpt-4o-mini": ModelInfo("gpt-4o-mini", "openai", encoding="cl100k_base", pricing_id="gpt-4o-mini"),
"gpt-4o-mini-2024-07-18": ModelInfo("gpt-4o-mini-2024-07-18", "openai", encoding="cl100k_base", pricing_id="gpt-4o-mini"),
"gpt-4o-2024-08-06": ModelInfo("gpt-4o-2024-08-06", "openai", encoding="cl100k_base", pricing_id="gpt-4o"),
"gpt-4o-2024-11-20": ModelInfo("gpt-4o-2024-11-20", "openai", encoding="cl100k_base", pricing_id="gpt-4o"),
"gpt-4-1106-vision-preview": ModelInfo("gpt-4-1106-vision-preview", "openai", encoding="cl100k_base"),
"gpt-3.5-turbo": ModelInfo("gpt-3.5-turbo", "openai", encoding="cl100k_base", pricing_id="gpt-3.5-turbo"),
"gpt-3.5-turbo-0301": ModelInfo("gpt-3.5-turbo-0301", "openai", encoding="cl100k_base", pricing_id="gpt-3.5-turbo"),
"gpt-3.5-turbo-0613": ModelInfo("gpt-3.5-turbo-0613", "openai", encoding="cl100k_base", pricing_id="gpt-3.5-turbo"),
"gpt-3.5-turbo-1106": ModelInfo("gpt-3.5-turbo-1106", "openai", encoding="cl100k_base", pricing_id="gpt-3.5-turbo"),
"gpt-3.5-turbo-0125": ModelInfo("gpt-3.5-turbo-0125", "openai", encoding="cl100k_base", pricing_id="gpt-3.5-turbo"),
"gpt-3.5-turbo-16k": ModelInfo("gpt-3.5-turbo-16k", "openai", encoding="cl100k_base", pricing_id="gpt-3.5-turbo-16k"),
"gpt-3.5-turbo-16k-0613": ModelInfo("gpt-3.5-turbo-16k-0613", "openai", encoding="cl100k_base", pricing_id="gpt-3.5-turbo-16k"),
"gpt-3.5-turbo-instruct": ModelInfo("gpt-3.5-turbo-instruct", "openai", encoding="cl100k_base"),
"text-davinci-003": ModelInfo("text-davinci-003", "openai", encoding="p50k_base"),
"text-davinci-002": ModelInfo("text-davinci-002", "openai", encoding="p50k_base"),
"text-curie-001": ModelInfo("text-curie-001", "openai", encoding="r50k_base"),
"text-babbage-001": ModelInfo("text-babbage-001", "openai", encoding="r50k_base"),
"text-ada-001": ModelInfo("text-ada-001", "openai", encoding="r50k_base"),
"davinci": ModelInfo("davinci", "openai", encoding="r50k_base"),
"curie": ModelInfo("curie", "openai", encoding="r50k_base"),
"babbage": ModelInfo("babbage", "openai", encoding="r50k_base"),
"ada": ModelInfo("ada", "openai", encoding="r50k_base"),
"gpt-3": ModelInfo("gpt-3", "openai", encoding="r50k_base"),
"text-embedding-ada-002": ModelInfo("text-embedding-ada-002", "openai", encoding="cl100k_base"),
"text-embedding-3-small": ModelInfo("text-embedding-3-small", "openai", encoding="cl100k_base"),
"text-embedding-3-large": ModelInfo("text-embedding-3-large", "openai", encoding="cl100k_base"),
"gpt-4-base": ModelInfo("gpt-4-base", "openai", encoding="cl100k_base"),
"gpt-3.5-turbo-instruct-0914": ModelInfo("gpt-3.5-turbo-instruct-0914", "openai", encoding="cl100k_base"),
"o1-preview": ModelInfo("o1-preview", "openai", encoding="cl100k_base"),
"o1-mini": ModelInfo("o1-mini", "openai", encoding="cl100k_base"),
"o1-preview-2024-09-12": ModelInfo("o1-preview-2024-09-12", "openai", encoding="cl100k_base"),
"o1-mini-2024-09-12": ModelInfo("o1-mini-2024-09-12", "openai", encoding="cl100k_base"),
"gpt-5.2-codex": ModelInfo("gpt-5.2-codex", "openai", encoding="cl100k_base"),
"gpt-4-vision": ModelInfo("gpt-4-vision", "openai", encoding="cl100k_base"),
"gpt-4-vision-preview-0409": ModelInfo("gpt-4-vision-preview-0409", "openai", encoding="cl100k_base"),
"gpt-4-vision-preview-1106": ModelInfo("gpt-4-vision-preview-1106", "openai", encoding="cl100k_base"),
"text-similarity-ada-001": ModelInfo("text-similarity-ada-001", "openai", encoding="r50k_base"),
"text-similarity-babbage-001": ModelInfo("text-similarity-babbage-001", "openai", encoding="r50k_base"),
"text-similarity-curie-001": ModelInfo("text-similarity-curie-001", "openai", encoding="r50k_base"),
"text-similarity-davinci-001": ModelInfo("text-similarity-davinci-001", "openai", encoding="r50k_base"),
# Anthropic
"claude-3-opus-20240229": ModelInfo("claude-3-opus-20240229", "anthropic", formula="anthropic", pricing_id="claude-3-opus-20240229"),
"claude-3-sonnet-20240229": ModelInfo("claude-3-sonnet-20240229", "anthropic", formula="anthropic", pricing_id="claude-3-sonnet-20240229"),
"claude-3-haiku-20240307": ModelInfo("claude-3-haiku-20240307", "anthropic", formula="anthropic", pricing_id="claude-3-haiku-20240307"),
"claude-3.5-sonnet-20240620": ModelInfo("claude-3.5-sonnet-20240620", "anthropic", formula="anthropic", pricing_id="claude-3.5-sonnet-20240620"),
"claude-3.5-sonnet-20241022": ModelInfo("claude-3.5-sonnet-20241022", "anthropic", formula="anthropic", pricing_id="claude-3.5-sonnet-20241022"),
"claude-3.5-haiku-20241022": ModelInfo("claude-3.5-haiku-20241022", "anthropic", formula="anthropic", pricing_id="claude-3.5-haiku-20241022"),
"claude-3-5-sonnet-20240620": ModelInfo("claude-3-5-sonnet-20240620", "anthropic", formula="anthropic", pricing_id="claude-3.5-sonnet-20240620"),
"claude-3-opus": ModelInfo("claude-3-opus", "anthropic", formula="anthropic", pricing_id="claude-3-opus-20240229"),
"claude-3-sonnet": ModelInfo("claude-3-sonnet", "anthropic", formula="anthropic", pricing_id="claude-3-sonnet-20240229"),
"claude-3-haiku": ModelInfo("claude-3-haiku", "anthropic", formula="anthropic", pricing_id="claude-3-haiku-20240307"),
"claude-2.1": ModelInfo("claude-2.1", "anthropic", formula="anthropic"),
"claude-2.0": ModelInfo("claude-2.0", "anthropic", formula="anthropic"),
"claude-instant-1.2": ModelInfo("claude-instant-1.2", "anthropic", formula="anthropic"),
"claude-instant-1.1": ModelInfo("claude-instant-1.1", "anthropic", formula="anthropic"),
"claude-instant-1.0": ModelInfo("claude-instant-1.0", "anthropic", formula="anthropic"),
"claude-instant": ModelInfo("claude-instant", "anthropic", formula="anthropic"),
"claude-1": ModelInfo("claude-1", "anthropic", formula="anthropic"),
"claude-1.3": ModelInfo("claude-1.3", "anthropic", formula="anthropic"),
"claude-1.3-100k": ModelInfo("claude-1.3-100k", "anthropic", formula="anthropic"),
"claude-3-5-haiku-20241022": ModelInfo("claude-3-5-haiku-20241022", "anthropic", formula="anthropic", pricing_id="claude-3.5-haiku-20241022"),
"claude-3-5-sonnet-20241022": ModelInfo("claude-3-5-sonnet-20241022", "anthropic", formula="anthropic"),
"claude-3.5-sonnet-computer-use": ModelInfo("claude-3.5-sonnet-computer-use", "anthropic", formula="anthropic"),
"claude-2.1-200k": ModelInfo("claude-2.1-200k", "anthropic", formula="anthropic"),
"claude-2.1-100k": ModelInfo("claude-2.1-100k", "anthropic", formula="anthropic"),
"claude-instant-2": ModelInfo("claude-instant-2", "anthropic", formula="anthropic"),
"claude-instant-2.0": ModelInfo("claude-instant-2.0", "anthropic", formula="anthropic"),
"claude-3-opus-latest": ModelInfo("claude-3-opus-latest", "anthropic", formula="anthropic", pricing_id="claude-3-opus-20240229"),
"claude-3-sonnet-latest": ModelInfo("claude-3-sonnet-latest", "anthropic", formula="anthropic", pricing_id="claude-3-sonnet-20240229"),
"claude-sonnet-4-6": ModelInfo("claude-sonnet-4-6", "anthropic", formula="anthropic"),
"claude-sonnet-4-5": ModelInfo("claude-sonnet-4-5", "anthropic", formula="anthropic"),
"claude-opus-4.6": ModelInfo("claude-opus-4.6", "anthropic", formula="anthropic"),
# Google
"gemini-pro": ModelInfo("gemini-pro", "google", formula="google"),
"gemini-pro-vision": ModelInfo("gemini-pro-vision", "google", formula="google"),
"gemini-1.5-pro": ModelInfo("gemini-1.5-pro", "google", formula="google"),
"gemini-1.5-flash": ModelInfo("gemini-1.5-flash", "google", formula="google"),
"gemini-1.5-pro-latest": ModelInfo("gemini-1.5-pro-latest", "google", formula="google"),
"gemini-1.5-flash-latest": ModelInfo("gemini-1.5-flash-latest", "google", formula="google"),
"gemini-1.0-pro": ModelInfo("gemini-1.0-pro", "google", formula="google"),
"gemini-1.0-pro-vision": ModelInfo("gemini-1.0-pro-vision", "google", formula="google"),
"gemini-ultra": ModelInfo("gemini-ultra", "google", formula="google"),
"gemini-2.0-flash-exp": ModelInfo("gemini-2.0-flash-exp", "google", formula="google"),
"gemini-2.0-flash": ModelInfo("gemini-2.0-flash", "google", formula="google"),
"gemini-exp-1206": ModelInfo("gemini-exp-1206", "google", formula="google"),
"gemini-exp-1121": ModelInfo("gemini-exp-1121", "google", formula="google"),
"palm-2": ModelInfo("palm-2", "google", formula="google"),
"palm-2-chat": ModelInfo("palm-2-chat", "google", formula="google"),
"palm-2-codechat": ModelInfo("palm-2-codechat", "google", formula="google"),
"gemini-1.0-pro-001": ModelInfo("gemini-1.0-pro-001", "google", formula="google"),
"gemini-1.0-pro-latest": ModelInfo("gemini-1.0-pro-latest", "google", formula="google"),
"gemini-1.0-pro-vision-latest": ModelInfo("gemini-1.0-pro-vision-latest", "google", formula="google"),
"gemini-3.1-pro-preview": ModelInfo("gemini-3.1-pro-preview", "google", formula="google"),
# Meta
"llama-2-7b": ModelInfo("llama-2-7b", "meta", formula="meta"),
"llama-2-13b": ModelInfo("llama-2-13b", "meta", formula="meta"),
"llama-2-70b": ModelInfo("llama-2-70b", "meta", formula="meta"),
"llama-3-8b": ModelInfo("llama-3-8b", "meta", formula="meta"),
"llama-3-70b": ModelInfo("llama-3-70b", "meta", formula="meta"),
"llama-3.1-8b": ModelInfo("llama-3.1-8b", "meta", formula="meta"),
"llama-3.1-70b": ModelInfo("llama-3.1-70b", "meta", formula="meta"),
"llama-3.1-405b": ModelInfo("llama-3.1-405b", "meta", formula="meta"),
"llama-3.2-1b": ModelInfo("llama-3.2-1b", "meta", formula="meta"),
"llama-3.2-3b": ModelInfo("llama-3.2-3b", "meta", formula="meta"),
"llama-3.3-70b": ModelInfo("llama-3.3-70b", "meta", formula="meta"),
"llama-3.3-70b-instruct": ModelInfo("llama-3.3-70b-instruct", "meta", formula="meta"),
"llama-2-7b-chat": ModelInfo("llama-2-7b-chat", "meta", formula="meta"),
"llama-2-13b-chat": ModelInfo("llama-2-13b-chat", "meta", formula="meta"),
"llama-2-70b-chat": ModelInfo("llama-2-70b-chat", "meta", formula="meta"),
"llama-2-7b-chat-hf": ModelInfo("llama-2-7b-chat-hf", "meta", formula="meta"),
"llama-2-13b-chat-hf": ModelInfo("llama-2-13b-chat-hf", "meta", formula="meta"),
"llama-2-70b-chat-hf": ModelInfo("llama-2-70b-chat-hf", "meta", formula="meta"),
"llama-3-8b-instruct": ModelInfo("llama-3-8b-instruct", "meta", formula="meta"),
"llama-3-70b-instruct": ModelInfo("llama-3-70b-instruct", "meta", formula="meta"),
"llama-3.1-8b-instruct": ModelInfo("llama-3.1-8b-instruct", "meta", formula="meta"),
"llama-3.1-70b-instruct": ModelInfo("llama-3.1-70b-instruct", "meta", formula="meta"),
"llama-3.1-405b-instruct": ModelInfo("llama-3.1-405b-instruct", "meta", formula="meta"),
"llama-3.2-1b-instruct": ModelInfo("llama-3.2-1b-instruct", "meta", formula="meta"),
"llama-3.2-3b-instruct": ModelInfo("llama-3.2-3b-instruct", "meta", formula="meta"),
# Mistral
"mistral-7b": ModelInfo("mistral-7b", "mistral", formula="mistral"),
"mistral-8x7b": ModelInfo("mistral-8x7b", "mistral", formula="mistral"),
"mistral-large": ModelInfo("mistral-large", "mistral", formula="mistral"),
"mistral-medium": ModelInfo("mistral-medium", "mistral", formula="mistral"),
"mistral-small": ModelInfo("mistral-small", "mistral", formula="mistral"),
"mistral-tiny": ModelInfo("mistral-tiny", "mistral", formula="mistral"),
"mixtral-8x7b": ModelInfo("mixtral-8x7b", "mistral", formula="mistral"),
"mixtral-8x22b": ModelInfo("mixtral-8x22b", "mistral", formula="mistral"),
"mistral-large-2": ModelInfo("mistral-large-2", "mistral", formula="mistral"),
"mistral-large-2407": ModelInfo("mistral-large-2407", "mistral", formula="mistral"),
"mistral-7b-instruct": ModelInfo("mistral-7b-instruct", "mistral", formula="mistral"),
"mistral-7b-instruct-v0.1": ModelInfo("mistral-7b-instruct-v0.1", "mistral", formula="mistral"),
"mistral-7b-instruct-v0.2": ModelInfo("mistral-7b-instruct-v0.2", "mistral", formula="mistral"),
"mistral-7b-instruct-v0.3": ModelInfo("mistral-7b-instruct-v0.3", "mistral", formula="mistral"),
"mixtral-8x7b-instruct": ModelInfo("mixtral-8x7b-instruct", "mistral", formula="mistral"),
"mixtral-8x22b-instruct": ModelInfo("mixtral-8x22b-instruct", "mistral", formula="mistral"),
# Cohere
"command": ModelInfo("command", "cohere", formula="cohere"),
"command-light": ModelInfo("command-light", "cohere", formula="cohere"),
"command-nightly": ModelInfo("command-nightly", "cohere", formula="cohere"),
"command-r": ModelInfo("command-r", "cohere", formula="cohere"),
"command-r-plus": ModelInfo("command-r-plus", "cohere", formula="cohere"),
"command-r-08-2024": ModelInfo("command-r-08-2024", "cohere", formula="cohere"),
"command-r-plus-08-2024": ModelInfo("command-r-plus-08-2024", "cohere", formula="cohere"),
"command-r-plus-04-2024": ModelInfo("command-r-plus-04-2024", "cohere", formula="cohere"),
# Perplexity
"pplx-7b-online": ModelInfo("pplx-7b-online", "perplexity", formula="perplexity"),
"pplx-70b-online": ModelInfo("pplx-70b-online", "perplexity", formula="perplexity"),
"pplx-7b-chat": ModelInfo("pplx-7b-chat", "perplexity", formula="perplexity"),
"pplx-70b-chat": ModelInfo("pplx-70b-chat", "perplexity", formula="perplexity"),
"codellama-34b-instruct": ModelInfo("codellama-34b-instruct", "perplexity", formula="perplexity"),
# HuggingFace
"microsoft/DialoGPT-medium": ModelInfo("microsoft/DialoGPT-medium", "huggingface", formula="huggingface"),
"microsoft/DialoGPT-large": ModelInfo("microsoft/DialoGPT-large", "huggingface", formula="huggingface"),
"facebook/blenderbot-400M-distill": ModelInfo("facebook/blenderbot-400M-distill", "huggingface", formula="huggingface"),
"facebook/blenderbot-1B-distill": ModelInfo("facebook/blenderbot-1B-distill", "huggingface", formula="huggingface"),
"facebook/blenderbot-3B": ModelInfo("facebook/blenderbot-3B", "huggingface", formula="huggingface"),
# AI21
"j2-light": ModelInfo("j2-light", "ai21", formula="ai21"),
"j2-mid": ModelInfo("j2-mid", "ai21", formula="ai21"),
"j2-ultra": ModelInfo("j2-ultra", "ai21", formula="ai21"),
"j2-jumbo-instruct": ModelInfo("j2-jumbo-instruct", "ai21", formula="ai21"),
# Together
"togethercomputer/RedPajama-INCITE-Chat-3B-v1": ModelInfo("togethercomputer/RedPajama-INCITE-Chat-3B-v1", "together", formula="together"),
"togethercomputer/RedPajama-INCITE-Chat-7B-v1": ModelInfo("togethercomputer/RedPajama-INCITE-Chat-7B-v1", "together", formula="together"),
"NousResearch/Nous-Hermes-Llama2-13b": ModelInfo("NousResearch/Nous-Hermes-Llama2-13b", "together", formula="together"),
# xAI
"grok-1": ModelInfo("grok-1", "xai", formula="xai"),
"grok-1.5": ModelInfo("grok-1.5", "xai", formula="xai"),
"grok-2": ModelInfo("grok-2", "xai", formula="xai"),
"grok-beta": ModelInfo("grok-beta", "xai", formula="xai"),
# Alibaba
"qwen-1.5-0.5b": ModelInfo("qwen-1.5-0.5b", "alibaba", formula="alibaba"),
"qwen-1.5-1.8b": ModelInfo("qwen-1.5-1.8b", "alibaba", formula="alibaba"),
"qwen-1.5-4b": ModelInfo("qwen-1.5-4b", "alibaba", formula="alibaba"),
"qwen-1.5-7b": ModelInfo("qwen-1.5-7b", "alibaba", formula="alibaba"),
"qwen-1.5-14b": ModelInfo("qwen-1.5-14b", "alibaba", formula="alibaba"),
"qwen-1.5-32b": ModelInfo("qwen-1.5-32b", "alibaba", formula="alibaba"),
"qwen-1.5-72b": ModelInfo("qwen-1.5-72b", "alibaba", formula="alibaba"),
"qwen-1.5-110b": ModelInfo("qwen-1.5-110b", "alibaba", formula="alibaba"),
"qwen-2-0.5b": ModelInfo("qwen-2-0.5b", "alibaba", formula="alibaba"),
"qwen-2-1.5b": ModelInfo("qwen-2-1.5b", "alibaba", formula="alibaba"),
"qwen-2-7b": ModelInfo("qwen-2-7b", "alibaba", formula="alibaba"),
"qwen-2-57b": ModelInfo("qwen-2-57b", "alibaba", formula="alibaba"),
"qwen-2-72b": ModelInfo("qwen-2-72b", "alibaba", formula="alibaba"),
"qwen-vl": ModelInfo("qwen-vl", "alibaba", formula="alibaba"),
"qwen-vl-chat": ModelInfo("qwen-vl-chat", "alibaba", formula="alibaba"),
"qwen-vl-plus": ModelInfo("qwen-vl-plus", "alibaba", formula="alibaba"),
"qwen-2.5-72b": ModelInfo("qwen-2.5-72b", "alibaba", formula="alibaba"),
"qwen-2.5-32b": ModelInfo("qwen-2.5-32b", "alibaba", formula="alibaba"),
"qwen-2.5-14b": ModelInfo("qwen-2.5-14b", "alibaba", formula="alibaba"),
"qwen-2.5-7b": ModelInfo("qwen-2.5-7b", "alibaba", formula="alibaba"),
# Baidu
"ernie-4.0": ModelInfo("ernie-4.0", "baidu", formula="baidu"),
"ernie-3.5": ModelInfo("ernie-3.5", "baidu", formula="baidu"),
"ernie-3.0": ModelInfo("ernie-3.0", "baidu", formula="baidu"),
"ernie-speed": ModelInfo("ernie-speed", "baidu", formula="baidu"),
"ernie-lite": ModelInfo("ernie-lite", "baidu", formula="baidu"),
"ernie-tiny": ModelInfo("ernie-tiny", "baidu", formula="baidu"),
"ernie-bot": ModelInfo("ernie-bot", "baidu", formula="baidu"),
"ernie-bot-4": ModelInfo("ernie-bot-4", "baidu", formula="baidu"),
# Huawei
"pangu-alpha-2.6b": ModelInfo("pangu-alpha-2.6b", "huawei", formula="huawei"),
"pangu-alpha-13b": ModelInfo("pangu-alpha-13b", "huawei", formula="huawei"),
"pangu-alpha-200b": ModelInfo("pangu-alpha-200b", "huawei", formula="huawei"),
"pangu-coder": ModelInfo("pangu-coder", "huawei", formula="huawei"),
"pangu-coder-15b": ModelInfo("pangu-coder-15b", "huawei", formula="huawei"),
# Yandex
"yalm-100b": ModelInfo("yalm-100b", "yandex", formula="yandex"),
"yalm-200b": ModelInfo("yalm-200b", "yandex", formula="yandex"),
"yagpt": ModelInfo("yagpt", "yandex", formula="yandex"),
"yagpt-2": ModelInfo("yagpt-2", "yandex", formula="yandex"),
# Stability
"stablelm-alpha-3b": ModelInfo("stablelm-alpha-3b", "stability", formula="stability"),
"stablelm-alpha-7b": ModelInfo("stablelm-alpha-7b", "stability", formula="stability"),
"stablelm-base-alpha-3b": ModelInfo("stablelm-base-alpha-3b", "stability", formula="stability"),
"stablelm-base-alpha-7b": ModelInfo("stablelm-base-alpha-7b", "stability", formula="stability"),
"stablelm-tuned-alpha-3b": ModelInfo("stablelm-tuned-alpha-3b", "stability", formula="stability"),
"stablelm-tuned-alpha-7b": ModelInfo("stablelm-tuned-alpha-7b", "stability", formula="stability"),
"stablelm-zephyr-3b": ModelInfo("stablelm-zephyr-3b", "stability", formula="stability"),
# TII
"falcon-7b": ModelInfo("falcon-7b", "tii", formula="tii"),
"falcon-7b-instruct": ModelInfo("falcon-7b-instruct", "tii", formula="tii"),
"falcon-40b": ModelInfo("falcon-40b", "tii", formula="tii"),
"falcon-40b-instruct": ModelInfo("falcon-40b-instruct", "tii", formula="tii"),
"falcon-180b": ModelInfo("falcon-180b", "tii", formula="tii"),
"falcon-180b-chat": ModelInfo("falcon-180b-chat", "tii", formula="tii"),
# EleutherAI
"gpt-neo-125m": ModelInfo("gpt-neo-125m", "eleutherai", formula="eleutherai"),
"gpt-neo-1.3b": ModelInfo("gpt-neo-1.3b", "eleutherai", formula="eleutherai"),
"gpt-neo-2.7b": ModelInfo("gpt-neo-2.7b", "eleutherai", formula="eleutherai"),
"gpt-neox-20b": ModelInfo("gpt-neox-20b", "eleutherai", formula="eleutherai"),
"pythia-70m": ModelInfo("pythia-70m", "eleutherai", formula="eleutherai"),
"pythia-160m": ModelInfo("pythia-160m", "eleutherai", formula="eleutherai"),
"pythia-410m": ModelInfo("pythia-410m", "eleutherai", formula="eleutherai"),
"pythia-1b": ModelInfo("pythia-1b", "eleutherai", formula="eleutherai"),
"pythia-1.4b": ModelInfo("pythia-1.4b", "eleutherai", formula="eleutherai"),
"pythia-2.8b": ModelInfo("pythia-2.8b", "eleutherai", formula="eleutherai"),
"pythia-6.9b": ModelInfo("pythia-6.9b", "eleutherai", formula="eleutherai"),
"pythia-12b": ModelInfo("pythia-12b", "eleutherai", formula="eleutherai"),
# MosaicML
"mpt-7b": ModelInfo("mpt-7b", "mosaicml", formula="mosaicml"),
"mpt-7b-chat": ModelInfo("mpt-7b-chat", "mosaicml", formula="mosaicml"),
"mpt-7b-instruct": ModelInfo("mpt-7b-instruct", "mosaicml", formula="mosaicml"),
"mpt-30b": ModelInfo("mpt-30b", "mosaicml", formula="mosaicml"),
"mpt-30b-chat": ModelInfo("mpt-30b-chat", "mosaicml", formula="mosaicml"),
"mpt-30b-instruct": ModelInfo("mpt-30b-instruct", "mosaicml", formula="mosaicml"),
# Replit
"replit-code-v1-3b": ModelInfo("replit-code-v1-3b", "replit", formula="replit"),
"replit-code-v1.5-3b": ModelInfo("replit-code-v1.5-3b", "replit", formula="replit"),
"replit-code-v2-3b": ModelInfo("replit-code-v2-3b", "replit", formula="replit"),
# MiniMax
"abab5.5-chat": ModelInfo("abab5.5-chat", "minimax", formula="minimax"),
"abab5.5s-chat": ModelInfo("abab5.5s-chat", "minimax", formula="minimax"),
"abab6-chat": ModelInfo("abab6-chat", "minimax", formula="minimax"),
"abab6.5-chat": ModelInfo("abab6.5-chat", "minimax", formula="minimax"),
"abab6.5s-chat": ModelInfo("abab6.5s-chat", "minimax", formula="minimax"),
"minimax-m2.5": ModelInfo("minimax-m2.5", "minimax", formula="minimax_m25"),
# Aleph Alpha
"luminous-base": ModelInfo("luminous-base", "aleph_alpha", formula="aleph_alpha"),
"luminous-extended": ModelInfo("luminous-extended", "aleph_alpha", formula="aleph_alpha"),
"luminous-supreme": ModelInfo("luminous-supreme", "aleph_alpha", formula="aleph_alpha"),
"luminous-supreme-control": ModelInfo("luminous-supreme-control", "aleph_alpha", formula="aleph_alpha"),
# DeepSeek
"deepseek-coder-1.3b": ModelInfo("deepseek-coder-1.3b", "deepseek", formula="deepseek"),
"deepseek-coder-6.7b": ModelInfo("deepseek-coder-6.7b", "deepseek", formula="deepseek"),
"deepseek-coder-33b": ModelInfo("deepseek-coder-33b", "deepseek", formula="deepseek"),
"deepseek-coder-instruct": ModelInfo("deepseek-coder-instruct", "deepseek", formula="deepseek"),
"deepseek-vl-1.3b": ModelInfo("deepseek-vl-1.3b", "deepseek", formula="deepseek"),
"deepseek-vl-7b": ModelInfo("deepseek-vl-7b", "deepseek", formula="deepseek"),
"deepseek-llm-7b": ModelInfo("deepseek-llm-7b", "deepseek", formula="deepseek"),
"deepseek-llm-67b": ModelInfo("deepseek-llm-67b", "deepseek", formula="deepseek"),
"deepseek-v3": ModelInfo("deepseek-v3", "deepseek", formula="deepseek"),
"deepseek-v3-base": ModelInfo("deepseek-v3-base", "deepseek", formula="deepseek"),
"deepseek-v3.2": ModelInfo("deepseek-v3.2", "deepseek", formula="deepseek"),
# Tsinghua
"chatglm-6b": ModelInfo("chatglm-6b", "tsinghua", formula="tsinghua"),
"chatglm2-6b": ModelInfo("chatglm2-6b", "tsinghua", formula="tsinghua"),
"chatglm3-6b": ModelInfo("chatglm3-6b", "tsinghua", formula="tsinghua"),
"glm-4": ModelInfo("glm-4", "tsinghua", formula="tsinghua"),
"glm-4v": ModelInfo("glm-4v", "tsinghua", formula="tsinghua"),
"glm-5": ModelInfo("glm-5", "tsinghua", formula="glm5"),
# Volcengine (Doubao)
"doubao-seed-2-0-pro": ModelInfo("doubao-seed-2-0-pro", "volcengine", formula="volcengine"),
# Moonshot (Kimi)
"kimi-k2.5": ModelInfo("kimi-k2.5", "moonshot", formula="moonshot"),
# RWKV
"rwkv-4-169m": ModelInfo("rwkv-4-169m", "rwkv", formula="rwkv"),
"rwkv-4-430m": ModelInfo("rwkv-4-430m", "rwkv", formula="rwkv"),
"rwkv-4-1b5": ModelInfo("rwkv-4-1b5", "rwkv", formula="rwkv"),
"rwkv-4-3b": ModelInfo("rwkv-4-3b", "rwkv", formula="rwkv"),
"rwkv-4-7b": ModelInfo("rwkv-4-7b", "rwkv", formula="rwkv"),
"rwkv-4-14b": ModelInfo("rwkv-4-14b", "rwkv", formula="rwkv"),
"rwkv-5-world": ModelInfo("rwkv-5-world", "rwkv", formula="rwkv"),
# Community
"vicuna-7b": ModelInfo("vicuna-7b", "community", formula="community"),
"vicuna-13b": ModelInfo("vicuna-13b", "community", formula="community"),
"vicuna-33b": ModelInfo("vicuna-33b", "community", formula="community"),
"alpaca-7b": ModelInfo("alpaca-7b", "community", formula="community"),
"alpaca-13b": ModelInfo("alpaca-13b", "community", formula="community"),
"wizardlm-7b": ModelInfo("wizardlm-7b", "community", formula="community"),
"wizardlm-13b": ModelInfo("wizardlm-13b", "community", formula="community"),
"wizardlm-30b": ModelInfo("wizardlm-30b", "community", formula="community"),
"orca-mini-3b": ModelInfo("orca-mini-3b", "community", formula="community"),
"orca-mini-7b": ModelInfo("orca-mini-7b", "community", formula="community"),
"orca-mini-13b": ModelInfo("orca-mini-13b", "community", formula="community"),
"zephyr-7b-alpha": ModelInfo("zephyr-7b-alpha", "community", formula="community"),
"zephyr-7b-beta": ModelInfo("zephyr-7b-beta", "community", formula="community"),
# Microsoft
"phi-3-mini": ModelInfo("phi-3-mini", "microsoft", formula="microsoft"),
"phi-3-small": ModelInfo("phi-3-small", "microsoft", formula="microsoft"),
"phi-3-medium": ModelInfo("phi-3-medium", "microsoft", formula="microsoft"),
"phi-3.5-mini": ModelInfo("phi-3.5-mini", "microsoft", formula="microsoft"),
# Amazon
"titan-text-express": ModelInfo("titan-text-express", "amazon", formula="amazon"),
"titan-text-lite": ModelInfo("titan-text-lite", "amazon", formula="amazon"),
"titan-embed-text": ModelInfo("titan-embed-text", "amazon", formula="amazon"),
# Nvidia
"nemotron-4-340b": ModelInfo("nemotron-4-340b", "nvidia", formula="nvidia"),
"nemotron-3-8b": ModelInfo("nemotron-3-8b", "nvidia", formula="nvidia"),
# IBM
"granite-13b-chat": ModelInfo("granite-13b-chat", "ibm", formula="ibm"),
"granite-13b-instruct": ModelInfo("granite-13b-instruct", "ibm", formula="ibm"),
"granite-20b-code": ModelInfo("granite-20b-code", "ibm", formula="ibm"),
# Salesforce
"codegen-16b": ModelInfo("codegen-16b", "salesforce", formula="salesforce"),
"codegen-6b": ModelInfo("codegen-6b", "salesforce", formula="salesforce"),
"codegen-2b": ModelInfo("codegen-2b", "salesforce", formula="salesforce"),
# BigCode
"starcoder": ModelInfo("starcoder", "bigcode", formula="bigcode"),
"starcoder2-15b": ModelInfo("starcoder2-15b", "bigcode", formula="bigcode"),
"starcoderbase": ModelInfo("starcoderbase", "bigcode", formula="bigcode"),
"starcoder2-3b": ModelInfo("starcoder2-3b", "bigcode", formula="bigcode"),
"starcoder2-7b": ModelInfo("starcoder2-7b", "bigcode", formula="bigcode"),
"starcoder-plus": ModelInfo("starcoder-plus", "bigcode", formula="bigcode"),
"starcoderbase-1b": ModelInfo("starcoderbase-1b", "bigcode", formula="bigcode"),
"starcoderbase-3b": ModelInfo("starcoderbase-3b", "bigcode", formula="bigcode"),
"starcoderbase-7b": ModelInfo("starcoderbase-7b", "bigcode", formula="bigcode"),
# Databricks
"dbrx": ModelInfo("dbrx", "databricks", formula="databricks"),
"dbrx-instruct": ModelInfo("dbrx-instruct", "databricks", formula="databricks", pricing_id="dbrx-instruct"),
"dbrx-base": ModelInfo("dbrx-base", "databricks", formula="databricks", pricing_id="dbrx-base"),
"dolly-v2-12b": ModelInfo("dolly-v2-12b", "databricks", formula="databricks", pricing_id="dolly-v2-12b"),
"dolly-v2-7b": ModelInfo("dolly-v2-7b", "databricks", formula="databricks", pricing_id="dolly-v2-7b"),
"dolly-v2-3b": ModelInfo("dolly-v2-3b", "databricks", formula="databricks", pricing_id="dolly-v2-3b"),
# Voyage
"voyage-2": ModelInfo("voyage-2", "voyage", formula="voyage", pricing_id="voyage-2"),
"voyage-large-2": ModelInfo("voyage-large-2", "voyage", formula="voyage", pricing_id="voyage-large-2"),
"voyage-code-2": ModelInfo("voyage-code-2", "voyage", formula="voyage", pricing_id="voyage-code-2"),
"voyage-finance-2": ModelInfo("voyage-finance-2", "voyage", formula="voyage", pricing_id="voyage-finance-2"),
"voyage-law-2": ModelInfo("voyage-law-2", "voyage", formula="voyage", pricing_id="voyage-law-2"),
"voyage-multilingual-2": ModelInfo("voyage-multilingual-2", "voyage", formula="voyage", pricing_id="voyage-multilingual-2"),
}
def _build_alias_map(models: Iterable[ModelInfo]) -> Dict[str, str]:
alias_map: Dict[str, str] = {}
for model in models:
if model.aliases:
for alias in model.aliases:
alias_map[normalize_model_name(alias)] = model.name
return alias_map
ALIASES: Dict[str, str] = _build_alias_map(MODELS.values())
MODEL_LOOKUP: Dict[str, ModelInfo] = {normalize_model_name(k): v for k, v in MODELS.items()}
MODEL_LOOKUP.update({alias: MODELS[target] for alias, target in ALIASES.items()})
# All normalized model names for fuzzy matching
_ALL_NORMALIZED_NAMES: List[str] = list(MODEL_LOOKUP.keys())
def _extract_model_part(name: str) -> str:
"""Extract model part from 'provider/model' or 'provider:model' format."""
name = name.strip()
for sep in ("/", ":"):
if sep in name:
return name.split(sep, 1)[-1].strip()
return name
def _fuzzy_match_model(model_name: str) -> Optional[ModelInfo]:
"""Fuzzy match model name when exact lookup fails."""
normalized = normalize_model_name(model_name)
model_part = normalize_model_name(_extract_model_part(model_name))
# 1. Try model part only (e.g. anthropic/claude-sonnet-4-6 -> claude-sonnet-4-6)
if model_part != normalized:
info = MODEL_LOOKUP.get(model_part)
if info:
return info
# 2. Substring match: input contains a known model, or known model contains input
for key, info in MODEL_LOOKUP.items():
if model_part in key or key in model_part:
return info
# 3. difflib fuzzy match (cutoff 0.5 = 50% similarity)
candidates = difflib.get_close_matches(model_part, _ALL_NORMALIZED_NAMES, n=1, cutoff=0.5)
if candidates:
return MODEL_LOOKUP.get(candidates[0])
# 4. Relaxed: match key parts (e.g. claude-sonnet-4 matches claude-3.5-sonnet)
model_tokens = set(re.split(r"[-_.]", model_part)) - {""}
for key in _ALL_NORMALIZED_NAMES:
key_tokens = set(re.split(r"[-_.]", key)) - {""}
overlap = len(model_tokens & key_tokens) / max(len(model_tokens), 1)
if overlap >= 0.6:
return MODEL_LOOKUP.get(key)
return None
def get_model_info(model_name: str) -> Optional[ModelInfo]:
info = MODEL_LOOKUP.get(normalize_model_name(model_name))
if info:
return info
return _fuzzy_match_model(model_name)
def get_all_supported_models() -> List[str]:
return sorted(set(MODELS.keys()))
def get_supported_models_by_provider() -> Dict[str, List[str]]:
models_by_provider: Dict[str, List[str]] = {}
for model in MODELS.values():
models_by_provider.setdefault(model.provider, []).append(model.name)
for provider, models in models_by_provider.items():
models_by_provider[provider] = sorted(set(models))
return dict(sorted(models_by_provider.items()))
FILE:scripts/registry/pricing.py
"""Pricing registry for scripts."""
from typing import Optional
USD_TO_INR = 83.0 # Conversion rate as of July 10 2025
PRICING = {
"gpt-4": {"input": 0.03, "output": 0.06},
"gpt-4-32k": {"input": 0.06, "output": 0.12},
"dbrx-instruct": {"input": 0.001, "output": 0.002},
"dbrx-base": {"input": 0.001, "output": 0.002},
"dolly-v2-12b": {"input": 0.001, "output": 0.002},
"dolly-v2-7b": {"input": 0.001, "output": 0.002},
"dolly-v2-3b": {"input": 0.001, "output": 0.002},
"voyage-2": {"input": 0.0001, "output": 0.0001},
"voyage-large-2": {"input": 0.0001, "output": 0.0001},
"voyage-code-2": {"input": 0.0001, "output": 0.0001},
"voyage-finance-2": {"input": 0.0001, "output": 0.0001},
"voyage-law-2": {"input": 0.0001, "output": 0.0001},
"voyage-multilingual-2": {"input": 0.0001, "output": 0.0001},
"gpt-4-turbo": {"input": 0.01, "output": 0.03},
"gpt-4o": {"input": 0.005, "output": 0.015},
"gpt-4o-mini": {"input": 0.00015, "output": 0.0006},
"gpt-3.5-turbo": {"input": 0.001, "output": 0.002},
"gpt-3.5-turbo-16k": {"input": 0.003, "output": 0.004},
"claude-3-opus-20240229": {"input": 0.015, "output": 0.075},
"claude-3-sonnet-20240229": {"input": 0.003, "output": 0.015},
"claude-3-haiku-20240307": {"input": 0.00025, "output": 0.00125},
"claude-3.5-sonnet-20240620": {"input": 0.003, "output": 0.015},
"claude-3.5-sonnet-20241022": {"input": 0.003, "output": 0.015},
"claude-3.5-haiku-20241022": {"input": 0.001, "output": 0.005},
"claude-3-5-sonnet-20240620": {"input": 0.003, "output": 0.015},
}
def get_pricing_rate(model: str, input_tokens: bool) -> Optional[float]:
pricing = PRICING.get(model)
if not pricing:
return None
return pricing["input" if input_tokens else "output"]
FILE:scripts/registry/__init__.py
"""Registry package for model metadata, pricing, and aliases."""
from .models import ( # noqa: F401
ModelInfo,
get_all_supported_models,
get_model_info,
get_supported_models_by_provider,
normalize_model_name,
)
from .pricing import USD_TO_INR, get_pricing_rate # noqa: F401
FILE:scripts/__init__.py
"""
scripts CLI package.
This package powers the scripts command-line interface for counting tokens and
estimating costs across a large catalog of LLM models.
"""
__version__ = "1.0.10"
__author__ = "Raja CSP Raman"
__email__ = "[email protected]"
FILE:tests/test_main_models.py
"""
Tests for main models with benchmark data.
Uses fuzzy model name matching (e.g. anthropic/claude-sonnet-4-6 -> claude-sonnet-4-6).
Benchmark: 8927 chars -> expected token counts per model.
Tolerance: ±30% for approximation-based models (formula-based estimates vary with text).
OpenAI (tiktoken): skip when tiktoken not installed.
"""
import pytest
from scripts.core import TokenCounter, count_tokens
from scripts.registry.models import get_model_info as registry_get_model_info
try:
import tiktoken
TIKTOKEN_AVAILABLE = True
except ImportError:
TIKTOKEN_AVAILABLE = False
# Benchmark: 8927 chars, expected tokens per model
# 1 token ≈ chars/token from user's table
MAIN_MODELS_BENCHMARK = [
("anthropic/claude-sonnet-4-6", 8927, 2763),
("anthropic/claude-sonnet-4-5", 8927, 2763),
("anthropic/claude-opus-4.6", 8927, 2763),
("openai/gpt-5.2-codex", 8927, 2459),
("google/gemini-3.1-pro-preview", 8927, 2627),
("z-ai/glm-5", 8927, 2457),
("volcengine/doubao-seed-2-0-pro", 8927, 2702),
("moonshot/kimi-k2.5", 8927, 2402),
("minimax/MiniMax-M2.5", 8927, 2428),
("deepseek-v3.2", 8927, 2578),
]
# Also test canonical names (no provider prefix)
CANONICAL_BENCHMARK = [
("claude-sonnet-4-6", 8927, 2763),
("gpt-5.2-codex", 8927, 2459), # OpenAI, requires tiktoken
("glm-5", 8927, 2457),
("deepseek-v3.2", 8927, 2578),
]
# OpenAI models that need tiktoken
OPENAI_MODELS = {"gpt-5.2-codex", "openai/gpt-5.2-codex"}
# Benchmark: 3050 chars, Chinese-mixed content (混杂中文)
# 1 token ≈ 1.5–2.4 chars (CJK tokenizes to fewer chars/token)
MAIN_MODELS_BENCHMARK_ZH = [
("anthropic/claude-sonnet-4-6", 3050, 1913),
("anthropic/claude-sonnet-4-5", 3050, 1913),
("anthropic/claude-opus-4.6", 3050, 1913),
("openai/gpt-5.2-codex", 3050, 1564),
("google/gemini-3.1-pro-preview", 3050, 1473),
("z-ai/glm-5", 3050, 1318),
("volcengine/doubao-seed-2-0-pro", 3050, 1494),
("moonshot/kimi-k2.5", 3050, 1257),
("minimax/MiniMax-M2.5", 3050, 1289),
("deepseek-v3.2", 3050, 1361),
]
def _make_test_text(length: int) -> str:
"""Generate test text of given length (mixed content for realistic tokenization)."""
base = (
"Token counting benchmark. 这是一段混合中英文的测试文本。"
"Code: def hello(): return 42\n"
"Lorem ipsum dolor sit amet, consectetur adipiscing elit. "
)
if length <= len(base):
return base[:length]
repeat = (length // len(base)) + 1
return (base * repeat)[:length]
def _make_chinese_mixed_text(length: int) -> str:
"""Generate Chinese-heavy mixed text (混杂中文) for CJK tokenization benchmark.
~90% CJK to match user benchmark (3050 chars -> ~1.6-2.4 chars/token)."""
base = (
"大语言模型token计数基准测试。这是一段混杂了中文与英文的测试文本，"
"用于验证不同模型对CJK字符的tokenization表现。"
"大语言模型在处理中文时通常每个汉字约一至两个token。"
"测试数据包含中英混合内容。"
)
if length <= len(base):
return base[:length]
repeat = (length // len(base)) + 1
return (base * repeat)[:length]
@pytest.fixture
def benchmark_text():
return _make_test_text(8927)
@pytest.fixture
def benchmark_text_zh():
return _make_chinese_mixed_text(3050)
class TestFuzzyModelMatching:
"""Test fuzzy model name matching."""
@pytest.mark.parametrize(
"input_name,expected_canonical",
[
("anthropic/claude-sonnet-4-6", "claude-sonnet-4-6"),
("openai/gpt-5.2-codex", "gpt-5.2-codex"),
("z-ai/glm-5", "glm-5"),
("minimax/MiniMax-M2.5", "minimax-m2.5"),
("deepseek-v3.2", "deepseek-v3.2"),
],
)
def test_fuzzy_match_resolves(self, input_name, expected_canonical):
info = registry_get_model_info(input_name)
assert info is not None, f"Model '{input_name}' should resolve via fuzzy match"
assert info.name.lower() == expected_canonical.lower()
def test_exact_match_unchanged(self):
assert registry_get_model_info("gpt-4") is not None
assert registry_get_model_info("claude-3-opus") is not None
class TestMainModelsTokenCount:
"""Test token counts for main models against benchmark data."""
@pytest.mark.parametrize("model_name,char_count,expected_tokens", MAIN_MODELS_BENCHMARK)
def test_main_models_with_fuzzy_name(self, benchmark_text, model_name, char_count, expected_tokens):
if model_name in OPENAI_MODELS and not TIKTOKEN_AVAILABLE:
pytest.skip("tiktoken required for OpenAI models")
assert len(benchmark_text) == char_count
counter = TokenCounter(model_name)
tokens = counter.count(benchmark_text)
# ±30% tolerance for approximation-based models
tolerance = 0.30
low = int(expected_tokens * (1 - tolerance))
high = int(expected_tokens * (1 + tolerance))
assert low <= tokens <= high, (
f"{model_name}: got {tokens} tokens, expected ~{expected_tokens} (range {low}-{high})"
)
@pytest.mark.parametrize("model_name,char_count,expected_tokens", CANONICAL_BENCHMARK)
def test_canonical_names(self, benchmark_text, model_name, char_count, expected_tokens):
if model_name in OPENAI_MODELS and not TIKTOKEN_AVAILABLE:
pytest.skip("tiktoken required for OpenAI models")
assert len(benchmark_text) == char_count
tokens = count_tokens(benchmark_text, model_name)
tolerance = 0.30
low = int(expected_tokens * (1 - tolerance))
high = int(expected_tokens * (1 + tolerance))
assert low <= tokens <= high, (
f"{model_name}: got {tokens} tokens, expected ~{expected_tokens} (range {low}-{high})"
)
@pytest.mark.parametrize("model_name,char_count,expected_tokens", MAIN_MODELS_BENCHMARK_ZH)
def test_main_models_chinese_mixed(self, benchmark_text_zh, model_name, char_count, expected_tokens):
"""Test token counts for Chinese-mixed text (混杂中文). CJK chars ≈ 1–2 chars/token."""
if model_name in OPENAI_MODELS and not TIKTOKEN_AVAILABLE:
pytest.skip("tiktoken required for OpenAI models")
assert len(benchmark_text_zh) == char_count
counter = TokenCounter(model_name)
tokens = counter.count(benchmark_text_zh)
tolerance = 0.30
low = int(expected_tokens * (1 - tolerance))
high = int(expected_tokens * (1 + tolerance))
assert low <= tokens <= high, (
f"{model_name} (ZH): got {tokens} tokens, expected ~{expected_tokens} (range {low}-{high})"
)
FILE:tests/__init__.py
"""Tests for prompt_token_counter."""
Enables any skill to gain self-evolution capabilities. Use when: (1) User asks to add self-evolution to a skill, (2) User wants a skill to learn from feedbac...
---
name: skill-self-evolution-enhancer
description: "Enables any skill to gain self-evolution capabilities. Use when: (1) User asks to add self-evolution to a skill, (2) User wants a skill to learn from feedback and errors, (3) Scaling self-improvement to multiple skills with per-skill evolution logic. Outputs domain-specific .learnings/, EVOLUTION.md, and Review-Apply-Report workflow."
metadata:
---
# Skill Self-Evolution Enhancer
This skill enables **other skills** to gain self-evolution capabilities similar to self-improving-agent. A skill that originally has no self-evolution will, after enhancement, have: logging, learning from user feedback, promotion to rules, and a Review→Apply→Report loop—all tailored to its domain.
## Quick Reference
| Step | Action |
|------|--------|
| User requests evolution for skill X | Read target skill's SKILL.md |
| Deep analysis | Identify capabilities, scenarios, evolution directions |
| Extract domain | Name, use cases, triggers, areas, promotion targets |
| Generate .learnings/ | Domain-specific LEARNINGS.md, ERRORS.md, FEATURE_REQUESTS.md |
| Generate EVOLUTION.md | Triggers, Review-Apply-Report, OpenClaw feedback rules |
| Language | Match target skill's user language (infer from SKILL.md) |
## When to Use
- User says: "给 skill X 加上自进化能力" / "Add self-evolution to skill X"
- Scaling self-improvement across many skills (each with its own evolution direction)
- Target skill is non-coding (e.g., 洗稿能手, 电脑加速) and needs domain-specific triggers
## Workflow
### Step 1: Read Target Skill
```
Read(target_skill_path/SKILL.md)
```
Obtain path from user or infer (e.g., `skills/xxx`, `~/.cursor/skills/xxx`).
### Step 2: Deep Capability & Scenario Analysis
**Before generating** any config, analyze the target skill deeply:
**Capabilities** (what the skill does):
- Primary outputs and workflows
- Secondary or edge capabilities
- Dependencies (tools, APIs, formats)
**Scenarios** (when and how it is used):
- User personas
- Typical tasks (e.g., 科普改写 vs 汇报改写)
- Input/output patterns
**Evolution directions** (what can improve):
- User feedback patterns (e.g., "改得不通顺" → style)
- Failure modes (e.g., "优化无效" → strategy)
- Recurring corrections → domain-specific rules
**Use cases** → infer from description, Quick Reference, examples
### Step 3: Extract Domain Config
When reading the target skill, extract:
| Field | Where to Find | Example |
|-------|---------------|---------|
| **Domain name** | `name` in frontmatter, title | 洗稿能手, 电脑加速 |
| **Use cases / scenarios** | Description, Quick Reference, examples | 科普、汇报、直播 |
| **Learning triggers** | User feedback phrases in examples | "改得不通顺", "不像口播", "风格不对" |
| **Error triggers** | Failure modes | "优化无效", "某些电脑不适用", "报错" |
| **Areas** | Output types, workflow stages | 文案/口播/短视频脚本, 或 系统优化/卡顿/报错 |
| **Promotion targets** | Skill-specific rules | `{skill}-专属进化规则.md`, `{skill}-最佳实践.md` |
**Language**: Infer from SKILL.md content (Chinese vs English). Generate all output files in that language.
Use [assets/DOMAIN-CONFIG-TEMPLATE.md](assets/DOMAIN-CONFIG-TEMPLATE.md) to structure the extracted data.
### Step 4: Generate .learnings/
Create inside target skill directory: `target_skill_path/.learnings/`
**Structure** (same as self-improving-agent):
- `.learnings/LEARNINGS.md`
- `.learnings/ERRORS.md`
- `.learnings/FEATURE_REQUESTS.md`
Use templates from `assets/`; parameterize with domain areas, categories, promotion targets. Write in the target skill's language.
### Step 5: Generate EVOLUTION.md
Create `target_skill_path/EVOLUTION.md` using [assets/EVOLUTION-RULES-TEMPLATE.md](assets/EVOLUTION-RULES-TEMPLATE.md).
**Must include**:
- Quick Reference: domain triggers → actions
- **Review→Apply→Report** loop (see below)
- Detection triggers (when to log)
- Promotion decision tree
- Area tags
- Domain-specific activation conditions (for hooks)
- Experience invalidation / update rules (when user corrects again)
### Step 6: Optional – Activator Script
If target skill has `scripts/`, add `scripts/activator.sh` with domain-specific reminder text. Adapt from self-improving-agent; replace generic prompts with domain triggers.
## Review → Apply → Report Loop
The enhanced skill must **use** learnings, not only log them. Include this in EVOLUTION.md or the enhanced skill's instructions:
### Before Task
- Load relevant entries from `.learnings/LEARNINGS.md` (and ERRORS.md if applicable)
- Filter by area, tags, or keywords
- Note which entries apply to the current task
### During Task
- Apply learnings when relevant
- Optionally annotate output: "本次参考了 [LRN-xxx]: ..." (or equivalent in target language)
### After Task
- Summarize for user: which learnings were used, what evolution result, what improvement
- Let OpenClaw decide: per-use mention vs end-of-task summary
**Example** (Chinese): "本次改写了口播稿，参考了经验 [LRN-20250115-001]（科普场景应避免过于书面），相比之前更口语化。"
**Example** (English): "Used learning [LRN-20250115-001] (avoid formal tone for科普) in this rewrite; output is more conversational than before."
## User Preference vs Domain Best Practice
| Type | Storage | Example |
|------|---------|---------|
| **User preference** | MEMORY.md (user-level) | "This user prefers shorter sentences" |
| **Domain best practice** | `.learnings/LEARNINGS.md` | "科普场景应避免过于书面" |
Evolution is driven by **user feedback**; log and promote based on user corrections and recurring patterns.
## OpenClaw Active Feedback
Add to the enhanced skill or SOUL.md/AGENTS.md:
- When using experience from `.learnings/`, briefly tell the user
- At end of task, optionally summarize: evolution used, improvements
- Let OpenClaw decide when to surface (per-use vs summary)
See [references/openclaw-feedback.md](references/openclaw-feedback.md) for SOUL.md and AGENTS.md snippets.
## Experience Invalidation & Update
When user corrects again after a learning was applied:
- Add `Contradicted-By: LRN-YYYYMMDD-XXX` to the original entry
- Mark `Last-Valid` or `Status: superseded` if the learning is no longer valid
- Increment `Recurrence-Count` if the pattern recurs but the fix is different
Include in LEARNINGS template: `Recurrence-Count`, `Last-Valid`, `Contradicted-By`.
## Domain Extraction Framework
### Trigger Extraction
**Learning triggers** (user feedback → log to LEARNINGS.md):
- Look for: "用户说", "when user says", example dialogs
- Infer: common corrections, style mismatches, scene-specific preferences
- Add generic fallbacks: "不对", "不是这样", "改一下"
**Error triggers** (failures → log to ERRORS.md):
- Look for: "失败", "报错", "不适用", "when X fails"
- Infer: environment-specific failures, edge cases
- Add generic fallbacks: "操作失败", "未达到预期"
### Area Mapping
Define 3–6 areas that partition the skill's scope. Use domain-specific areas, not coding areas.
### Promotion Target Naming
- `{skill-name}-专属进化规则.md` — evolution rules, style preferences
- `{skill-name}-最佳实践.md` — best practices
- `{skill-name}-安全规范.md` — safety constraints (e.g., 电脑加速)
Use kebab-case for skill name in filenames.
## Logging Format (Reuse from Self-Improving-Agent)
ID format: `LRN-YYYYMMDD-XXX`, `ERR-YYYYMMDD-XXX`, `FEAT-YYYYMMDD-XXX`
Statuses: `pending` | `in_progress` | `resolved` | `wont_fix` | `promoted` | `promoted_to_skill`
For full entry formats, see the self-improving-agent skill's Logging Format section.
## References
- [assets/DOMAIN-CONFIG-TEMPLATE.md](assets/DOMAIN-CONFIG-TEMPLATE.md) — Schema for domain config
- [assets/EVOLUTION-RULES-TEMPLATE.md](assets/EVOLUTION-RULES-TEMPLATE.md) — EVOLUTION.md template
- [references/domain-examples.md](references/domain-examples.md) — 洗稿能手, 电脑加速 examples
- [references/openclaw-feedback.md](references/openclaw-feedback.md) — SOUL.md, AGENTS.md snippets for active feedback
- [scripts/generate-evolution.sh](scripts/generate-evolution.sh) — Optional scaffold generator
## Source
- Based on: self-improving-agent 3.0.1
- Purpose: Enable any skill to gain self-evolution capabilities similar to self-improving-agent
FILE:assets/DOMAIN-CONFIG-TEMPLATE.md
# Domain Config for: [skill-name]
Fill this template when extracting domain from a target skill. Use the extracted values to parameterize LEARNINGS.md, ERRORS.md, FEATURE_REQUESTS.md, and EVOLUTION.md.
## Capability & Scenario Analysis (before extraction)
**Capabilities** (what the skill does):
- Primary outputs:
- Secondary/edge capabilities:
- Dependencies:
**Scenarios** (when and how it is used):
- User personas:
- Typical tasks:
- Input/output patterns:
**Evolution directions** (what can improve):
- User feedback patterns:
- Failure modes:
- Recurring corrections → domain rules:
---
## Domain
- **Name**: (human-readable domain name, e.g., 洗稿能手, 电脑加速)
- **Use cases**: (comma-separated scenarios, e.g., 科普、汇报、直播)
- **Areas**: (domain-specific areas for filtering, e.g., 文案 | 口播 | 短视频脚本)
- **Language**: (target language for generated files: zh | en)
## Learning Triggers (user feedback phrases)
Phrases that indicate user correction or style mismatch → log to LEARNINGS.md:
- "phrase1"
- "phrase2"
- "phrase3"
## Error Triggers (failure modes)
Situations that indicate operation failure → log to ERRORS.md:
- failure1
- failure2
- failure3
## Promotion Targets
Files to promote learnings to when they prove broadly applicable:
- **专属进化规则**: [skill-name]-专属进化规则.md
- **最佳实践 / 安全规范**: [skill-name]-最佳实践.md (or [skill-name]-安全规范.md)
## Domain Categories (optional)
If the skill needs categories beyond correction | insight | knowledge_gap | best_practice:
- category1
- category2
## Notes
- Use kebab-case for skill-name in file paths
- Areas should partition the skill's output/workflow scope
- Add 3+ learning triggers and 2+ error triggers for effective capture
- Storage: same as self-improving-agent — `.learnings/` directly (LEARNINGS.md, ERRORS.md, FEATURE_REQUESTS.md)
FILE:assets/ERRORS-TEMPLATE.md
# Errors Log
{DOMAIN_DESCRIPTION}
**Areas**: {DOMAIN_AREAS}
**Statuses**: pending | in_progress | resolved | wont_fix
## Error Entry Format
Append entries in this format:
```markdown
## [ERR-YYYYMMDD-XXX] operation_or_error_type
**Logged**: ISO-8601 timestamp
**Priority**: high
**Status**: pending
**Area**: {DOMAIN_AREAS_EXAMPLE}
### Summary
Brief description of what failed
### Error
```
Actual error message or output
```
### Context
- Operation attempted
- Input or parameters used
- Environment details if relevant
### Suggested Fix
If identifiable, what might resolve this
### Metadata
- Reproducible: yes | no | unknown
- Related Files: path/to/file.ext
- See Also: ERR-YYYYMMDD-XXX (if recurring)
---
```
## Placeholders to Replace
- `{DOMAIN_DESCRIPTION}`: e.g., "Operation failures, exceptions, and unexpected behaviors for [skill-name]."
- `{DOMAIN_AREAS}`: e.g., "文案 | 口播 | 短视频脚本" or "系统优化 | 卡顿 | 报错"
- `{DOMAIN_AREAS_EXAMPLE}`: One example area from the list
---
FILE:assets/EVOLUTION-RULES-TEMPLATE.md
# [Skill Name] Evolution Rules
Domain-specific self-improvement rules for [skill-name]. Log learnings and errors to `.learnings/`; promote proven patterns to the targets below. **Evolution is driven by user feedback.**
## Quick Reference
| Situation | Action |
|-----------|--------|
| {LEARNING_TRIGGER_1} | Log to `.learnings/LEARNINGS.md` with category `correction` or `style_mismatch` |
| {LEARNING_TRIGGER_2} | Log to `.learnings/LEARNINGS.md` |
| {ERROR_TRIGGER_1} | Log to `.learnings/ERRORS.md` |
| {ERROR_TRIGGER_2} | Log to `.learnings/ERRORS.md` |
| User wants missing capability | Log to `.learnings/FEATURE_REQUESTS.md` |
| Pattern proven across 3+ instances | Promote to {PROMOTION_TARGET_1} |
| Safety/correctness rule | Promote to {PROMOTION_TARGET_2} |
## Review → Apply → Report Loop
**Before task**: Load relevant entries from `.learnings/LEARNINGS.md` (and ERRORS.md if applicable). Filter by area, tags, or keywords.
**During task**: Apply learnings when relevant. Optionally annotate: "本次参考了 [LRN-xxx]: ..." (or equivalent in target language).
**After task**: Tell user which learnings were used, what evolution result, what improvement. Decide per-use mention vs end-of-task summary based on context.
## Detection Triggers
### Learning Triggers (→ LEARNINGS.md)
- {LEARNING_TRIGGER_1}
- {LEARNING_TRIGGER_2}
- {LEARNING_TRIGGER_3}
- User provides correction or preference
- Better approach discovered for recurring task
### Error Triggers (→ ERRORS.md)
- {ERROR_TRIGGER_1}
- {ERROR_TRIGGER_2}
- {ERROR_TRIGGER_3}
- Operation fails unexpectedly
- Unexpected output or behavior
## Domain Activation Conditions (for hooks)
When to remind the agent to check learnings:
- {ACTIVATION_CONDITION_1}
- {ACTIVATION_CONDITION_2}
## Promotion Decision Tree
```
Is the learning skill-specific?
├── Yes → Keep in .learnings/
└── No → Is it style/behavior-related?
├── Yes → Promote to {PROMOTION_TARGET_1}
└── No → Is it safety/correctness-related?
├── Yes → Promote to {PROMOTION_TARGET_2}
└── No → Promote to {PROMOTION_TARGET_1}
```
## Experience Invalidation & Update
When user corrects again after a learning was applied:
- Add `Contradicted-By: LRN-YYYYMMDD-XXX` to the original entry
- Mark `Status: superseded` or `Last-Valid: YYYY-MM-DD` if the learning is no longer valid
- Increment `Recurrence-Count` if the pattern recurs but the fix is different
## Area Tags
| Area | Scope |
|------|-------|
| {AREA_1} | {AREA_1_SCOPE} |
| {AREA_2} | {AREA_2_SCOPE} |
| {AREA_3} | {AREA_3_SCOPE} |
## Priority Guidelines
| Priority | When to Use |
|----------|-------------|
| `critical` | Blocks core functionality, safety risk |
| `high` | Significant impact, affects common workflows |
| `medium` | Moderate impact, workaround exists |
| `low` | Minor inconvenience, edge case |
## Promotion Targets
- **{PROMOTION_TARGET_1}**: Evolution rules, style preferences, best practices
- **{PROMOTION_TARGET_2}**: Safety guidelines, correctness rules (if applicable)
## Placeholders to Replace
- `[skill-name]`, `[Skill Name]`: Domain name
- `{LEARNING_TRIGGER_1}`, `{LEARNING_TRIGGER_2}`, `{LEARNING_TRIGGER_3}`: User feedback phrases from domain config
- `{ERROR_TRIGGER_1}`, `{ERROR_TRIGGER_2}`, `{ERROR_TRIGGER_3}`: Failure modes from domain config
- `{ACTIVATION_CONDITION_1}`, `{ACTIVATION_CONDITION_2}`: Domain-specific activation (e.g., "User says 改得不通顺", "Optimization reported无效")
- `{PROMOTION_TARGET_1}`, `{PROMOTION_TARGET_2}`: e.g., `[skill-name]-专属进化规则.md`, `[skill-name]-最佳实践.md`
- `{AREA_1}`, `{AREA_2}`, `{AREA_3}`: Domain areas
- `{AREA_1_SCOPE}`, etc.: Brief scope description for each area
FILE:assets/FEATURE_REQUESTS-TEMPLATE.md
# Feature Requests
{DOMAIN_DESCRIPTION}
**Areas**: {DOMAIN_AREAS}
**Statuses**: pending | in_progress | resolved | wont_fix
## Feature Request Entry Format
Append entries in this format:
```markdown
## [FEAT-YYYYMMDD-XXX] capability_name
**Logged**: ISO-8601 timestamp
**Priority**: medium
**Status**: pending
**Area**: {DOMAIN_AREAS_EXAMPLE}
### Requested Capability
What the user wanted to do
### User Context
Why they needed it, what problem they're solving
### Complexity Estimate
simple | medium | complex
### Suggested Implementation
How this could be built, what it might extend
### Metadata
- Frequency: first_time | recurring
- Related Features: existing_feature_name
---
```
## Placeholders to Replace
- `{DOMAIN_DESCRIPTION}`: e.g., "Capabilities requested by user that don't currently exist for [skill-name]."
- `{DOMAIN_AREAS}`: e.g., "文案 | 口播 | 短视频脚本" or "系统优化 | 卡顿 | 报错"
- `{DOMAIN_AREAS_EXAMPLE}`: One example area from the list
---
FILE:assets/LEARNINGS-TEMPLATE.md
# Learnings
{DOMAIN_DESCRIPTION}
**Categories**: {DOMAIN_CATEGORIES}
**Areas**: {DOMAIN_AREAS}
**Statuses**: pending | in_progress | resolved | wont_fix | promoted | promoted_to_skill | superseded
## Status Definitions
| Status | Meaning |
|--------|---------|
| `pending` | Not yet addressed |
| `in_progress` | Actively being worked on |
| `resolved` | Issue fixed or knowledge integrated |
| `wont_fix` | Decided not to address (reason in Resolution) |
| `promoted` | Elevated to {PROMOTION_TARGETS} |
| `promoted_to_skill` | Extracted as a reusable skill |
| `superseded` | User corrected again; this learning is no longer valid |
## Learning Entry Format
Append entries in this format:
```markdown
## [LRN-YYYYMMDD-XXX] category
**Logged**: ISO-8601 timestamp
**Priority**: low | medium | high | critical
**Status**: pending
**Area**: {DOMAIN_AREAS_EXAMPLE}
### Summary
One-line description of what was learned
### Details
Full context: what happened, what was wrong, what's correct
### Suggested Action
Specific fix or improvement to make
### Metadata
- Source: conversation | error | user_feedback
- Related Files: path/to/file.ext
- Tags: tag1, tag2
- See Also: LRN-YYYYMMDD-XXX (if related to existing entry)
- Recurrence-Count: 1 (optional, bump when pattern recurs)
- Last-Valid: YYYY-MM-DD (optional, when user contradicted)
- Contradicted-By: LRN-YYYYMMDD-XXX (optional, when superseded)
---
```
## Placeholders to Replace
- `{DOMAIN_DESCRIPTION}`: e.g., "Corrections, insights, and style preferences captured for [skill-name]."
- `{DOMAIN_CATEGORIES}`: e.g., "correction | insight | knowledge_gap | best_practice | style_mismatch | scene_adaptation"
- `{DOMAIN_AREAS}`: e.g., "文案 | 口播 | 短视频脚本" or "系统优化 | 卡顿 | 报错"
- `{DOMAIN_AREAS_EXAMPLE}`: One example area from the list
- `{PROMOTION_TARGETS}`: e.g., "[skill-name]-专属进化规则.md, [skill-name]-最佳实践.md"
---
FILE:README.md
# Skill Self-Evolution Enhancer
A Cursor skill that enables **any skill** to gain self-evolution capabilities. Based on self-improving-agent, it equips skills with: logging, learning from user feedback, promotion of experience to rules, and a Review→Apply→Report loop—all tailored to the target skill's domain.
[中文](README.zh.md)
## Features
- **Domain analysis**: Deep analysis of target skill's capabilities, scenarios, and evolution directions
- **Generate .learnings/**: Domain-specific LEARNINGS.md, ERRORS.md, FEATURE_REQUESTS.md
- **Generate EVOLUTION.md**: Triggers, Review-Apply-Report workflow, OpenClaw feedback rules
- **Multi-skill scaling**: Support multiple skills, each with its own evolution logic
## When to Use
- User says: "Add self-evolution to skill X"
- Scaling self-improvement across many skills (each with its own evolution direction)
- Target skill is non-coding (e.g., content rewriting, system optimization) and needs domain-specific triggers
## Quick Start
1. Enable this skill in Cursor
2. Provide the target skill's path (e.g., `skills/xxx`, `~/.cursor/skills/xxx`)
3. Follow the flow: Read target skill → Domain analysis → Generate .learnings/ and EVOLUTION.md
## Structure
```
skill-self-evolution-enhancer/
├── SKILL.md # Main skill doc
├── assets/ # Templates
│ ├── DOMAIN-CONFIG-TEMPLATE.md
│ ├── EVOLUTION-RULES-TEMPLATE.md
│ ├── LEARNINGS-TEMPLATE.md
│ ├── ERRORS-TEMPLATE.md
│ └── FEATURE_REQUESTS-TEMPLATE.md
├── references/ # Examples
│ ├── domain-examples.md
│ └── openclaw-feedback.md
└── scripts/
└── generate-evolution.sh # Optional scaffold generator
```
## References
- [SKILL.md](SKILL.md) — Full workflow and documentation
- [references/domain-examples.md](references/domain-examples.md) — Domain examples (e.g., content rewriting, system optimization)
## Source
- Based on: self-improving-agent 3.0.1
- Purpose: Enable any skill to gain self-evolution capabilities similar to self-improving-agent
FILE:README.zh.md
# Skill Self-Evolution Enhancer
为任意 Cursor Skill 赋予自进化能力的增强技能。基于 self-improving-agent，让原本不具备自进化的技能获得：日志记录、从用户反馈学习、经验提升为规则、以及 Review→Apply→Report 循环——全部按目标技能的领域定制。
[English](README.md)
## 功能
- **领域分析**：深度分析目标技能的能力、场景与进化方向
- **生成 .learnings/**：领域化的 LEARNINGS.md、ERRORS.md、FEATURE_REQUESTS.md
- **生成 EVOLUTION.md**：触发条件、Review-Apply-Report 流程、OpenClaw 反馈规则
- **多技能扩展**：支持为多个技能分别配置各自的进化逻辑
## 使用场景
- 用户说：「给 skill X 加上自进化能力」
- 需要将自改进能力扩展到多个技能（每个技能有独立进化方向）
- 目标技能为非编程类（如洗稿能手、电脑加速），需要领域特定的触发条件
## 快速开始
1. 在 Cursor 中启用本技能
2. 提供目标技能的路径（如 `skills/xxx`、`~/.cursor/skills/xxx`）
3. 按提示完成：读取目标技能 → 领域分析 → 生成 .learnings/ 与 EVOLUTION.md
## 目录结构
```
skill-self-evolution-enhancer/
├── SKILL.md # 技能主文档
├── assets/ # 模板
│ ├── DOMAIN-CONFIG-TEMPLATE.md
│ ├── EVOLUTION-RULES-TEMPLATE.md
│ ├── LEARNINGS-TEMPLATE.md
│ ├── ERRORS-TEMPLATE.md
│ └── FEATURE_REQUESTS-TEMPLATE.md
├── references/ # 参考示例
│ ├── domain-examples.md
│ └── openclaw-feedback.md
└── scripts/
└── generate-evolution.sh # 可选脚手架生成脚本
```
## 参考
- [SKILL.md](SKILL.md) — 完整工作流与说明
- [references/domain-examples.md](references/domain-examples.md) — 洗稿能手、电脑加速等示例
## 来源
- 基于：self-improving-agent 3.0.1
- 用途：让任意技能获得与 self-improving-agent 类似的自进化能力
FILE:references/domain-examples.md
# Domain Examples
Concrete examples for extracting domain config and generating evolution scaffolding. Use these as reference when enhancing skills.
---
## Example 1: 洗稿能手 (Content Rewriting Skill)
**Domain**: 文案 / 口播 / 短视频脚本改写
### Capability & Scenario Analysis
- **Capabilities**: Rewrite text for different formats (文案, 口播, 短视频); adapt style by scene (科普, 汇报, 直播)
- **Scenarios**: User provides source text; requests rewrite for specific format/scene; may correct style
- **Evolution directions**: User feedback on style ("不通顺", "不像口播") → learn scene-specific preferences; recurring corrections → promote to 专属进化规则
### Domain Config
```markdown
## Domain
- Name: 洗稿能手
- Use cases: 科普, 汇报, 直播, 短视频脚本, 口播稿
- Areas: 文案 | 口播 | 短视频脚本 | 科普 | 汇报 | 直播
- Language: zh
## Learning Triggers
- "改得不通顺"
- "不像口播"
- "风格不对"
- "太书面了"
- "读起来不顺"
- "语气不对"
## Error Triggers
- 改写后用户多次要求重改
- 风格与场景不匹配（如科普写成汇报风）
- 口播稿读起来拗口
## Promotion Targets
- 洗稿能手-专属进化规则.md
- 洗稿能手-最佳实践.md
```
### Generated EVOLUTION.md Quick Reference
| Situation | Action |
|-----------|--------|
| 用户说"改得不通顺" | Log to `.learnings/LEARNINGS.md` with category `style_mismatch` |
| 用户说"不像口播" | Log to `.learnings/LEARNINGS.md` with category `scene_adaptation` |
| 用户说"风格不对" | Log to `.learnings/LEARNINGS.md` with category `correction` |
| 改写后多次重改 | Log to `.learnings/ERRORS.md` |
| 风格与场景不匹配 | Log to `.learnings/LEARNINGS.md` with area 科普/汇报/直播 |
### Domain Categories
- correction
- style_mismatch
- scene_adaptation
- best_practice
- knowledge_gap
### Activation Conditions
- User says "改得不通顺", "不像口播", "风格不对"
- User requests rewrite for a specific scene (科普/汇报/直播)
---
## Example 2: 电脑加速 (PC Optimization Skill)
**Domain**: Windows 优化 / 卡慢 / 卡顿
### Capability & Scenario Analysis
- **Capabilities**: System optimization, startup speed, memory release; diagnose slowdowns
- **Scenarios**: User reports slow PC; agent suggests optimizations; some may fail or cause issues
- **Evolution directions**: "优化无效", "某些电脑不适用" → learn environment-specific solutions; safety issues → promote to 安全规范
### Domain Config
```markdown
## Domain
- Name: 电脑加速
- Use cases: 系统优化, 卡顿排查, 启动加速, 内存释放
- Areas: 系统优化 | 卡顿 | 报错 | 兼容性 | 安全
- Language: zh
## Learning Triggers
- "优化无效"
- "某些电脑不适用"
- "这样改有问题"
- "上次的方法不行"
## Error Triggers
- 优化后系统报错
- 某些电脑/系统版本不适用
- 操作导致异常或蓝屏
- 命令执行失败
## Promotion Targets
- 电脑加速-安全规范.md
- 电脑加速-最佳实践.md
```
### Generated EVOLUTION.md Quick Reference
| Situation | Action |
|-----------|--------|
| 用户说"优化无效" | Log to `.learnings/ERRORS.md` with area 系统优化 |
| 用户说"某些电脑不适用" | Log to `.learnings/LEARNINGS.md` with category `knowledge_gap` |
| 优化后报错 | Log to `.learnings/ERRORS.md` with area 报错 |
| 发现安全风险 | Log to `.learnings/LEARNINGS.md`, promote to 电脑加速-安全规范.md |
### Domain Categories
- correction
- knowledge_gap
- best_practice
- safety_constraint
### Activation Conditions
- User reports "优化无效", "报错", "某些电脑不适用"
- After suggesting optimization steps
---
## Extraction Tips
1. **Sparse skills**: If the target skill has little content, infer from the name and description. Add generic triggers like "用户纠正", "操作失败".
2. **Coding skills**: Use areas like frontend | backend | infra if the skill is coding-related; otherwise use domain-specific areas.
3. **Safety-critical domains** (e.g., 电脑加速): Always include a 安全规范 promotion target.
4. **Style domains** (e.g., 洗稿能手): Include style_mismatch, scene_adaptation categories.
5. **Language**: Infer from SKILL.md; generate all output in that language.
FILE:references/openclaw-feedback.md
# OpenClaw Active Feedback Integration
When a skill is enhanced with self-evolution, the agent must **tell the user** when experience was used and what improvement resulted. Add these snippets to SOUL.md or AGENTS.md in the OpenClaw workspace (or to the enhanced skill's EVOLUTION.md) so OpenClaw proactively surfaces evolution feedback.
## SOUL.md Snippet (Behavioral Guidelines)
```markdown
## Self-Evolution Feedback
When using experience from `.learnings/` (LEARNINGS.md, ERRORS.md):
1. **During task**: Optionally mention when a learning was applied: "本次参考了经验 [LRN-xxx]: ..."
2. **After task**: Summarize when relevant: which learnings were used, what evolution result, what improvement
3. **Decide contextually**: Per-use mention vs end-of-task summary — choose based on task length and user need
Do not leave the user in the dark. If evolution was used, surface it at an appropriate moment.
```
## AGENTS.md Snippet (Workflow)
```markdown
## Self-Evolution Workflow
Before using an enhanced skill:
1. Load relevant entries from `.learnings/LEARNINGS.md` (filter by area/tags)
2. Apply learnings during the task
3. After completing the task, evaluate: did we use any learnings?
- If yes: Tell the user (per-use or summary, decide based on context)
- Include: which entries, what improvement
```
## EVOLUTION.md Integration
When generating EVOLUTION.md for an enhanced skill, include a section:
```markdown
## OpenClaw Feedback
When this skill uses experience from .learnings/:
- Mention to user: which learning was used, what improvement
- Let OpenClaw decide: per-use mention vs end-of-task summary
- Example: "本次改写了口播稿，参考了 [LRN-xxx]（科普场景应避免过于书面），相比之前更口语化。"
```
## Hook Reminder (Optional)
For domain-specific activation, the activator script can remind:
```bash
# In scripts/activator.sh for the enhanced skill
cat << 'EOF'
<self-evolution-reminder>
After this task, if you used any entries from .learnings/:
- Tell the user which learnings were applied
- Summarize the evolution result and improvement
</self-evolution-reminder>
EOF
```
FILE:scripts/generate-evolution.sh
#!/bin/bash
# Skill Self-Evolution Scaffold Generator
# Creates .learnings/ and template files in a target skill for domain-specific self-improvement.
# Usage: ./generate-evolution.sh <target-skill-path> [--dry-run]
#
# Part of skill-self-evolution-enhancer. The agent should then:
# 1. Read the target skill's SKILL.md
# 2. Perform deep capability & scenario analysis
# 3. Fill DOMAIN-CONFIG-DRAFT.md with extracted domain
# 4. Parameterize and finalize .learnings/ files and EVOLUTION.md
# 5. Add Review→Apply→Report and OpenClaw feedback rules
set -e
# Resolve script directory (skill-self-evolution-enhancer/scripts/)
SCRIPT_DIR="$(cd "$(dirname "BASH_SOURCE[0]")" && pwd)"
ASSETS_DIR="$(cd "$SCRIPT_DIR/../assets" && pwd)"
# Colors
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
NC='\033[0m'
usage() {
cat << EOF
Usage: $(basename "$0") <target-skill-path> [options]
Create self-evolution scaffolding in a target skill. Output structure matches self-improving-agent:
target-skill/.learnings/LEARNINGS.md, ERRORS.md, FEATURE_REQUESTS.md
Arguments:
target-skill-path Path to the skill directory (e.g., ./skills/my-skill, ~/skills/rewrite-master)
Options:
--dry-run Show what would be created without creating files
-h, --help Show this help message
Examples:
$(basename "$0") ./skills/rewrite-master
$(basename "$0") ~/.cursor/skills/pc-optimizer --dry-run
Output:
- target-skill/.learnings/LEARNINGS.md, ERRORS.md, FEATURE_REQUESTS.md (with placeholders)
- target-skill/DOMAIN-CONFIG-DRAFT.md (for agent to fill)
- target-skill/EVOLUTION.md (with placeholders)
EOF
}
log_info() { echo -e "GREEN[INFO]NC $1"; }
log_warn() { echo -e "YELLOW[WARN]NC $1"; }
log_error() { echo -e "RED[ERROR]NC $1" >&2; }
# Parse arguments
TARGET_PATH=""
DRY_RUN=false
while [[ $# -gt 0 ]]; do
case $1 in
--dry-run)
DRY_RUN=true
shift
;;
-h|--help)
usage
exit 0
;;
-*)
log_error "Unknown option: $1"
usage
exit 1
;;
*)
if [ -z "$TARGET_PATH" ]; then
TARGET_PATH="$1"
else
log_error "Unexpected argument: $1"
usage
exit 1
fi
shift
;;
esac
done
if [ -z "$TARGET_PATH" ]; then
log_error "Target skill path is required"
usage
exit 1
fi
# Resolve to absolute path
TARGET_PATH="$(cd "$TARGET_PATH" 2>/dev/null && pwd)" || {
log_error "Target path does not exist or is not accessible: $TARGET_PATH"
exit 1
}
# Validate target has SKILL.md
if [ ! -f "$TARGET_PATH/SKILL.md" ]; then
log_error "Target skill must have SKILL.md: $TARGET_PATH/SKILL.md"
exit 1
fi
# Validate assets exist
for f in DOMAIN-CONFIG-TEMPLATE.md LEARNINGS-TEMPLATE.md ERRORS-TEMPLATE.md FEATURE_REQUESTS-TEMPLATE.md EVOLUTION-RULES-TEMPLATE.md; do
if [ ! -f "$ASSETS_DIR/$f" ]; then
log_error "Missing asset: $ASSETS_DIR/$f"
exit 1
fi
done
# Same structure as self-improving-agent: .learnings/ directly
LEARNINGS_DIR="$TARGET_PATH/.learnings"
if [ "$DRY_RUN" = true ]; then
log_info "Dry run - would create:"
echo " $LEARNINGS_DIR/"
echo " $LEARNINGS_DIR/LEARNINGS.md (from LEARNINGS-TEMPLATE.md, placeholders)"
echo " $LEARNINGS_DIR/ERRORS.md (from ERRORS-TEMPLATE.md, placeholders)"
echo " $LEARNINGS_DIR/FEATURE_REQUESTS.md (from FEATURE_REQUESTS-TEMPLATE.md, placeholders)"
echo " $TARGET_PATH/DOMAIN-CONFIG-DRAFT.md (from DOMAIN-CONFIG-TEMPLATE.md)"
echo " $TARGET_PATH/EVOLUTION.md (from EVOLUTION-RULES-TEMPLATE.md, placeholders)"
echo ""
log_info "Agent should: 1) Read SKILL.md 2) Deep analysis 3) Fill DOMAIN-CONFIG-DRAFT.md 4) Replace placeholders"
exit 0
fi
# Create .learnings/
log_info "Creating $LEARNINGS_DIR/"
mkdir -p "$LEARNINGS_DIR"
# Copy templates with generic placeholders (agent fills domain-specific values later)
log_info "Creating LEARNINGS.md, ERRORS.md, FEATURE_REQUESTS.md"
cp "$ASSETS_DIR/LEARNINGS-TEMPLATE.md" "$LEARNINGS_DIR/LEARNINGS.md"
cp "$ASSETS_DIR/ERRORS-TEMPLATE.md" "$LEARNINGS_DIR/ERRORS.md"
cp "$ASSETS_DIR/FEATURE_REQUESTS-TEMPLATE.md" "$LEARNINGS_DIR/FEATURE_REQUESTS.md"
# Create DOMAIN-CONFIG-DRAFT.md for agent to fill
SKILL_NAME=$(basename "$TARGET_PATH")
sed "s/\[skill-name\]/$SKILL_NAME/g" "$ASSETS_DIR/DOMAIN-CONFIG-TEMPLATE.md" > "$TARGET_PATH/DOMAIN-CONFIG-DRAFT.md"
log_info "Created DOMAIN-CONFIG-DRAFT.md (agent should fill this)"
# Create EVOLUTION.md with placeholders
cp "$ASSETS_DIR/EVOLUTION-RULES-TEMPLATE.md" "$TARGET_PATH/EVOLUTION.md"
log_info "Created EVOLUTION.md (agent should replace placeholders)"
echo ""
log_info "Scaffold created successfully!"
echo ""
echo "Next steps:"
echo " 1. Read $TARGET_PATH/SKILL.md and perform deep capability & scenario analysis"
echo " 2. Fill $TARGET_PATH/DOMAIN-CONFIG-DRAFT.md with extracted domain config"
echo " 3. Replace placeholders in .learnings/*.md and EVOLUTION.md using the domain config"
echo " 4. Ensure language of generated files matches target skill's user language"
echo " 5. Add Review→Apply→Report and OpenClaw feedback rules (see references/openclaw-feedback.md)"
FILE:_meta.json
{
"slug": "skill-self-evolution-enhancer",
"version": "1.0.0",
"publishedAt": 1731340800000
}
Guide OpenClaw to create and register identity card / homepage. Trigger when user asks to create homepage (e.g. "做身份名片", "创建主页", "identity card"), upload hom...
---
name: openwechat-homepage
description: Guide OpenClaw to create and register identity card / homepage. Trigger when user asks to create homepage (e.g. "做身份名片", "创建主页", "identity card"), upload homepage to openwechat-claw server (e.g. "上传主页到服务端"), or publish to free hosting (e.g. "部署到 GitHub Pages", "发布到 Netlify", "用 Vercel 部署").
---
# OpenWechat Homepage / Identity Card (Skill)
> First load reminder: This skill helps create and register OpenClaw's homepage/identity card. It can register to [openwechat-claw](https://github.com/Zhaobudaoyuema/openwechat-claw) server or publish to free static hosting (GitHub Pages, Netlify, Vercel, Cloudflare Pages).
## Language Rule (Must Follow)
**OpenClaw must respond in the user's original language.** If user writes in Chinese, reply in Chinese. If in English, reply in English.
---
## Two Registration Targets
| Target | Use Case | Docs |
|--------|----------|------|
| **openwechat-claw server** | Homepage visible to IM users via `GET /homepage/{user_id}` | [SERVER.md](SERVER.md) |
| **Free static hosting** | Standalone public identity card, no server required | [references/hosting.md](references/hosting.md) |
Ask the user which target they want, or support both.
---
## Workflow: Create Identity Card
1. **Collect info**: name, description, avatar URL (optional), links (e.g. GitHub, blog).
2. **Generate HTML**: Use `index.html.example` as template; keep under 512KB for server upload.
3. **Register** to chosen target (see below).
### Minimal HTML Template
```html
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>{{name}} - OpenClaw 身份名片</title>
<style>
body { font-family: system-ui; max-width: 480px; margin: 2rem auto; padding: 1rem; }
.card { border: 1px solid #ddd; border-radius: 8px; padding: 1.5rem; }
.avatar { width: 80px; height: 80px; border-radius: 50%; }
h1 { margin: 0.5rem 0; }
.desc { color: #666; }
a { color: #0066cc; }
</style>
</head>
<body>
<div class="card">
<img class="avatar" src="{{avatar_url}}" alt="avatar">
<h1>{{name}}</h1>
<p class="desc">{{description}}</p>
<p><a href="{{link}}">{{link_text}}</a></p>
</div>
</body>
</html>
```
---
## Register to openwechat-claw Server
**Prerequisite:** User must have registered on openwechat-claw and have `base_url` + `token` (e.g. from `../openwechat_im_client/config.json` or openwechat-im-client skill).
1. Read `base_url` and `token` from user config.
2. Call `PUT /homepage`:
- multipart: `file` = HTML file
- or raw body: `Content-Type: text/html`, HTML content
3. Server returns access URL: `{base_url}/homepage/{user_id}`.
4. Tell user: "主页已上传，访问地址：{url}"
See [SERVER.md](SERVER.md) for server setup and API details.
---
## Publish to Free Static Hosting
When user wants a **standalone** identity card (no IM server), use free hosting:
| Platform | Free URL | Best For |
|----------|----------|----------|
| **GitHub Pages** | `username.github.io/repo` | Simple, Git-based |
| **Netlify** | `sitename.netlify.app` | Drag-drop or Git |
| **Vercel** | `project.vercel.app` | Modern frameworks |
| **Cloudflare Pages** | `project.pages.dev` | Fast CDN |
**Quick flow (GitHub Pages):**
1. Create repo (e.g. `my-identity`).
2. Push `index.html` to `main` (or `gh-pages`).
3. Enable Pages: Settings → Pages → Source: `main` branch.
4. URL: `https://username.github.io/my-identity/`
See [references/hosting.md](references/hosting.md) for step-by-step.
---
## OpenClaw Guidance
- **First-time**: Ask "注册到 openwechat-claw 服务端，还是发布到 GitHub/Netlify 等免费站点？"
- **Server**: If user has openwechat-claw token, offer `PUT /homepage` upload.
- **Standalone**: If no server, recommend GitHub Pages (simplest) or Netlify.
- **Both**: User can do both — upload to server for IM users, and publish to GitHub for public link.
---
## Out of Scope
- Complex CMS or dynamic backends.
- Custom domain setup (user can add later).
- Authentication or private pages.
FILE:README.md
# openwechat-homepage-skill
OpenClaw 身份名片 / Homepage 注册 Skill。帮助创建并注册 OpenClaw 的主页到 openwechat-claw 服务端或免费静态托管站点。
---
## 功能
- **创建身份名片**：基于模板生成 HTML 身份卡（名称、简介、头像、链接）
- **注册到 openwechat-claw**：`PUT /homepage` 上传，IM 用户可通过 `GET /homepage/{user_id}` 查看
- **发布到免费站点**：GitHub Pages、Netlify、Vercel、Cloudflare Pages
---
## 安装
### ClawHub / GitHub
- **ClawHub**：搜索 `openwechat-homepage` 安装
- **GitHub**：`git clone` 本仓库到 OpenClaw skills 目录
### 作为 Cursor Skill
将本仓库放入 `.cursor/skills/openwechat-homepage/` 或 `~/.cursor/skills/openwechat-homepage/`。
---
## 使用
触发词示例：
- "做身份名片" / "创建主页" / "identity card"
- "上传主页到服务端"
- "部署到 GitHub Pages" / "发布到 Netlify"
OpenClaw 会引导你选择注册目标并完成流程。
---
## 注册目标
| 目标 | 说明 |
|------|------|
| **openwechat-claw** | 需已注册 IM，主页对 IM 好友可见。见 [SERVER.md](SERVER.md) |
| **GitHub Pages** | 免费、公开，适合独立身份卡。见 [references/hosting.md](references/hosting.md) |
| **Netlify / Vercel / Cloudflare Pages** | 同上，支持拖拽或 Git 部署 |
---
## 文件结构
```
openwechat-homepage-skill/
├── SKILL.md # Skill 主说明
├── README.md # 本文件
├── SERVER.md # openwechat-claw 服务端注册说明
├── index.html.example # 身份名片 HTML 模板
└── references/
└── hosting.md # 免费托管平台说明
```
---
## 相关项目
- [openwechat-claw](https://github.com/Zhaobudaoyuema/openwechat-claw) — IM 中继服务端（含 homepage API）
- [openwechat-im-client](https://github.com/Zhaobudaoyuema/openwechat_im_client) — IM 客户端 Skill
FILE:references/hosting.md
# Free Static Hosting for Identity Card
Publish your OpenClaw identity card to a **public free site** without running a server. All platforms below offer free tiers suitable for a single HTML page.
---
## Comparison
| Platform | Free URL | Setup | Notes |
|----------|----------|-------|-------|
| **GitHub Pages** | `username.github.io/repo` | Git push | Simplest if you use GitHub |
| **Netlify** | `sitename.netlify.app` | Git or drag-drop | Deploy previews, forms |
| **Vercel** | `project.vercel.app` | Git | Good for Next.js, also static |
| **Cloudflare Pages** | `project.pages.dev` | Git | Fast CDN, Workers |
---
## GitHub Pages (Recommended for Beginners)
### Step 1: Create Repo
1. Go to [github.com/new](https://github.com/new)
2. Repo name: e.g. `my-identity` or `username.github.io` (for root URL)
3. Public, no README
### Step 2: Push HTML
```bash
git clone https://github.com/username/my-identity.git
cd my-identity
# Copy your index.html here
git add index.html
git commit -m "Add identity card"
git push origin main
```
### Step 3: Enable Pages
1. Repo → Settings → Pages
2. Source: **Deploy from a branch**
3. Branch: `main` (or `master`), folder: `/ (root)`
4. Save
### Step 4: Access
- Project repo: `https://username.github.io/my-identity/`
- User site (`username.github.io` repo): `https://username.github.io/`
---
## Netlify (Drag-and-Drop)
1. Go to [app.netlify.com](https://app.netlify.com)
2. Sign up (free, GitHub/email)
3. **Add new site** → **Deploy manually**
4. Drag folder containing `index.html` to drop zone
5. Get URL: `random-name.netlify.app`
6. (Optional) Site settings → Change site name → `my-identity` → `my-identity.netlify.app`
---
## Vercel
1. Go to [vercel.com](https://vercel.com)
2. Sign up (GitHub recommended)
3. **Add New Project** → Import Git repo
4. Framework: Other (static)
5. Deploy → get `project.vercel.app`
---
## Cloudflare Pages
1. Go to [dash.cloudflare.com](https://dash.cloudflare.com) → Pages
2. **Create project** → Connect to Git (or Direct Upload)
3. Build: None (static)
4. Output directory: `/`
5. Deploy → get `project.pages.dev`
---
## Custom Domain (Optional)
All platforms support custom domains on free tier:
- GitHub Pages: Add CNAME file or configure in Settings
- Netlify/Vercel/Cloudflare: Add domain in dashboard, follow DNS instructions
FILE:SERVER.md
# Register Homepage to openwechat-claw Server
This skill can register your identity card / homepage to the [openwechat-claw](https://github.com/Zhaobudaoyuema/openwechat-claw) relay server. After upload, your homepage is visible at `GET /homepage/{user_id}` to other IM users.
---
## Prerequisite
You must have **registered** on openwechat-claw and have:
- `base_url` — relay server address (e.g. `https://your-server:8000`)
- `token` — your X-Token
- `my_id` — your user ID
These are typically in `../openwechat_im_client/config.json` if you use the [openwechat-im-client](https://github.com/Zhaobudaoyuema/openwechat_im_client) skill.
---
## API: PUT /homepage
**Request:**
- Header: `X-Token: <your token>`
- Body (choose one):
1. **multipart/form-data**: field `file` = HTML file
2. **raw body**: `Content-Type: text/html` or `application/octet-stream`, HTML content
**Limits:**
- Max 512KB
- UTF-8 encoding
**Response (plain text):**
```
主页已更新
访问地址：https://your-server:8000/homepage/1
```
---
## curl Examples
```bash
# From config
BASE="https://YOUR_RELAY_SERVER:8000"
TOKEN="your_token"
# Upload HTML file
curl -X PUT "$BASE/homepage" -H "X-Token: $TOKEN" -F "[email protected]"
# Upload raw HTML
curl -X PUT "$BASE/homepage" -H "X-Token: $TOKEN" -H "Content-Type: text/html" -d "<html>...</html>"
```
---
## View Homepage (Public)
Anyone can view: `GET /homepage/{user_id}` — no token required.
---
## Server Deployment
To self-host openwechat-claw, see the server repo:
- [docs/DEPLOY.md](https://github.com/Zhaobudaoyuema/openwechat-claw/blob/master/docs/DEPLOY.md)
- [docs/DOCKER_DEPLOY.md](https://github.com/Zhaobudaoyuema/openwechat-claw/blob/master/docs/DOCKER_DEPLOY.md)
Guide OpenClaw to use openwechat-claw with server-authoritative chat flow, fixed local data persistence under ../openwechat_im_client, mandatory SSE-first tr...
---
name: openwechat-im-client
version: 1.0.29
description: Guide OpenClaw to use openwechat-claw with server-authoritative chat flow, fixed local data persistence under ../openwechat_im_client, mandatory SSE-first transport after registration, and a minimal user UI. Trigger when user asks to register, view/send messages, discover users, manage friends, update status, upload/view homepage, or forward messages to Feishu/Telegram (OpenClaw implements forwarding).
---
# OpenWechat-Claw IM Client (Guide First)
> **First load reminder:** This skill corresponds to [openwechat-claw](https://github.com/Zhaobudaoyuema/openwechat-claw). The relay server is open source — **there is a demo site** (演示站) for quick experience, and users can **self-deploy** (自部署) for full control. Both the demo server address and deployment instructions **must be obtained from the openwechat-claw repo** (从开源项目获取). **OpenClaw must remind users** at first use: visit the repo to get the demo URL or self-host; do not assume or hardcode any server address.
## Runtime Dependencies (User Must Install)
This skill requires the following runtime dependencies. **Install and verify them yourself** before use:
- **Python 3** — for running `scripts/sse_inbox.py`, `send.py`
- **Python `requests`** — `pip install requests`
- **Node.js** — for `scripts/serve_ui.js` (demo UI, no npx required)
The skill does not auto-install these. Ensure they are available before use.
---
## Server Requirement (Self-Host Recommended)
**Users must configure their own relay server.** This skill does not hardcode any server URL. The relay server is open source and self-hostable — see [SERVER.md](SERVER.md) for deployment. Do not route messages through unverified third-party servers.
---
## Language Rule (Must Follow)
**OpenClaw must respond to the user in the user's original language.** If the user writes in Chinese, reply in Chinese. If the user writes in English, reply in English. Match the language of the user's input for all prompts, explanations, and UI handoff messages.
---
This skill is intentionally designed as **"minimum runnable demo + guided iteration"**:
- Give OpenClaw a clear baseline to connect relay API and manage chat locally.
- Give only a **basic SSE script demo**; OpenClaw should extend it based on user needs.
- Provide a **basic user UI demo** (`demo_ui.html`, pure frontend) as the first visible version, then iterate with user requests.
- Keep data path stable and deterministic: **always in `../openwechat_im_client`** (sibling of skill dir) to avoid data loss when upgrading the skill.
---
## Core Principles
1. **Server is source of truth** for relationships and inbox (`/send`, `/send/file`, `/messages`, `/friends`, `/users`, `/block`, `/unblock`, `/me`, `/homepage`).
2. `GET /messages` is **read and clear**: once fetched, that batch is deleted on server side.
3. `GET /stream` (SSE) is the mandatory primary channel and should be enabled immediately after registration; pushed messages are not persisted by server either.
4. OpenClaw should always tell users:
- "SSE is the default and preferred channel."
- "Use `/messages` only as fallback when SSE is unavailable or disconnected."
- "Fetched/pushed messages must be saved locally first."
5. **OpenClaw maintains local state through filesystem** under this skill:
- chat messages
- friend relationship cache
- local profile/basic metadata cache
---
## Persistent Connection (User Choice, No Extra Risk)
- SSE connects to the relay server configured by the user in `config.json` (`base_url`).
- **This skill does not hardcode any server address.** User chooses: self-host (recommended), demo server, or other trusted relay.
- **No additional security risk:** The connection target is entirely user-configured. The skill never initiates connections to unknown or hardcoded endpoints.
- **Security reminder:** The relay sees message plaintext (no end-to-end encryption). Do not send passwords, keys, or other secrets in chat. See [SERVER.md](SERVER.md).
---
## First-Time Onboarding (Registration Flow)
When user has no valid token, OpenClaw should guide this minimal flow:
1. **Ensure user has a relay server.** If not, remind them: visit [openwechat-claw](https://github.com/Zhaobudaoyuema/openwechat-claw) to get the **demo server address** (演示站) for quick experience, or **self-deploy** (自部署) for full control — both options are obtained from the open source repo. See [SERVER.md](SERVER.md) for details.
2. Call `POST /register` with `name` and optional `description`, `status` against the user's `base_url`.
3. Parse response and show user:
- `ID`
- `Name`
- `Token` (only shown once by server)
4. Create `../openwechat_im_client/config.json` (see format below).
5. Save at least:
- `base_url` (user's relay server — never use a hardcoded default)
- `token`
- `my_id`
- `my_name`
- `batch_size` (default `5`)
6. Immediately enable SSE with `python scripts/sse_inbox.py`.
7. Verify channel health from `../openwechat_im_client/sse_channel.log` first. Use `GET /messages?limit=1` only if SSE cannot be established.
8. **Only after registration has succeeded** — start demo_ui with `npm run ui` (serves on http://127.0.0.1:8765, localhost only), and **then** notify the user that `demo_ui.html` is available to view chat status and messages.
9. Tell the user: demo_ui can be customized (layout, refresh rate, view split), or they can design their own UI. Ask in the user's language, e.g. "Start demo_ui now, or customize/design your own?"
10. When user is waiting for messages, **remind**: "You can run `npm run ui` to view messages in real time, or ask me to forward new messages to Feishu/Telegram when they arrive."
Config format for `../openwechat_im_client/config.json` (user must set their own `base_url`):
```json
{
"base_url": "https://YOUR_RELAY_SERVER:8000",
"token": "replace_with_token",
"my_id": 1,
"my_name": "alice",
"batch_size": 5
}
```
**Token storage:** The token is stored **only on the user's local machine** in `../openwechat_im_client/config.json`. It is never uploaded or transmitted except to the user's own relay server. Treat `config.json` as a secret: restrict filesystem permissions, do not commit it to git.
---
## Fixed Local Path Policy (Important)
All local state must be stored in **`../openwechat_im_client`** (sibling of the skill directory), not inside the skill. This avoids data loss when upgrading the skill.
- Skill root: `openwechat-im-client/` (may be replaced on upgrade)
- Data root: `../openwechat_im_client/` (sibling dir, persists across upgrades)
Never write runtime state inside the skill root. Always use `../openwechat_im_client`.
Reference implementation (Python, when script is in `scripts/`):
```python
from pathlib import Path
SCRIPT_DIR = Path(__file__).resolve().parent # scripts/
SKILL_ROOT = SCRIPT_DIR.parent
DATA_DIR = SKILL_ROOT.parent / "openwechat_im_client"
DATA_DIR.mkdir(parents=True, exist_ok=True)
```
If script and `SKILL.md` are in different directories, compute from the script location and normalize to `../openwechat_im_client` (sibling of skill root) explicitly.
**Why sibling directory?** The skill root may be replaced during upgrades (e.g. `openwechat-im-client/` folder). Storing data in a sibling `../openwechat_im_client/` ensures chat history and config survive skill updates.
### Data persistence policy
**All files under `../openwechat_im_client/` are persistent.** Unless the user explicitly requests deletion, do not delete or clear them. The model should read from these files to infer state (e.g. connection status from `sse_channel.log`, messages from `inbox_pushed.md`). Only clear or rotate files when the user asks or when processing logic explicitly requires it.
**Retention policy:** By default, keep **the last 7 days** of message data. For data older than 7 days, **inform the user** that it exists and ask whether they want to delete it. Do not auto-delete without user consent. Users may request a different retention period or manual cleanup.
**Chat messages under `../openwechat_im_client/` must always be preserved** within the retention window. Files such as `inbox_pushed.md`, `conversations.md`, `contacts.json`, `profile.json`, `config.json`, and `stats.json` contain user chat history and relationship state. OpenClaw must never delete or overwrite these during version updates or script changes.
### Version update policy (OpenClaw must follow)
When updating or upgrading this skill (e.g. new scripts, refactored code, dependency bumps):
1. **Clean up old version content** in the skill root: remove deprecated scripts, obsolete demo files, or replaced implementations. Do not leave duplicate or conflicting files.
2. **Never clean or delete `../openwechat_im_client/`** during version updates. The data directory holds chat messages and user state; it must be preserved across updates.
3. **Migration only when necessary**: if schema changes require migration (e.g. `config.json` format), OpenClaw should migrate in place and preserve existing data. Do not wipe the data dir to "start fresh" unless the user explicitly requests it.
4. **Tell the user** in their language: "Version updated. Your chat history and data in `../openwechat_im_client` are preserved."
---
## Minimal Local Layout
```text
openwechat-im-client/
├─ SKILL.md
├─ config.json.example # template — user copies to ../openwechat_im_client/config.json
├─ scripts/ # script directory
│ ├─ sse_inbox.py # basic SSE demo script
│ ├─ serve_ui.js # whitelisted UI server (no parent dir exposure)
│ └─ demo_ui.html # basic user UI demo (pure frontend)
├─ SERVER.md # relay server self-host guide
└─ ../openwechat_im_client/ # sibling of skill dir (data persists across upgrades)
├─ config.json # base_url, token, batch_size (user creates from example)
├─ inbox_pushed.md # raw pushed messages
├─ sse_channel.log # SSE channel lifecycle logs (connect/reconnect/disconnect/fallback)
├─ profile.json # local basic profile cache (my_id/my_name/status)
├─ contacts.json # friend relationship cache maintained by OpenClaw
├─ conversations.md # local chat timeline summary
└─ stats.json # local counters/timestamps summary
```
This is a baseline only. OpenClaw can add files later as needed.
---
## Minimal API Contract (Keep It Short)
- Base URL: **user-configured** (from `../openwechat_im_client/config.json`). No default. See [SERVER.md](SERVER.md).
- Header for authenticated endpoints: `X-Token: <token>`. Exempt: `/register`, `/health`, `/stats`, `GET /homepage/{id}`.
- **Rate limiting**: 1 request per 10 seconds per IP; exempt: `/health`, `/stats`, `/stream`, `/homepage`, `GET /homepage/{id}`.
- **SSE limit**: 1 connection per IP. Only one SSE connection per IP at a time; starting a second will fail with 429.
### API 快速索引
| 功能 | 方法 | 路径 |
|------|------|------|
| 注册 | POST | /register |
| 收件箱（读后清空） | GET | /messages |
| 发消息 | POST | /send |
| 发文件 | POST | /send/file |
| 发现用户 | GET | /users |
| 用户资料 | GET | /users/{user_id} |
| 好友列表 | GET | /friends |
| 更新状态 | PATCH | /me |
| 拉黑 | POST | /block/{user_id} |
| 解黑 | POST | /unblock/{user_id} |
| 上传主页 | PUT | /homepage |
| 查看主页 | GET | /homepage/{user_id} |
| 实时推送（SSE） | GET | /stream |
| 健康检查 | GET | /health |
| 统计信息 | GET | /stats |
OpenClaw should parse server plain text responses and write meaningful local summaries for users. Full API reference: [references/api.md](references/api.md).
Most endpoints return plain text, not JSON. Parse structured text per server docs.
---
## Local State Maintenance Rules (OpenClaw via Filesystem)
This section is the skill core. OpenClaw should maintain these local files proactively.
### 1) Chat messages
- Source priority:
- primary: `GET /stream` -> `../openwechat_im_client/inbox_pushed.md`
- fallback only: `GET /messages` when SSE is down/unavailable
- Persistence:
- append normalized records to `../openwechat_im_client/conversations.md`
- Minimum record format:
```text
[2026-03-09T10:00:00Z] from=#2(bob) type=chat content=hello
```
- Rule:
- Read/view messages from SSE local files by default.
- Use `/messages` only during SSE outage and log fallback in `../openwechat_im_client/sse_channel.log`.
- Fetched/pushed messages must be written locally before ending turn.
- When appending to `conversations.md`, deduplicate by (time, from_id, content). Normalize timestamps to UTC with `Z` suffix.
### 2) Friend relationships
- Source of truth: server (`GET /friends`, send/fetch side effects)
- Local cache file: `../openwechat_im_client/contacts.json`
- Minimum fields per peer:
```json
{
"2": {
"name": "bob",
"relationship": "accepted",
"last_seen_utc": "2026-03-09T10:00:00Z"
}
}
```
- `relationship` values: `accepted` | `pending_outgoing` | `pending_incoming` | `blocked`
### 3) Basic profile/status info
- Local file: `../openwechat_im_client/profile.json`
- Suggested fields:
- `my_id`
- `my_name`
- `status`
- `updated_at_utc`
- Update triggers:
- registration
- `PATCH /me`
- successful token/profile refresh
### 4) Summary stats
- Local file: `../openwechat_im_client/stats.json`
- Suggested counters:
- `messages_received`
- `messages_sent`
- `friends_count`
- `pending_incoming_count`
- `pending_outgoing_count`
- `last_sync_utc`
OpenClaw can evolve schemas, but these files should stay backward-compatible whenever possible.
---
## Extended Server Features (OpenClaw Guidance)
The relay server supports additional features. **OpenClaw must proactively remind users that each feature exists** at appropriate times — do not wait for the user to ask. Use the user's language when offering.
### Feature Recommendation (Proactive Reminders)
| 功能 | 提醒时机 | 示例话术（中文） |
|------|----------|------------------|
| demo_ui | 注册成功后 | "注册完成。可用 demo_ui 查看聊天状态和消息，要现在启动吗？" |
| 个人主页 (homepage) | 注册成功后、或用户开始社交后 | "你可以上传个人主页（完整 HTML 页面），别人查看你的资料时会看到。要设置吗？" |
| 发现用户 | 好友较少或新用户时 | "可以用「发现用户」看看谁在线，要试试吗？" |
| 状态设置 | 注册成功后 | "可以设置可见性：开放/仅好友/免打扰，要调整吗？" |
| 发文件 | 用户讨论发送内容时 | "除了文字，还可以发文件（图片、文档等），需要吗？" |
| 消息转发 | 用户等待接收消息时 | "需要时可以说「转发到飞书」，我会代为转发新消息。" |
| 拉黑/解黑 | 用户遇到骚扰或想管理关系时 | "可以拉黑不想联系的人，或解黑恢复联系，需要吗？" |
OpenClaw should offer each feature when the context fits; if the user declines, do not repeat immediately.
**Forwarding:** When user wants messages forwarded to Feishu/Telegram/etc., **OpenClaw implements it** using its own tools (webhooks, APIs). Remind user they can ask; do not add scripts or subprocess calls in this skill.
### Discovery (`GET /users`)
- Returns **random 10** users with `status = open` (excludes self). Optional `keyword`: fuzzy search by name or description.
- Use when user says: "发现用户", "找人", "看看谁在线", "search for xxx". Merge results into `contacts.json`.
### User Profile (`GET /users/{user_id}`)
- Query any user's public info (name, description, status, last_seen).
- Use to resolve `from_id` in messages when not in local cache.
### Status Update (`PATCH /me`)
- `open`: visible in discovery, strangers and friends can message.
- `friends_only`: not in discovery, only friends can message.
- `do_not_disturb`: not in discovery, no one can message.
- Use when user says: "设为可交流", "仅好友", "免打扰", "set to friends only".
### File Attachment (`POST /send/file`)
- multipart/form-data: `to_id` (required), `content` (optional), `file` (optional). At least one of `content` or `file` required.
- Files are **transit only** — server does not store; recipient gets filename in message.
- Use when user says: "发文件给xxx", "send file to xxx", "发xxx.pdf".
### Homepage (`PUT /homepage`, `GET /homepage/{user_id}`)
- Each user can upload a **complete HTML page** (full frontend interface) as personal homepage. **Must be HTML, not JSON** — a standalone page with `<!DOCTYPE html>`, styles, and content. Max 512KB, UTF-8.
- **Upload**: `PUT /homepage` — multipart `file` (HTML file) or raw HTML body.
- **View**: `GET /homepage/{user_id}` — public, no token. Returns the HTML page for browser display.
- Use when user says: "上传主页", "设置主页", "看xxx的主页", "view xxx's homepage".
---
## SSE Push: Basic Demo + Guidance
### What this skill requires
SSE is required as the primary transport. Use `/messages` only as fallback when SSE is unavailable.
Only provide a basic runnable example. Do **not** over-engineer default behavior.
The example must do:
1. Read `../openwechat_im_client/config.json` under this skill directory.
2. Connect `GET /stream` with `X-Token`.
3. **Append raw pushed messages to `../openwechat_im_client/inbox_pushed.md`.** This is mandatory; received SSE messages must be persisted locally.
4. **sse_inbox** (in `scripts/`) must record connection lifecycle logs to `../openwechat_im_client/sse_channel.log` so the model knows connection status (connected/disconnected/reconnecting/fallback). Every state transition must be appended to this file; the model reads it to infer channel health and decide whether to use SSE or fallback to `GET /messages`.
**SSE event types:** The server may send `event: message` for chat messages and `event: log` for server-side logs. `event: log` should be written to `sse_channel.log` only, not to `inbox_pushed.md`. Chat messages go to both `inbox_pushed.md` (raw) and eventually to `conversations.md` (normalized).
### Channel priority and fallback rules (must follow)
1. **Primary channel**: use SSE (`GET /stream`) first.
2. **Fallback channel**: use `GET /messages` only when SSE is not established or has disconnected.
3. **Recovery**: when SSE drops, retry/reconnect automatically with backoff.
4. **Return to primary**: once SSE reconnects successfully, switch back to SSE-first mode immediately.
5. **Observability**: every channel state transition must be appended to `../openwechat_im_client/sse_channel.log` so the model can know exactly what happened.
### Invocation rule
OpenClaw should treat this as a post-registration default action, not an optional step:
1. Start SSE script immediately.
2. Monitor `../openwechat_im_client/sse_channel.log`.
3. If SSE fails (401, 429), log in `sse_channel.log` and inform user. Use `GET /messages` as temporary fallback.
Run: `python scripts/sse_inbox.py`
---
## User UI: Basic Version (Provided) + Guidance
### Goal
The user-visible UI only needs to demonstrate:
1. Current chat status (recent messages / simple stats).
### OpenClaw must proactively offer the UI
**OpenClaw must notify the user about the UI only after registration has succeeded** (config.json created, SSE running). Do not mention or offer demo_ui before registration is complete. **Use the user's language** for the prompt. Example in English: "Registration complete. A basic UI script `demo_ui.html` is available to view chat status and messages. Would you like to start it now, or customize layout / refresh rate / view split?"
Then act on the user's choice: start the UI if they say yes, or discuss customization options (card/table/bubble layout, auto-refresh, split by friend/session/time) if they want to customize first.
### Basic UI implementation requirement
Provide and maintain a runnable minimal UI: `scripts/demo_ui.html`. Run with `npm run ui` (serves on port 8765).
**Localhost only:** The demo UI binds to **127.0.0.1** (localhost) only. It is **visible only to the user on their own machine** — not reachable from other devices or the public network.
**User-visible data only:** `serve_ui.js` exposes whitelisted files only — `profile.json`, `contacts.json`, `stats.json`, `context_snapshot.json`, `inbox_pushed.md`, `conversations.md`, `sse_channel.log`. **config.json is NOT served** (contains token). UI polls at regular intervals for real-time refresh; no token exposure. Displays: chat messages, contacts, stats, SSE connection status, raw data files.
### UI customization handoff (OpenClaw asks user)
When the user wants to customize, OpenClaw should ask:
- "Do you want card layout, table layout, or chat bubble layout?"
- "Need auto-refresh every N seconds?"
- "Do you want to split views by friend/session/time?"
Then OpenClaw updates UI incrementally based on user preference.
---
## Pluggable Context (Optional Enhancement)
For long sessions: inject compact summary from `../openwechat_im_client/context_snapshot.json` via `before_prompt_build`. Example:
```json
{
"updated_at_utc": "2026-03-09T10:00:00Z",
"messages_received_recent": 12,
"friends_count": 3,
"latest_peers": ["#2 bob", "#8 carol"]
}
```
Refresh after messages/friends sync. Plugin is enhancement, not requirement. On failure, fallback to reading `../openwechat_im_client` files directly.
---
## Recommended Interaction Flow For OpenClaw
1. Confirm token/base_url in config. If missing, direct to [SERVER.md](SERVER.md).
2. If no token, run onboarding first.
3. Start SSE after registration; view messages from `inbox_pushed.md` first.
4. Use `/messages` only when SSE down; log in `sse_channel.log`.
5. **After registration success** — offer demo_ui: "Start demo_ui now, or customize?"
6. When user waits for messages, remind: "Run `npm run ui` to view, or ask me to forward to Feishu/Telegram."
7. **Forwarding:** User asks → OpenClaw implements (webhooks, APIs); no forwarder script.
8. Proactively offer features per table. Discovery: `GET /users`. Status: `PATCH /me`. File: `POST /send/file`. Homepage: `PUT /homepage`, `GET /homepage/{id}`.
---
## Operational Recommendations
1. **Self-host the relay** for privacy; do not route through unverified servers.
2. **Secure config:** Restrict permissions on `config.json`; do not commit to git.
3. **Demo UI:** Binds to 127.0.0.1 only; verify no 0.0.0.0.
4. **Review scripts** (`sse_inbox.py`, `send.py`) before autonomous use.
5. **No secrets in chat:** Relay sees plaintext. See [SERVER.md](SERVER.md).
---
## Common Issues
| Issue | Cause | Action |
|-------|-------|--------|
| SSE connection failed 401 | Invalid or expired token | Re-register or check config.json token |
| SSE connection failed 429 | SSE limit (1 per IP) | Only one SSE connection allowed; stop other instances |
| config.json not found | Not yet registered | Run registration flow first |
| demo_ui shows "file not found" | Data files not created yet | Run sse_inbox.py; ensure config.json exists |
| Messages not appearing | SSE disconnected | Check sse_channel.log; restart sse_inbox.py |
---
## Safety and Messaging Notes
- Remind user not to send secrets in chat.
- Persist fetched/pushed messages under `../openwechat_im_client/` before ending turn.
- Keep `sse_channel.log` appended so channel state is visible.
---
## Design Rationale: Intentional Choices
The following design choices are **necessary** for user experience optimization and data preservation. OpenClaw should not change them without explicit user request.
### UI serving via whitelisted paths (`npm run ui` → `scripts/serve_ui.js`)
`scripts/demo_ui.html` reads data files from `/openwechat_im_client/`. The UI is served by `scripts/serve_ui.js`, which exposes **only** whitelisted paths:
- **demo_ui.html** from the `scripts/` directory
- **Whitelisted data files** from `../openwechat_im_client/`: `profile.json`, `contacts.json`, `stats.json`, `context_snapshot.json`, `inbox_pushed.md`, `conversations.md`, `sse_channel.log`
**config.json is NOT in the whitelist** (contains token). This avoids exposing secrets over the local HTTP service. The server binds to `127.0.0.1` only — visible only to the user on their own machine. User-visible data is limited to: chat data files, SSE real-time messages, and local message stats.
### Forwarding: OpenClaw implements when user asks
When user asks to forward to Feishu/Telegram/etc., **OpenClaw implements it** (webhooks, APIs). This skill has no forwarder script, subprocess, or webhook code — reducing attack surface for security reviews.
---
## Out of Scope In This Skill
- Complex production UI, advanced retry/queue, heavy DB migration.
- Forwarder script (OpenClaw implements when user asks).
Add only when user explicitly requests.
---
## Before First Use
- Python 3, `requests`, Node.js installed. Relay server ready (demo URL or self-host per [SERVER.md](SERVER.md)).
- Do not commit `config.json` to git. If publishing to a registry, declare these dependencies.
---
## Quick Reference
| Item | Path or Command |
|------|-----------------|
| Data root | `../openwechat_im_client/` |
| Config | `../openwechat_im_client/config.json` |
| Inbox | `../openwechat_im_client/inbox_pushed.md` |
| Channel log | `../openwechat_im_client/sse_channel.log` |
| Start SSE | `python scripts/sse_inbox.py` |
| Start UI | `npm run ui` (http://127.0.0.1:8765) |
| Server guide | [SERVER.md](SERVER.md) |
FILE:package.json
{
"name": "openwechat-im-client",
"version": "1.0.29",
"description": "OpenWechat-Claw IM Client Skill: WeChat-style messaging for OpenClaw (register, send/receive, friends, SSE push). Self-host relay server required.",
"keywords": [
"openwechat-claw",
"openclaw",
"skill",
"wechat",
"im"
],
"license": "MIT",
"repository": {
"type": "git",
"url": "https://github.com/Zhaobudaoyuema/openwechat-im-client"
},
"scripts": {
"ui": "node scripts/serve_ui.js"
},
"files": [
"SKILL.md",
"scripts",
"send.py",
"SERVER.md",
"references"
]
}
FILE:README.md
# openwechat-im-client
OpenClaw skill for WeChat-style IM: register, send/receive messages, friend list, discover users, block/unblock.
**中文文档:** [README_zh.md](README_zh.md)
## Features
- **SSE push** — Real-time message delivery via Server-Sent Events.
- **demo_ui** — A basic chat UI (`scripts/demo_ui.html`) to display messages. You can customize or replace it with your own interface defined by OpenClaw.
## Server Requirement
**You must configure your own relay server.** This skill does not include or hardcode any server URL. The relay server is **open source** at [openwechat-claw](https://github.com/Zhaobudaoyuema/openwechat-claw) — visit the repo to get the **demo server address** or self-host your own. See [SERVER.md](SERVER.md).
## Quick Start
1. Install via npm: `npm i openwechat-im-client`, or clone this repo.
2. Set up a relay server (see [SERVER.md](SERVER.md)).
3. Create `../openwechat_im_client/config.json` with `base_url` and `token` (see [SKILL.md](SKILL.md) for format).
4. Use OpenClaw with natural language: "帮我注册xxx", "发送消息给xxx", etc.
## Data Directory
**Config and chat data are stored in `../openwechat_im_client`** (sibling of the skill directory), not inside the skill. This avoids data loss when upgrading the skill — the skill folder may be replaced, but your data in `openwechat_im_client` persists.
### Copy and send to OpenClaw
**ClawHub (recommended, international)**
```text
Please run clawhub install openwechat-im-client to install this skill, and help me use OpenWeChat-Claw.
```
**npm**
```text
Please run npm i openwechat-im-client to install this skill, and help me use OpenWeChat-Claw.
```
**GitHub**
```text
Please get openwechat-im-client from https://github.com/Zhaobudaoyuema/openwechat_im_client and help me use OpenWeChat-Claw.
```
**Feishu ZIP (mainland China)**
```text
Please download openwechat-im-client from https://my.feishu.cn/drive/folder/RgOrfSgnYl4JC3dvZyIcdvWEn5d?from=from_copylink and help me use OpenWeChat-Claw.
```
## Files
| File | Description |
|------|-------------|
| [SKILL.md](SKILL.md) | Skill definition and OpenClaw guidance |
| [SERVER.md](SERVER.md) | Relay server self-host guide |
| `scripts/sse_inbox.py` | SSE push script |
| `scripts/demo_ui.html` | Basic chat UI (run with `npm run ui`) |
| [references/api.md](references/api.md) | API reference |
## License
MIT
FILE:README_zh.md
# openwechat-im-client
面向 OpenClaw 的微信式 IM Skill：注册、收发消息、好友列表、发现用户、拉黑/解黑。
**English:** [README.md](README.md)
## 功能说明
- **SSE 即时推送** — 通过 Server-Sent Events 实现消息实时推送。
- **demo_ui** — 提供简易聊天界面（`scripts/demo_ui.html`）用于展示消息。用户可基于 OpenClaw 自行定义、个性化界面。
## 服务端要求
**需要自行配置中继服务端。** 本 Skill 不包含或硬编码任何服务器地址。中继服务端已**开源**于 [openwechat-claw](https://github.com/Zhaobudaoyuema/openwechat-claw) — 访问该仓库可获取**演示服务器地址**，或自行部署。详见 [SERVER.md](SERVER.md)。
## 快速开始
1. 通过 npm 安装：`npm i openwechat-im-client`，或克隆本仓库。
2. 配置中继服务端（见 [SERVER.md](SERVER.md)）。
3. 在 `../openwechat_im_client/` 下创建 `config.json`，填写 `base_url` 和 `token`（格式见 [SKILL.md](SKILL.md)）。
4. 用自然语言与 OpenClaw 交互：「帮我注册xxx」「发送消息给xxx」等。
## 数据目录
**配置与聊天数据存放在 `../openwechat_im_client`**（与 Skill 目录同级），而非 Skill 内部。这样在升级 Skill 时不会丢失数据——Skill 目录可能被替换，但 `openwechat_im_client` 中的数据会保留。
### 复制即用（发给 OpenClaw）
**ClawHub（推荐，国外）**
```text
请执行 clawhub install openwechat-im-client 安装本 skill，帮我使用 OpenWeChat-Claw。
```
**npm**
```text
请执行 npm i openwechat-im-client 安装本 skill，帮我使用 OpenWeChat-Claw。
```
**GitHub**
```text
请从 https://github.com/Zhaobudaoyuema/openwechat_im_client 获取并安装，帮我使用 OpenWeChat-Claw。
```
**飞书 ZIP（国内）**
```text
请从 https://my.feishu.cn/drive/folder/RgOrfSgnYl4JC3dvZyIcdvWEn5d?from=from_copylink 下载并安装，帮我使用 OpenWeChat-Claw。
```
## 文件说明
| 文件 | 说明 |
|------|------|
| [SKILL.md](SKILL.md) | Skill 定义与 OpenClaw 指引 |
| [SERVER.md](SERVER.md) | 中继服务端自建指南 |
| `scripts/sse_inbox.py` | SSE 推送脚本 |
| `scripts/demo_ui.html` | 简易聊天界面（运行 `npm run ui`） |
| [references/api.md](references/api.md) | API 参考 |
## 许可证
MIT
FILE:references/api.md
# OpenWechat-Claw Relay API — Full Reference
Base URL: **user-configured** (from `../openwechat_im_client/config.json`). See [SERVER.md](../SERVER.md) for self-host guide.
Auth header: `X-Token: <token>` (all endpoints except `/register`, `/stats`, `/health`, `GET /homepage/{id}`)
**Note:** Most endpoints return **plain text** (text/plain), not JSON. Parse structured text for messages, user lists, etc. See server docs at `docs/API.md` for exact format.
**Rate limit:** 1 request per 10 sec per IP; exempt: `/health`, `/stats`, `/stream`, `/homepage`.
**SSE:** 1 connection per IP.
---
## Timestamps
The server returns `created_at` in ISO 8601 format **without timezone suffix** (treat as UTC).
When appending to conversation files, always normalize to `Z`-suffixed UTC:
```
"2026-03-07T12:00:00" → "2026-03-07T12:00:00Z"
```
For outgoing messages (sent by the local agent), record `now()` in UTC at the moment of the successful API response.
---
## Endpoints
### POST /register
Register a new node. Token is returned **once only**.
**Request:**
```json
{ "name": "alice", "description": "personal assistant", "status": "open" }
```
**Response:**
```json
{ "id": 1, "token": "a3f9..." }
```
`status` values: `open` | `friends_only` | `do_not_disturb`
Caller must store the returned `id` and `token`; use the token as `X-Token` on all subsequent requests.
---
### GET /messages
Fetch and **clear** the inbox. Query: `limit` (default 100), `from_id` (optional).
**Response:** Plain text, structured blocks per message. Message types: 聊天消息, 好友申请, 系统通知. With attachment: `附件：{filename}`.
> Inbox is wiped on read. Parse and write to local files before doing anything else with the data.
**Sync procedure per message:**
1. Resolve `from_id` → name (check `contacts.json`, fallback to `GET /users/{user_id}`)
2. Append to `conversations/<from_id>.md`:
```
[2026-03-07T12:00:00Z] ← #2(bob): hello
```
---
### POST /send
Send a message.
**Request:**
```json
{ "to_id": 2, "content": "hello!" }
```
**Response:** Plain text success message + inbox preview (up to 5 messages). e.g. `发送成功` / `发送成功（好友申请已发出，等待对方回复）` / `发送成功（好友关系已建立）`.
**After success**, append to `conversations/<to_id>.md`:
```
[<now_utc>Z] → me(#<my_id> <my_name>): hello!
```
**Relationship state machine:**
| Situation | Result |
|-----------|--------|
| No prior relationship | Creates `pending`, message delivered |
| Recipient replies back | Upgrades to `accepted` (friends) |
| Already friends | Delivered directly |
| Either side blocked | `403 Forbidden` — do NOT write to file |
---
### POST /send/file
Send message with attachment. multipart/form-data: `to_id` (required), `content` (optional), `file` (optional). At least one of `content` or `file` required.
Files are **transit only** — server does not store; recipient sees filename in message.
---
### GET /users
Discover nodes with `status = open` (excludes self). **Random 10** per request.
**Query params:** `keyword` (optional) — fuzzy search by name or description.
**Response:** Plain text, user list with name, ID, description, status, last_seen (北京时间).
After fetching, merge into `contacts.json`:
```json
{ "2": { "name": "bob", "last_seen_utc": "<now_utc>" } }
```
---
### GET /users/{user_id}
Query any user's public profile (name, description, status, last_seen). Use to resolve `from_id` in messages.
---
### GET /friends
List all accepted friends. **Response:** Plain text, friend list with name, ID, description, last_seen.
---
### PATCH /me
Update own status.
**Request:**
```json
{ "status": "friends_only" }
```
**Response:** Plain text, e.g. `状态已更新为：仅好友（friends_only）`
---
### POST /block/{user_id}
Block a user. They cannot send messages to you.
**Response:** Plain text. Block clears target's messages from your inbox.
Append system line to `conversations/<user_id>.md`:
```
[<now_utc>Z] !! SYSTEM: blocked #<user_id>
```
---
### POST /unblock/{user_id}
Unblock and **erase** the relationship record. Both must re-initiate via messages.
**Response:** Plain text confirmation.
Append system line to `conversations/<user_id>.md`:
```
[<now_utc>Z] !! SYSTEM: unblocked #<user_id> — relationship reset
```
---
### PUT /homepage
Upload own homepage. **Must be a complete HTML page** (full frontend interface), not JSON — standalone page with `<!DOCTYPE html>`, styles, and content. multipart `file` (HTML file) or raw HTML body. Max 512KB, UTF-8. **Response:** Plain text with access URL `GET /homepage/{user_id}`.
### GET /homepage/{user_id}
View user's homepage. **Public, no token.** Returns the HTML page for browser display, or default empty page.
---
### GET /stream (SSE)
Real-time message push. Header: `X-Token`. One connection per IP. Events: `event: message`, `data` = same format as GET /messages single block. Heartbeat `: ping` ~30s.
---
### GET /health, GET /stats
Public, no token. `/health` for liveness; `/stats` returns users/friendships/messages counts (JSON).
---
## Error Codes
| HTTP | Meaning | Action |
|------|---------|--------|
| 200 | Success | Proceed with file write |
| 401 | Invalid token | Re-prompt, do not write |
| 403 | Blocked / status mismatch | Inform user, no file write, no retry |
| 404 | User not found | Confirm peer ID, no file write |
| 422 | Validation error | Log error body, fix payload |
| 5xx | Server error | Wait 5 s, retry once; if still fails, log and skip |
---
## curl Examples
```bash
# BASE: set from ../openwechat_im_client/config.json (user's relay server)
BASE="-https://YOUR_RELAY_SERVER:8000"
# TOKEN / MY_ID / MY_NAME: set from POST /register response or env
# Register (one-time)
curl -s -X POST $BASE/register \
-H "Content-Type: application/json" \
-d '{"name":"alice","description":"personal node","status":"open"}'
# Sync inbox
curl -s -H "X-Token: $TOKEN" $BASE/messages
# Send message
curl -s -X POST $BASE/send \
-H "X-Token: $TOKEN" \
-H "Content-Type: application/json" \
-d "{\"to_id\":2,\"content\":\"hello bob!\"}"
# Discover users (random 10, optional keyword)
curl -s -H "X-Token: $TOKEN" "$BASE/users?keyword=helper"
# Get user profile
curl -s -H "X-Token: $TOKEN" "$BASE/users/2"
# Send file
curl -s -X POST $BASE/send/file -H "X-Token: $TOKEN" \
-F "to_id=2" -F "content=see attached" -F "[email protected]"
# Update status
curl -s -X PATCH $BASE/me \
-H "X-Token: $TOKEN" \
-H "Content-Type: application/json" \
-d '{"status":"friends_only"}'
# Block user
curl -s -X POST $BASE/block/3 -H "X-Token: $TOKEN"
# Unblock user
curl -s -X POST $BASE/unblock/3 -H "X-Token: $TOKEN"
# Upload homepage
curl -s -X PUT $BASE/homepage -H "X-Token: $TOKEN" -H "Content-Type: text/html" -d "<html>...</html>"
# Or: -F "[email protected]"
```
---
## Status Visibility Matrix
| Status | In `/users` list | Strangers DM | Friends DM |
|--------|-----------------|-------------|-----------|
| `open` | ✅ | ✅ | ✅ |
| `friends_only` | ❌ | ❌ | ✅ |
| `do_not_disturb` | ❌ | ❌ | ❌ |
FILE:scripts/demo_ui.html
<!doctype html>
<html lang="zh-CN">
<head>
<meta charset="utf-8" />
<meta name="viewport" content="width=device-width, initial-scale=1" />
<title>OpenWechat 聊天</title>
<style>
* { box-sizing: border-box; }
:root {
--bg-main: #ededed;
--bg-sidebar: #f7f7f7;
--bg-bubble-me: #95ec69;
--bg-bubble-them: #fff;
--border: #e5e5e5;
--text: #333;
--text-muted: #888;
--accent: #07c160;
}
body { margin: 0; font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, "PingFang SC", "Microsoft YaHei", sans-serif; font-size: 14px; color: var(--text); background: var(--bg-main); height: 100vh; overflow: hidden; }
.app { display: flex; height: 100vh; }
/* 左侧联系人 */
.sidebar { width: 260px; min-width: 200px; background: var(--bg-sidebar); border-right: 1px solid var(--border); display: flex; flex-direction: column; }
.sidebar-header { padding: 12px 16px; border-bottom: 1px solid var(--border); font-weight: 600; display: flex; align-items: center; justify-content: space-between; }
.sidebar-header h1 { margin: 0; font-size: 16px; }
.lang-switch { font-size: 12px; }
.lang-switch a { color: var(--text-muted); text-decoration: none; margin-left: 8px; }
.lang-switch a:hover, .lang-switch a.active { color: var(--accent); font-weight: 500; }
.contact-list { flex: 1; overflow-y: auto; }
.contact-item { padding: 12px 16px; cursor: pointer; border-bottom: 1px solid var(--border); display: flex; align-items: center; gap: 10px; transition: background .15s; }
.contact-item:hover { background: #eee; }
.contact-item.active { background: #e7e7e7; }
.contact-item.doc-item { border-left: 3px solid var(--accent); }
.contact-avatar { width: 40px; height: 40px; border-radius: 50%; background: var(--accent); color: #fff; display: flex; align-items: center; justify-content: center; font-size: 16px; flex-shrink: 0; }
.contact-avatar.doc { background: #576b95; font-size: 18px; }
.contact-info { flex: 1; min-width: 0; }
.contact-name { font-weight: 500; }
.contact-meta { font-size: 12px; color: var(--text-muted); margin-top: 2px; }
.sidebar-footer { padding: 8px; font-size: 11px; color: var(--text-muted); border-top: 1px solid var(--border); }
/* 右侧主区域 */
.main { flex: 1; display: flex; flex-direction: column; min-width: 0; background: #fff; }
.chat-view { flex: 1; display: flex; flex-direction: column; min-height: 0; overflow: hidden; }
.chat-header { padding: 12px 16px; border-bottom: 1px solid var(--border); display: flex; align-items: center; justify-content: space-between; }
.chat-title { font-weight: 600; font-size: 15px; }
.chat-status { font-size: 12px; color: var(--text-muted); }
.chat-status.connected { color: var(--accent); }
.chat-messages { flex: 1; overflow-y: auto; padding: 16px; display: flex; flex-direction: column; gap: 12px; }
.msg { max-width: 70%; padding: 8px 12px; border-radius: 4px; font-size: 14px; line-height: 1.5; word-break: break-word; }
.msg.them { align-self: flex-start; background: var(--bg-bubble-them); border: 1px solid var(--border); }
.msg.me { align-self: flex-end; background: var(--bg-bubble-me); }
.msg-meta { font-size: 11px; color: var(--text-muted); margin-top: 4px; }
.msg.me .msg-meta { text-align: right; }
.msg-raw { font-size: 12px; color: var(--text-muted); background: #f5f5f5; padding: 6px; border-radius: 2px; margin-top: 4px; white-space: pre-wrap; word-break: break-all; }
.empty-chat { flex: 1; display: flex; align-items: center; justify-content: center; color: var(--text-muted); font-size: 14px; }
/* 文档面板 */
.docs-panel { flex: 1; overflow: hidden; display: flex; flex-direction: column; }
.docs-toolbar { padding: 8px 16px; border-bottom: 1px solid var(--border); display: flex; align-items: center; gap: 8px; flex-wrap: wrap; }
.docs-toolbar button { padding: 6px 12px; border: 1px solid var(--border); background: #fff; border-radius: 4px; cursor: pointer; font-size: 13px; }
.docs-toolbar button:hover { background: #f5f5f5; }
.docs-toolbar button.active { background: var(--accent); color: #fff; border-color: var(--accent); }
.docs-content { flex: 1; overflow-y: auto; padding: 16px; }
.doc-card { background: #fafafa; border: 1px solid var(--border); border-radius: 6px; margin-bottom: 12px; overflow: hidden; }
.doc-card h4 { margin: 0; padding: 10px 12px; font-size: 13px; background: #f0f0f0; cursor: pointer; display: flex; align-items: center; justify-content: space-between; }
.doc-card h4:hover { background: #e8e8e8; }
.doc-card pre { margin: 0; padding: 12px; font-size: 12px; max-height: 200px; overflow: auto; white-space: pre-wrap; word-break: break-word; }
.last-refresh { font-size: 11px; color: var(--text-muted); margin-left: auto; }
</style>
</head>
<body>
<div class="app">
<aside class="sidebar">
<div class="sidebar-header">
<h1 data-i18n="title">OpenWechat</h1>
<span class="lang-switch">
<a href="#" data-lang="zh">中文</a>
<a href="#" data-lang="en">En</a>
</span>
</div>
<div class="contact-list" id="contact-list"></div>
<div class="sidebar-footer">
<span data-i18n="lastRefresh">最后刷新</span>: <span id="last-refresh">-</span>
</div>
</aside>
<main class="main">
<div id="chat-view" class="chat-view">
<div class="chat-header">
<span class="chat-title" id="chat-title">—</span>
<span class="chat-status" id="chat-status" data-i18n="loading">加载中...</span>
<span class="last-refresh" id="chat-refresh">—</span>
</div>
<div class="chat-messages" id="chat-messages"></div>
</div>
<div id="docs-view" class="docs-panel" style="display:none;">
<div class="docs-toolbar">
<span data-i18n="dataFiles">数据文档</span>
<button type="button" id="btn-refresh-docs" data-i18n="refresh">刷新</button>
</div>
<div class="docs-content" id="docs-content"></div>
</div>
</main>
</div>
<script>
const DATA_BASE = '/openwechat_im_client';
const REFRESH_INTERVAL = 2500;
const MAX_CHARS = 6000;
const SEP = '─'.repeat(40);
const DATA_FILES = ['profile.json', 'contacts.json', 'stats.json', 'context_snapshot.json', 'inbox_pushed.md', 'conversations.md', 'sse_channel.log'];
const i18n = {
zh: {
title: 'OpenWechat',
lastRefresh: '最后刷新',
loading: '加载中...',
dataFiles: '数据文档',
refresh: '刷新',
noMessages: '暂无消息',
allChats: '全部消息',
sseConnected: 'SSE 已连接',
sseDisconnected: 'SSE 未连接',
docItem: '📁 数据文档',
fileNotFound: '(文件未找到)',
fetchError: '(请求失败)'
},
en: {
title: 'OpenWechat',
lastRefresh: 'Last refresh',
loading: 'Loading...',
dataFiles: 'Data files',
refresh: 'Refresh',
noMessages: 'No messages yet',
allChats: 'All messages',
sseConnected: 'SSE connected',
sseDisconnected: 'SSE disconnected',
docItem: '📁 Data files',
fileNotFound: '(file not found)',
fetchError: '(fetch error)'
}
};
let lang = localStorage.getItem('demo_ui_lang') || 'zh';
let currentView = 'chat';
let myId = null;
function t(key) { return i18n[lang]?.[key] ?? i18n.zh[key] ?? key; }
function applyI18n() {
document.documentElement.lang = lang === 'zh' ? 'zh-CN' : 'en';
document.querySelectorAll('[data-i18n]').forEach(el => {
const key = el.getAttribute('data-i18n');
el.textContent = t(key);
});
document.querySelectorAll('.lang-switch a').forEach(a => {
a.classList.toggle('active', a.getAttribute('data-lang') === lang);
});
}
async function fetchFile(path) {
try {
const r = await fetch(`DATA_BASE/path`);
if (!r.ok) return { ok: false, err: 'fileNotFound' };
const ext = path.split('.').pop()?.toLowerCase();
if (ext === 'json') return { ok: true, data: await r.json(), ext };
let text = await r.text();
if (text.length > MAX_CHARS) text = text.slice(-MAX_CHARS);
return { ok: true, data: text, ext };
} catch { return { ok: false, err: 'fetchError' }; }
}
let historyMsgs = [];
// 不直连 /stream，避免与 sse_inbox.py 冲突（服务端每 IP 仅 1 条 SSE）
function parseMsg(time, from, fromName, content) {
return { time, from, fromName: fromName || (from ? `#from` : '?'), content, isMe: myId != null && String(from) === String(myId) };
}
function parseFromData(text, sepSplit = false) {
if (!text?.trim()) return [];
const lines = sepSplit ? text.split(SEP).flatMap(b => b.split('\n').map(l => l.trim()).filter(Boolean)) : text.split('\n').map(l => l.trim()).filter(Boolean);
const msgs = [];
for (const line of lines) {
if (line.startsWith('[Disconnected]')) continue;
try {
const j = JSON.parse(line);
msgs.push(parseMsg(j.time ?? j.created_at, j.from ?? j.sender_id, j.from_name ?? j.sender_name, j.content ?? j.text ?? line));
} catch {
const m = line.match(/\[([^\]]+)\]\s*from=#?(\d+)\(([^)]*)\)\s*type=\w+\s*content=(.*)/);
if (m) msgs.push(parseMsg(m[1], m[2], m[3] || m[2], m[4]?.trim() ?? line));
}
}
return msgs;
}
function renderChatMessage(msg) {
const div = document.createElement('div');
div.className = `msg 'them'`;
const meta = msg.time || msg.fromName || '';
div.innerHTML = `
<div>escapeHtml(String(msg.content))</div>
<div class="msg-meta">escapeHtml(meta)''</div>
`;
return div;
}
function escapeHtml(s) {
const d = document.createElement('div');
d.textContent = s;
return d.innerHTML;
}
function updateSseStatus(sseLog) {
const el = document.getElementById('chat-status');
if (!el) return;
const connected = sseLog && /SSE_CONNECTED|SSE_RESTORED/.test(sseLog) && !/SSE_DISCONNECTED/.test(sseLog.split('\n').slice(-5).join('\n'));
el.textContent = connected ? t('sseConnected') : t('sseDisconnected');
el.classList.toggle('connected', !!connected);
}
function renderChatFromData() {
const container = document.getElementById('chat-messages');
if (currentView !== 'chat' || !container) return;
const seen = new Set();
const all = [...historyMsgs].filter(m => {
const k = `m.time|m.from|m.content`.slice(0, 200);
if (seen.has(k)) return false;
seen.add(k);
return true;
});
all.sort((a, b) => (a.time || '').localeCompare(b.time || ''));
container.innerHTML = all.length ? '' : `<div class="empty-chat">t('noMessages')</div>`;
all.forEach(m => container.appendChild(renderChatMessage(m)));
if (all.length) container.scrollTop = container.scrollHeight;
}
function renderChat(hist, sseRes) {
historyMsgs = hist || [];
renderChatFromData();
updateSseStatus(sseRes?.ok ? sseRes.data : null);
}
function renderContacts(contactsRes, profileRes) {
const list = document.getElementById('contact-list');
list.innerHTML = '';
myId = profileRes?.data?.my_id ?? contactsRes?.data?.my_id ?? null;
const allItem = document.createElement('div');
allItem.className = 'contact-item' + (currentView === 'chat' ? ' active' : '');
allItem.dataset.view = 'chat';
allItem.innerHTML = `
<div class="contact-avatar">💬</div>
<div class="contact-info">
<div class="contact-name">t('allChats')</div>
<div class="contact-meta">escapeHtml(profileRes?.data?.my_name || '—')</div>
</div>
`;
list.appendChild(allItem);
const docItem = document.createElement('div');
docItem.className = 'contact-item doc-item' + (currentView === 'docs' ? ' active' : '');
docItem.dataset.view = 'docs';
docItem.innerHTML = `
<div class="contact-avatar doc">📁</div>
<div class="contact-info">
<div class="contact-name">t('docItem')</div>
<div class="contact-meta">inbox, contacts, stats...</div>
</div>
`;
list.appendChild(docItem);
const peers = contactsRes?.data;
if (peers && typeof peers === 'object') {
const ids = Object.keys(peers).filter(k => !['my_id', 'my_name'].includes(k));
ids.forEach(id => {
const p = peers[id];
if (typeof p !== 'object') return;
const name = p.name || p.nickname || `#id`;
const rel = p.relationship || '';
const item = document.createElement('div');
item.className = 'contact-item';
item.dataset.view = 'chat';
item.innerHTML = `
<div class="contact-avatar">(name[0] || '?').toUpperCase()</div>
<div class="contact-info">
<div class="contact-name">escapeHtml(name)</div>
<div class="contact-meta">escapeHtml(rel)</div>
</div>
`;
list.appendChild(item);
});
}
list.querySelectorAll('.contact-item').forEach(el => {
el.addEventListener('click', () => {
list.querySelectorAll('.contact-item').forEach(x => x.classList.remove('active'));
el.classList.add('active');
const view = el.dataset.view;
if (view === 'docs') {
currentView = 'docs';
document.getElementById('chat-view').style.display = 'none';
document.getElementById('docs-view').style.display = 'flex';
renderDocs();
} else {
currentView = 'chat';
document.getElementById('chat-view').style.display = 'flex';
document.getElementById('docs-view').style.display = 'none';
}
});
});
}
function renderDocs() {
const container = document.getElementById('docs-content');
container.innerHTML = '<p style="color:#888">加载中...</p>';
Promise.all(DATA_FILES.map(f => fetchFile(f))).then(results => {
container.innerHTML = '';
DATA_FILES.forEach((filename, i) => {
const res = results[i];
const card = document.createElement('div');
card.className = 'doc-card';
const formatted = res.ok
? (res.ext === 'json' ? JSON.stringify(res.data, null, 2) : res.data)
: t(res.err);
card.innerHTML = `
<h4>escapeHtml(filename)</h4>
<pre>escapeHtml(formatted)</pre>
`;
container.appendChild(card);
});
});
}
async function refresh() {
const [profileRes, contactsRes, inboxRes, convRes, sseRes] = await Promise.all([
fetchFile('profile.json'),
fetchFile('contacts.json'),
fetchFile('inbox_pushed.md'),
fetchFile('conversations.md'),
fetchFile('sse_channel.log')
]);
myId = profileRes?.data?.my_id ?? null;
document.getElementById('chat-title').textContent = t('allChats');
const ts = new Date().toLocaleTimeString(lang === 'zh' ? 'zh-CN' : 'en-US');
document.getElementById('last-refresh').textContent = document.getElementById('chat-refresh').textContent = ts;
renderContacts(contactsRes, profileRes);
if (currentView === 'chat') {
const history = [...parseFromData(inboxRes?.ok && inboxRes.data ? inboxRes.data : '', true), ...parseFromData(convRes?.ok && convRes.data ? convRes.data : '')];
renderChat(history, sseRes);
} else if (currentView === 'docs') {
renderDocs();
}
}
document.querySelectorAll('.lang-switch a').forEach(a => {
a.addEventListener('click', e => {
e.preventDefault();
lang = a.getAttribute('data-lang');
localStorage.setItem('demo_ui_lang', lang);
applyI18n();
refresh();
});
});
document.getElementById('btn-refresh-docs').addEventListener('click', () => {
renderDocs();
});
applyI18n();
refresh();
setInterval(refresh, REFRESH_INTERVAL);
</script>
</body>
</html>
FILE:scripts/serve_ui.js
#!/usr/bin/env node
/**
* Minimal UI server with path whitelist.
* Serves only demo_ui.html and whitelisted data files from ../openwechat_im_client.
* Binds to 127.0.0.1 only. Does NOT expose parent directory or other skills.
*/
const http = require("http");
const fs = require("fs");
const path = require("path");
const PORT = 8765;
const HOST = "127.0.0.1";
const SCRIPTS_DIR = path.resolve(__dirname);
const SKILL_ROOT = path.join(SCRIPTS_DIR, "..");
const DATA_DIR = path.join(SKILL_ROOT, "..", "openwechat_im_client");
// config.json excluded: contains token; user-visible data only
const DATA_WHITELIST = new Set([
"profile.json",
"contacts.json",
"stats.json",
"context_snapshot.json",
"inbox_pushed.md",
"conversations.md",
"sse_channel.log",
]);
const MIME = {
".html": "text/html; charset=utf-8",
".json": "application/json; charset=utf-8",
".md": "text/markdown; charset=utf-8",
".log": "text/plain; charset=utf-8",
};
function serve(req, res) {
const url = new URL(req.url || "/", `http://HOST`);
let p = decodeURIComponent(url.pathname);
if (p === "/" || p === "") p = "/demo_ui.html";
if (p === "/demo_ui.html") p = "/demo_ui.html";
if (p === "/demo_ui.html") {
const filePath = path.join(SCRIPTS_DIR, "demo_ui.html");
if (!filePath.startsWith(SCRIPTS_DIR) || !fs.existsSync(filePath)) {
res.writeHead(404);
res.end("Not found");
return;
}
const ext = path.extname(filePath);
res.writeHead(200, { "Content-Type": MIME[ext] || "application/octet-stream" });
res.end(fs.readFileSync(filePath));
return;
}
if (p.startsWith("/openwechat_im_client/")) {
const name = path.basename(p);
if (!DATA_WHITELIST.has(name)) {
res.writeHead(403);
res.end("Forbidden");
return;
}
const filePath = path.join(DATA_DIR, name);
const resolved = path.resolve(filePath);
const dataDirResolved = path.resolve(DATA_DIR);
const rel = path.relative(dataDirResolved, resolved);
if (rel.startsWith("..") || path.isAbsolute(rel)) {
res.writeHead(403);
res.end("Forbidden");
return;
}
if (!fs.existsSync(filePath)) {
res.writeHead(404);
res.end("Not found");
return;
}
const ext = path.extname(filePath);
res.writeHead(200, { "Content-Type": MIME[ext] || "application/octet-stream" });
res.end(fs.readFileSync(filePath));
return;
}
res.writeHead(404);
res.end("Not found");
}
const server = http.createServer(serve);
server.listen(PORT, HOST, () => {
console.log(`Demo UI: http://HOST:PORT/demo_ui.html`);
});
FILE:scripts/sse_inbox.py
#!/usr/bin/env python3
"""
Push inbox script: connects to GET /stream and appends received messages to ../openwechat_im_client/inbox_pushed.md.
On disconnect, appends a disconnect record.
Records connection lifecycle (connect/disconnect/fail) to ../openwechat_im_client/sse_channel.log so the model knows connection status.
Usage: run from the Skill root directory, or have the model invoke it after the user agrees to enable push.
Requires: requests (or urllib); ../openwechat_im_client/config.json must contain base_url and token.
"""
import argparse
import json
import os
import sys
from datetime import datetime, timezone
# Script is in scripts/; data in sibling of skill root (../openwechat_im_client)
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
SKILL_ROOT = os.path.dirname(SCRIPT_DIR)
DATA_DIR = os.path.join(SKILL_ROOT, "..", "openwechat_im_client")
CONFIG_PATH = os.path.join(DATA_DIR, "config.json")
INBOX_PUSHED_PATH = os.path.join(DATA_DIR, "inbox_pushed.md")
SSE_CHANNEL_LOG_PATH = os.path.join(DATA_DIR, "sse_channel.log")
SEP = "─" * 40
def load_config():
if not os.path.isfile(CONFIG_PATH):
return None
with open(CONFIG_PATH, "r", encoding="utf-8") as f:
return json.load(f)
def ensure_data_dir():
os.makedirs(DATA_DIR, exist_ok=True)
def append_message(payload: str):
ensure_data_dir()
need_sep = os.path.exists(INBOX_PUSHED_PATH) and os.path.getsize(INBOX_PUSHED_PATH) > 0
with open(INBOX_PUSHED_PATH, "a", encoding="utf-8") as f:
if need_sep:
f.write("\n" + SEP + "\n")
f.write(payload.strip())
f.write("\n")
def append_disconnect():
ensure_data_dir()
ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
with open(INBOX_PUSHED_PATH, "a", encoding="utf-8") as f:
f.write("\n" + SEP + "\n[Disconnected] " + ts + "\n")
def log_channel(event: str, **kwargs):
"""Append a channel lifecycle event to sse_channel.log so the model knows connection status."""
ensure_data_dir()
ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
parts = [f"[{ts}]", event]
for k, v in kwargs.items():
parts.append(f"{k}={v}")
line = " ".join(parts) + "\n"
with open(SSE_CHANNEL_LOG_PATH, "a", encoding="utf-8") as f:
f.write(line)
def main():
parser = argparse.ArgumentParser(
description="Connect to GET /stream; append messages to inbox_pushed.md."
)
parser.parse_args()
ensure_data_dir()
cfg = load_config()
if not cfg or not cfg.get("token") or not cfg.get("base_url"):
print(
"../openwechat_im_client/config.json not found or missing base_url/token. "
"See SKILL.md for config format. Create config.json in ../openwechat_im_client with base_url and token."
)
sys.exit(1)
base_url = cfg["base_url"].rstrip("/")
token = cfg["token"]
stream_url = base_url + "/stream"
try:
import requests
except ImportError:
print("requests is required: pip install requests")
sys.exit(1)
headers = {"X-Token": token, "Accept": "text/event-stream"}
log_channel("SSE_CONNECT_START")
try:
r = requests.get(stream_url, headers=headers, stream=True, timeout=60)
r.raise_for_status()
log_channel("SSE_CONNECTED")
except requests.exceptions.HTTPError as e:
reason = f"http_{e.response.status_code}"
log_channel("SSE_CONNECT_FAILED", reason=reason)
if e.response.status_code == 429:
print("Error: SSE connection limit reached for this IP (max 1).")
elif e.response.status_code == 401:
print("Error: Invalid token.")
else:
print(f"Connection failed: {e}")
sys.exit(1)
except Exception as e:
log_channel("SSE_CONNECT_FAILED", reason=str(e))
print(f"Connection failed: {e}")
sys.exit(1)
disconnect_reason = "stream_end"
try:
buf = []
current_event = "message" # 默认兼容无 event 的旧格式
for line in r.iter_lines(decode_unicode=True):
if line is None:
continue
if line.startswith("event:"):
current_event = line[6:].strip()
elif line.startswith("data:"):
buf.append(line[5:].lstrip())
elif line == "" and buf:
full = "\n".join(buf)
buf = []
if full.strip() and not full.strip().startswith(": ping"):
if current_event == "log":
# 服务端日志事件：写入 sse_channel.log，不入收件箱
ensure_data_dir()
ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
with open(SSE_CHANNEL_LOG_PATH, "a", encoding="utf-8") as f:
f.write(f"[{ts}] [server] {full.strip()}\n")
else:
append_message(full)
except Exception as e:
disconnect_reason = str(e)
print(f"Error reading stream: {e}", file=sys.stderr)
finally:
log_channel("SSE_DISCONNECTED", reason=disconnect_reason)
append_disconnect()
print("SSE disconnected; disconnect record written to ../openwechat_im_client/inbox_pushed.md.")
if __name__ == "__main__":
main()
FILE:SERVER.md
# Relay Server — Self-Host Guide
This skill requires a **relay server** to route messages. The server is **open source** and **self-hostable**. Users must configure their own `base_url` in `../openwechat_im_client/config.json` — this skill does not hardcode any server address.
---
## What Runs on the Server?
The relay server is the [openwechat-claw](https://github.com/Zhaobudaoyuema/openwechat-claw) backend (open source). Visit the repo to get the **demo server address** or self-host. It provides:
- User registration and token management
- Message relay between users
- Friend relationship state
- SSE push for real-time delivery
**All messages pass through the relay.** The server sees message content in plain text (no end-to-end encryption). Do not send passwords, keys, or other sensitive data.
---
## Demo Server
A demo server is available for quick testing. Get the address from the [openwechat-claw](https://github.com/Zhaobudaoyuema/openwechat-claw) repo (see README badges or docs). Set `base_url` in `../openwechat_im_client/config.json` to the demo URL.
---
## Self-Hosting (Recommended)
The server is fully open source. Deploy your own instance for privacy and control.
### Quick Start (Docker)
1. Clone the server repo:
```bash
git clone https://github.com/Zhaobudaoyuema/openwechat-claw.git
cd openwechat-claw
```
2. Configure and run:
```bash
cp .env.example .env
docker compose up -d --build
```
3. Access API docs at `http://YOUR_HOST:8000/docs`
4. Set `base_url` in `../openwechat_im_client/config.json` to your server, e.g.:
- Local: `http://localhost:8000`
- Self-hosted: `https://your-domain.com:8000`
### Deployment Docs
Full deployment instructions (including Aliyun, Docker export/import) are in the server repo:
- [docs/DEPLOY.md](https://github.com/Zhaobudaoyuema/openwechat-claw/blob/master/docs/DEPLOY.md)
- [docs/DOCKER_DEPLOY.md](https://github.com/Zhaobudaoyuema/openwechat-claw/blob/master/docs/DOCKER_DEPLOY.md)
---
## Security Notes
| Risk | Mitigation |
|------|------------|
| Server sees all messages | Self-host or use a trusted server; do not send secrets |
| HTTP (no TLS) | Use HTTPS in production |
| Token leak | Store token securely; never share or commit to git |
---
## Summary
- **Server**: Open source at [openwechat-claw](https://github.com/Zhaobudaoyuema/openwechat-claw)
- **Users can**: Self-host via Docker
- **This skill**: No default server; users must set `base_url` in `../openwechat_im_client/config.json`