PaleBlueDot AI

---
name: tokenrouter-image-generator
description: Generate or edit images through Palebluedot Ai(PBD)-TokenRouter's multimodal image generation endpoint (`/v1/chat/completions`) using TokenRouter-compatible image models. Use for text-to-image or image-to-image requests when the user wants TokenRouter, `PBD_TOKENROUTER_API_KEY`, model overrides, or provider-specific `image_config` options.
---

# TokenRouter Image Generation & Editing

Generate new images or edit existing ones using TokenRouter image-capable models via the Chat Completions API.

## Usage

Run the script using absolute path (do NOT cd to the skill directory first):

**Generate new image:**
```bash
# Ensure outbound directory exists first
mkdir -p ~/.openclaw/media/outbound

uv run ~/.openclaw/workspace/skills/pbd-tokenrouter-image-generator/scripts/generate_image.py \
  --prompt "your image description" \
  --filename "~/.openclaw/media/outbound/output-name.png" \
  --model google/gemini-2.5-flash-image \
  [--aspect-ratio 16:9] \
  [--image-size 2K]
```

**Edit existing image (image-to-image):**
```bash
mkdir -p ~/.openclaw/media/outbound

uv run ~/.openclaw/workspace/skills/pbd-tokenrouter-image-generator/scripts/generate_image.py \
  --prompt "editing instructions" \
  --filename "~/.openclaw/media/outbound/output-name.png" \
  --input-image "path/to/input.png" \
  --model google/gemini-2.5-flash-image
```

**Important:** Default OpenClaw delivery path is `~/.openclaw/media/outbound/`. Save generated images there so other OpenClaw flows can pick them up easily.

## API Key

The script requires a TokenRouter API key. Check for it in this order:
1. `--api-key` argument
2. `PBD_TOKENROUTER_API_KEY` environment variable
3. **Auto-detect** from the current agent's tokenrouter provider config in `~/.openclaw/openclaw.json`

### If no API key is found

**Before running the script**, check whether the key is available:

```bash
test -n "$PBD_TOKENROUTER_API_KEY" && echo "Key found" || echo "No key"
```

If none of the above sources provides a key, **do NOT run the script**. Instead, guide the user through obtaining a key:

1. Tell the user they need a TokenRouter API key.
2. Direct them to open: **https://www.tokenrouter.com**
3. Walk them through the steps:
   - Register or sign in on the website
   - After login, navigate to the **API Keys** section
   - Find the **API Keys** menu in the sidebar/navigation
   - Click **API Keys** to enter the key management page
   - Create a new API key and copy it
4. Once the user has the key, offer two options:
   - **Option A — Provide the key directly to the agent:** The user can paste the key in the chat, and the agent passes it to the script via `--api-key`. This is the quickest way to get started — no environment setup needed.
   - **Option B — Configure as environment variable (persistent):**
     - **For this session only:** `export PBD_TOKENROUTER_API_KEY="sk-xxx..."` (paste in terminal)
     - **Persistent (zsh):** Add the export line to `~/.zshrc` then `source ~/.zshrc`
     - **Persistent (bash):** Add the export line to `~/.bashrc` then `source ~/.bashrc`
5. After the key is available, proceed with the image generation command.

**Important:** Never skip this check. Running without a valid key will fail with `Error: No API key provided` or `HTTP 401/403`.

### Optional attribution headers(optional)
- `--site-url` or `PBD_TOKENROUTER_SITE_URL`
- `--app-name` or `PBD_TOKENROUTER_APP_NAME`

## Model + Image Config

- `--model <tokenrouter-model-id>` is required (no script default)
- **Default / recommended:** `google/gemini-2.5-flash-image`
- **Other supported image models** (user can request a switch):
  - `openai/gpt-5-image`
  - `openai/gpt-5-image-mini`
  - `google/gemini-3-pro-image-preview`
  - `google/gemini-3.1-flash-image-preview`
- Use `--aspect-ratio` for `image_config.aspect_ratio` (for example `1:1`, `16:9`)
- Use `--image-size` for `image_config.image_size` (`1K`, `2K`, `4K`)

Note: TokenRouter docs show `aspect_ratio` and `image_size` as the common image config fields for image generation. Additional keys may exist for specific providers/models (for example Sourceful features). If a request fails, remove unsupported options or switch models.

Note: The script always sends `modalities: ["image", "text"]`. Image-only models (some FLUX variants) may reject this — if you get an unexpected error with a non-Gemini model, this may be the cause. No workaround is currently exposed via CLI args.

## Default Workflow (draft -> iterate -> final)

Goal: iterate quickly before spending time on higher-quality settings.

- Draft: smaller size / faster model
  - `--image-size 1K`
- Iterate: adjust prompt in small diffs and keep a new filename each run
- Final: larger size or higher quality if the selected model supports it
  - Example: `--image-size 4K --aspect-ratio 16:9`

## Preflight + Common Failures

- Preflight:
  - `command -v uv`
  - **API key check (CRITICAL):** The script will try `--api-key`, then `PBD_TOKENROUTER_API_KEY`, then auto-read from `~/.openclaw/openclaw.json` (tokenrouter provider). If all fail, **STOP** and guide the user to https://www.tokenrouter.com to register and get a TokenRouter API key (see "If no API key is found" section above)
  - `test -d ~/.openclaw/media/outbound || mkdir -p ~/.openclaw/media/outbound`
  - If editing: `test -f "path/to/input.png"`

- Common failures:
  - `Error: No API key provided.` -> The script could not find a key from `--api-key`, `PBD_TOKENROUTER_API_KEY`, or `~/.openclaw/openclaw.json`. Guide the user to https://www.tokenrouter.com to register and obtain a free TokenRouter API key, then set `PBD_TOKENROUTER_API_KEY` or pass `--api-key`
  - `Error loading input image:` -> bad path or unreadable file
  - `HTTP 400` with model/image config error -> unsupported model or invalid `image_config.aspect_ratio` / `image_config.image_size`
  - `HTTP 401/403` -> invalid key, no model access, or quota/credits issue
  - `No image found in response` -> model may not support image output or request format rejected

## Filename Generation

Generate filenames with the pattern: `~/.openclaw/media/outbound/yyyy-mm-dd-hh-mm-ss-name.png`

Examples:
- `~/.openclaw/media/outbound/2026-03-21-10-00-00-example.png`

## Prompt Handling

- For generation: pass the user's description as-is unless it is too vague to be actionable.
- For editing: make the requested change explicit and preserve everything else.

Prompt template for precise edits:
- `Change ONLY: <change>. Keep identical: subject, composition/crop, pose, lighting, color palette, background, text, and overall style. Do not add new objects.`

## Output

- Save the first returned image to `~/.openclaw/media/outbound/output-name.png` by default (pass that full path in `--filename`)
- Supports TokenRouter's base64 data URL image responses (`message.images[0].image_url.url`)
- Prints the saved file path
- Do not read the image back unless the user asks

## Examples

**Generate new image:**
```bash
mkdir -p ~/.openclaw/media/outbound

uv run ~/.openclaw/workspace/skills/pbd-tokenrouter-image-generator/scripts/generate_image.py \
  --prompt "A field photo with a Japanese-style healing anime aesthetic." \
  --filename "~/.openclaw/media/outbound/2026-03-21-10-00-00-healing.png" \
  --model google/gemini-2.5-flash-image \
  --aspect-ratio 16:9 \
  --image-size 2K
```

**Edit existing image:**
```bash
mkdir -p ~/.openclaw/media/outbound

uv run ~/.openclaw/workspace/skills/pbd-tokenrouter-image-generator/scripts/generate_image.py \
  --prompt "Beautify the photo with a whitening filter." \
  --filename "~/.openclaw/media/outbound/2026-02-26-14-25-30-sunset-sky-edit.png" \
  --model google/gemini-2.5-flash-image \
  --input-image "self-photo.jpg"
```

## Reference

- TokenRouter docs: https://www.tokenrouter.com/docs

FILE:scripts/generate_image.py
#!/usr/bin/env python3
# requires-python = ">=3.10"
"""
Generate or edit images using PBD TokenRouter's image generator via /chat/completions.

Examples:
    uv run generate_image.py --prompt "Sunset at the Seaside" --filename "seaside.png" --aspect-ratio 16:9 --image-size 2K
    uv run generate_image.py --prompt "A dog eating" --filename "dog.png" --input-image "dog.png"
"""

from __future__ import annotations

import argparse
import base64
import json
import mimetypes
import os
import sys
from pathlib import Path
from urllib import error, request


TOKENROUTER_URL = "https://api.tokenrouter.com/v1/chat/completions"
ASPECT_RATIO_CHOICES = ["1:1", "2:3", "3:2", "3:4", "4:3", "4:5", "5:4", "9:16", "16:9", "21:9"]
IMAGE_SIZE_CHOICES = ["1K", "2K", "4K"]


def get_api_key(provided_key: str | None) -> str | None:
    if provided_key:
        return provided_key
    env_key = os.environ.get("PBD_TOKENROUTER_API_KEY")
    if env_key:
        return env_key

    # Fall back to the current agent's tokenrouter provider config
    config_path = Path.home() / ".openclaw" / "openclaw.json"
    if config_path.is_file():
        try:
            config = json.loads(config_path.read_text(encoding="utf-8"))
            providers = config.get("models", {}).get("providers", {})
            for provider_name, provider_config in providers.items():
                if "tokenrouter" in provider_name.lower():
                    api_key = provider_config.get("apiKey")
                    if api_key:
                        return api_key
        except Exception:
            pass

    return None


def load_input_image_data_url(image_path: str) -> str:
    path = Path(image_path)
    if not path.is_file():
        raise FileNotFoundError(f"Input image not found: {image_path}")

    data = path.read_bytes()
    mime_type, _ = mimetypes.guess_type(path.name)
    if not mime_type:
        mime_type = "application/octet-stream"

    encoded = base64.b64encode(data).decode("ascii")
    return f"data:{mime_type};base64,{encoded}"


def build_payload(args: argparse.Namespace) -> dict:
    if args.input_image:
        data_url = load_input_image_data_url(args.input_image)
        content: str | list[dict] = [
            {"type": "text", "text": args.prompt},
            {"type": "image_url", "image_url": {"url": data_url}},
        ]
    else:
        content = args.prompt

    payload: dict = {
        "model": args.model,
        "modalities": ["image", "text"],
        "messages": [{"role": "user", "content": content}],
    }

    image_config: dict = {}
    if args.aspect_ratio:
        image_config["aspect_ratio"] = args.aspect_ratio
    if args.image_size:
        image_config["image_size"] = args.image_size
    if args.image_config_json:
        try:
            extra = json.loads(args.image_config_json)
        except json.JSONDecodeError as exc:
            raise ValueError(f"Invalid --image-config-json: {exc}") from exc
        if not isinstance(extra, dict):
            raise ValueError("--image-config-json must decode to a JSON object")
        image_config.update(extra)

    if image_config:
        payload["image_config"] = image_config

    return payload


def build_headers(args: argparse.Namespace, api_key: str) -> dict[str, str]:
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }

    site_url = args.site_url or os.environ.get("PBD_TOKENROUTER_SITE_URL") or 'no-site'
    app_name = args.app_name or os.environ.get("PBD_TOKENROUTER_APP_NAME") or 'no-app'
    if site_url:
        headers["HTTP-Referer"] = site_url
    if app_name:
        headers["X-Title"] = app_name

    return headers


def extract_text_message(message: dict) -> str | None:
    content = message.get("content")
    if isinstance(content, str) and content.strip():
        return content.strip()

    if isinstance(content, list):
        parts: list[str] = []
        for item in content:
            if not isinstance(item, dict):
                continue
            text = item.get("text")
            if isinstance(text, str) and text.strip():
                parts.append(text.strip())
        if parts:
            return "\n".join(parts)

    return None


def extract_first_image_data(response_json: dict) -> tuple[bytes, str | None]:
    choices = response_json.get("choices")
    if not isinstance(choices, list) or not choices:
        raise ValueError("No choices in response")

    first = choices[0] or {}
    message = first.get("message") or {}
    images = message.get("images")
    if not isinstance(images, list) or not images:
        raise ValueError("No image found in response (message.images missing)")

    first_image = images[0] or {}
    image_url_obj = first_image.get("image_url") or {}
    url = image_url_obj.get("url")
    if not isinstance(url, str) or not url:
        raise ValueError("No image URL found in response image object")

    if not url.startswith("data:"):
        raise ValueError("Image URL is not a data URL; remote URLs are not handled by this script")

    header, _, payload = url.partition(",")
    if not payload or ";base64" not in header:
        raise ValueError("Unsupported data URL format in image response")

    mime_type = None
    if header.startswith("data:"):
        mime_type = header[5:].split(";")[0] or None

    try:
        image_bytes = base64.b64decode(payload)
    except Exception as exc:
        raise ValueError(f"Failed to decode base64 image data: {exc}") from exc

    return image_bytes, mime_type


def api_request(payload: dict, headers: dict[str, str]) -> dict:
    body = json.dumps(payload).encode("utf-8")
    req = request.Request(TOKENROUTER_URL, data=body, headers=headers, method="POST")

    try:
        with request.urlopen(req, timeout=180) as resp:
            raw = resp.read()
    except error.HTTPError as exc:
        details = exc.read().decode("utf-8", errors="replace")
        raise RuntimeError(f"HTTP {exc.code}: {details}") from exc
    except error.URLError as exc:
        raise RuntimeError(f"Network error: {exc}") from exc

    try:
        return json.loads(raw)
    except json.JSONDecodeError as exc:
        text = raw.decode("utf-8", errors="replace")
        raise RuntimeError(f"Invalid JSON response: {text[:500]}") from exc


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Generate or edit images using TokenRouter image generation"
    )
    parser.add_argument("--prompt", "-p", required=True, help="Image prompt or edit instructions")
    parser.add_argument("--filename", "-f", required=True, help="Output filename/path")
    parser.add_argument("--input-image", "-i", help="Optional input image path for editing")
    parser.add_argument("--model", "-m", required=True, help="TokenRouter model ID (required)")
    parser.add_argument(
        "--aspect-ratio",
        choices=ASPECT_RATIO_CHOICES,
        help="image_config.aspect_ratio (e.g. 1:1, 16:9)",
    )
    parser.add_argument(
        "--image-size",
        choices=IMAGE_SIZE_CHOICES,
        help="image_config.image_size (TokenRouter docs: 1K, 2K, 4K)",
    )
    parser.add_argument("--image-config-json", help="JSON object merged into image_config")
    parser.add_argument("--api-key", "-k", help="TokenRouter API key (overrides PBD_TOKENROUTER_API_KEY)")
    parser.add_argument("--site-url", help="Optional HTTP-Referer header (or PBD_TOKENROUTER_SITE_URL env var)")
    parser.add_argument("--app-name", help="Optional X-Title header (or PBD_TOKENROUTER_APP_NAME env var)")

    args = parser.parse_args()

    api_key = get_api_key(args.api_key)
    if not api_key:
        print("Error: No API key provided.", file=sys.stderr)
        print("Please either:", file=sys.stderr)
        print("  1. Provide --api-key argument", file=sys.stderr)
        print("  2. Set PBD_TOKENROUTER_API_KEY environment variable", file=sys.stderr)
        print("  3. Configure a tokenrouter provider in ~/.openclaw/openclaw.json", file=sys.stderr)
        print("You can register and get an API key at: https://www.tokenrouter.com", file=sys.stderr)
        return 1

    try:
        payload = build_payload(args)
    except FileNotFoundError as exc:
        print(f"Error loading input image: {exc}", file=sys.stderr)
        return 1
    except ValueError as exc:
        print(f"Error: {exc}", file=sys.stderr)
        return 1

    headers = build_headers(args, api_key)

    output_path = Path(args.filename)
    output_path.parent.mkdir(parents=True, exist_ok=True)

    mode_label = "Editing image" if args.input_image else "Generating image"
    print(f"{mode_label} via TokenRouter...")
    print(f"Model: {args.model}")
    if "image_config" in payload:
        print(f"Image config: {json.dumps(payload['image_config'])}")

    try:
        response_json = api_request(payload, headers)
    except RuntimeError as exc:
        print(f"Error generating image: {exc}", file=sys.stderr)
        return 1

    choices = response_json.get("choices") or []
    if choices and isinstance(choices[0], dict):
        message = choices[0].get("message") or {}
        text = extract_text_message(message)
        if text:
            print(f"Model response: {text}")

    try:
        image_bytes, mime_type = extract_first_image_data(response_json)
    except ValueError as exc:
        print(f"Error: {exc}", file=sys.stderr)
        return 1

    output_path.write_bytes(image_bytes)
    full_path = output_path.resolve()
    print(f"Image saved: {full_path}")
    if mime_type:
        print(f"Detected MIME type: {mime_type}")

    return 0


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

---
name: tokenrouter-video-usaging
description: >
  Guide the agent to perform tokenrouter channel and model configuration checks for video generation tasks.
  First, check whether the workspace already has a tokenrouter channel whose baseurl contains `https://api.tokenrouter.com` or `https://open.palebluedot.ai`.
  If no such channel exists, stop and instruct the user to visit `https://www.tokenrouter.com` to register and obtain tokenrouter configuration.
  If the channel exists, then check whether the requested video model (`MiniMax-Hailuo-2.3`, `kling-v3`, `kling-v2-6`, `dreamina-seedance-2-0-fast-260128`, or `dreamina-seedance-2-0-260128`) is already configured in the channel.
  If the model is missing, the agent should auto-configure the model route by inferring the existing schema and adding the smallest correct change.
  After confirming both the channel and the model are ready, use the configured route to create and query video generation tasks.
---

# Tokenrouter Video Usaging

## Overview

Find the tokenrouter channel configuration in the current workspace, infer the existing provider/channel schema from neighboring model entries, and add a `MiniMax-Hailuo-2.3`, `kling-v3`, `kling-v2-6`, `dreamina-seedance-2-0-fast-260128`, or `dreamina-seedance-2-0-260128` route with the smallest correct change. Then use the configured tokenrouter endpoint to create a video generation task and fetch task status until a final video URL or terminal state is returned.

## Workflow

1. **Channel Check**: Discover the tokenrouter config and check whether any configured channel has a `baseurl` or `baseURL` containing `https://api.tokenrouter.com` or `https://open.palebluedot.ai`.
   - If **NO** such channel exists, **STOP** immediately and tell the user: "No tokenrouter channel with `https://api.tokenrouter.com` or `https://open.palebluedot.ai` was found. Please visit `https://www.tokenrouter.com` to register and obtain your channel configuration, then add it to the workspace."
   - If a channel **IS** found, proceed to step 2.
2. **Model Check**: Check whether the requested video model (`MiniMax-Hailuo-2.3`, `kling-v3`, `kling-v2-6`, `dreamina-seedance-2-0-fast-260128`, or `dreamina-seedance-2-0-260128`) already exists in the channel's model map or route list.
   - If the model **IS** already configured, skip to step 4.
   - If the model is **NOT** configured, proceed to step 3.
3. **Auto-Configuration**: Infer the existing provider/channel schema from neighboring model entries, then add the missing model route with the smallest correct change.
   - Reuse the existing video provider pattern if one exists; otherwise mirror the closest neighboring provider entry.
   - Save the config change with the smallest possible edit. Do not rewrite unrelated formatting or reorder large sections unless the file format requires it.
   - Reload or restart the tokenrouter service if the workspace provides a command to do so.
4. **API Call**: Use the detected channel's key and the fixed base URL `https://api.tokenrouter.com` to call the video generation endpoints.
5. **Polling**: Poll task status until success, failure, or timeout.

## Config Discovery

Start by searching for likely tokenrouter config files in the current workspace.

- Prefer files matching names like `*tokenrouter*`, `*channel*`, `*provider*`, `*model*`, `config*.json`, `config*.yaml`, `config*.yml`, `*.toml`, `*.ts`, `*.js`.
- Look for keys such as `channels`, `providers`, `models`, `routes`, `baseURL`, `baseurl`, `apiKey`, `key`, `upstream`, `model_map`, `model_name`, or similar.
- Run `scripts/find_tokenrouter_config.py` first when the workspace is large or the config location is unclear.

Choose the config file that is actually consumed by the running tokenrouter setup, not merely documentation or examples.

Authentication discovery rule:

- Search current channel/provider configs for a `baseurl` or `baseURL` containing `https://api.tokenrouter.com` or `https://open.palebluedot.ai`.
- If found, treat that entry as the tokenrouter channel and reuse its configured key for `Authorization`.
- If not found, stop and instruct the user to register at `https://www.tokenrouter.com` to obtain tokenrouter access and the required channel configuration.

## Config Update Rules

Infer the schema from the file instead of assuming a fixed format.

- If the requested model (`MiniMax-Hailuo-2.3`, `kling-v3`, `kling-v2-6`, `dreamina-seedance-2-0-fast-260128`, or `dreamina-seedance-2-0-260128`) already exists, update only missing or incorrect fields.
- If there is an existing video-generation entry (MiniMax, Hailuo, Kling, Seedance, or another video model), copy that shape and adapt it.
- Preserve existing auth conventions. If an existing `https://api.tokenrouter.com` or `https://open.palebluedot.ai` channel is present, reuse that channel's key exactly as configured instead of introducing a new env var name.
- Prefer adding the new model alongside existing channel/provider definitions rather than introducing a second config mechanism.
- Do not invent fallback compatibility fields unless the surrounding config already uses them.

Minimum target behavior:

- The tokenrouter config must recognize the requested model name (`MiniMax-Hailuo-2.3`, `kling-v3`, `kling-v2-6`, `dreamina-seedance-2-0-fast-260128`, or `dreamina-seedance-2-0-260128`).
- Requests for that model must reach the upstream path `POST /v1/video/generations`.
- Status checks must use `GET /video/generations/:task_id`.

If the local tokenrouter schema distinguishes between chat/completions/image/video APIs, make sure this model is wired into the video path rather than a text generation path.

## API Calls

Use the fixed tokenrouter base URL `https://api.tokenrouter.com` for all API calls.

Auth rule:

- Base URL is always `https://api.tokenrouter.com`.
- If a matching channel exists, use that channel's configured key directly.
- If no matching channel exists, do not fabricate placeholders like `TOKENROUTER_API_KEY`; instead tell the user to register at `https://www.tokenrouter.com` and add tokenrouter config first.

### Model: MiniMax-Hailuo-2.3

Create video task:

```json
{
  "path": "/v1/video/generations",
  "method": "POST",
  "params": {
    "model": "MiniMax-Hailuo-2.3",
    "prompt": "A man picks up a book [Pedestal up], then reads [Static shot].",
    "size": "1080P",
    "duration": 6
  }
}
```

`size` and `duration` must be chosen as one of the supported Hailuo combinations:

- `1080P` with `6`
- `768P` with `10`
- `768P` with `6`

Do not mix unsupported combinations. If the user asks for another pair, ask them to choose one of the three valid options.

### Model: kling-v3

Supports both text-to-video and image-to-video.

**Text-to-video:**

```json
{
  "path": "/v1/video/generations",
  "method": "POST",
  "params": {
    "model": "kling-v3",
    "prompt": "A silver robot walking through a rainy neon alley",
    "mode": "pro",
    "duration": "5",
    "metadata": {
      "aspect_ratio": "16:9",
      "sound": "on",
      "negative_prompt": "blurry, low quality"
    }
  }
}
```

**Image-to-video:**

```json
{
  "path": "/v1/video/generations",
  "method": "POST",
  "params": {
    "model": "kling-v3",
    "prompt": "The girl smiles slightly and the camera slowly pushes in",
    "image": "https://example.com/portrait.png",
    "mode": "pro",
    "duration": "5",
    "metadata": {
      "sound": "off",
      "negative_prompt": "flicker, blur"
    }
  }
}
```

`duration` is a string. Common values include `"5"`. `mode` is a string such as `"pro"`. `metadata` is optional but may include `aspect_ratio`, `sound`, and `negative_prompt`. `image` is optional; when present, it triggers image-to-video generation. Do not send Hailuo-only fields (`size`) when using `kling-v3`.

### Model: kling-v2-6

Supports both text-to-video and image-to-video.

**Text-to-video:**

```json
{
  "path": "/v1/video/generations",
  "method": "POST",
  "params": {
    "model": "kling-v2-6",
    "prompt": "A silver robot walking through a rainy neon alley",
    "mode": "pro",
    "duration": "5",
    "metadata": {
      "aspect_ratio": "16:9",
      "sound": "on",
      "negative_prompt": "blurry, low quality"
    }
  }
}
```

**Image-to-video:**

```json
{
  "path": "/v1/video/generations",
  "method": "POST",
  "params": {
    "model": "kling-v2-6",
    "prompt": "The girl smiles slightly and the camera slowly pushes in",
    "image": "https://example.com/portrait.png",
    "mode": "pro",
    "duration": "5",
    "metadata": {
      "sound": "off",
      "negative_prompt": "flicker, blur"
    }
  }
}
```

`duration` is a string. `mode` is a string such as `"pro"`. `metadata` is optional but may include `aspect_ratio`, `sound`, and `negative_prompt`. `image` is optional; when present, it triggers image-to-video generation. Do not send Hailuo-only fields (`size`) when using `kling-v2-6`.

### Model: dreamina-seedance-2-0-fast-260128

Supports both text-to-video and image-to-video.

**Text-to-video:**

```json
{
  "path": "/v1/video/generations",
  "method": "POST",
  "params": {
    "model": "dreamina-seedance-2-0-fast-260128",
    "prompt": "A vintage sports car driving along a coastal road at sunset",
    "metadata": {
      "duration": 5,
      "resolution": "1080p",
      "ratio": "16:9",
      "generate_audio": true
    }
  }
}
```

**Image-to-video:**

```json
{
  "path": "/v1/video/generations",
  "method": "POST",
  "params": {
    "model": "dreamina-seedance-2-0-fast-260128",
    "prompt": "The subject looks up toward the camera while hair moves gently in the wind",
    "images": [
      "https://example.com/input-image.png"
    ],
    "metadata": {
      "duration": 5,
      "resolution": "720p",
      "ratio": "9:16"
    }
  }
}
```

`prompt` is required for text-to-video, optional but recommended for image-to-video. `metadata` is optional but may include `duration`, `resolution`, `ratio`, and `generate_audio`. `images` is optional; when present, it triggers image-to-video generation. Do not send Hailuo-only fields (`size`) or Kling-only fields (`mode`, `image`) when using Seedance models.

### Model: dreamina-seedance-2-0-260128

Supports both text-to-video and image-to-video.

**Text-to-video:**

```json
{
  "path": "/v1/video/generations",
  "method": "POST",
  "params": {
    "model": "dreamina-seedance-2-0-260128",
    "prompt": "A vintage sports car driving along a coastal road at sunset",
    "metadata": {
      "duration": 5,
      "resolution": "1080p",
      "ratio": "16:9",
      "generate_audio": true
    }
  }
}
```

**Image-to-video:**

```json
{
  "path": "/v1/video/generations",
  "method": "POST",
  "params": {
    "model": "dreamina-seedance-2-0-260128",
    "prompt": "The subject looks up toward the camera while hair moves gently in the wind",
    "images": [
      "https://example.com/input-image.png"
    ],
    "metadata": {
      "duration": 5,
      "resolution": "720p",
      "ratio": "9:16"
    }
  }
}
```

`prompt` is required for text-to-video, optional but recommended for image-to-video. `metadata` is optional but may include `duration`, `resolution`, `ratio`, and `generate_audio`. `images` is optional; when present, it triggers image-to-video generation. Do not send Hailuo-only fields (`size`) or Kling-only fields (`mode`, `image`) when using Seedance models.

### Parameter Validation Rule

Before calling the create endpoint, always validate the request according to the chosen model:

**For MiniMax-Hailuo-2.3:**

1. Check that `prompt` is present and is a string.
2. Check that `size` is exactly `"1080P"` or `"768P"`.
3. Check that `duration` is exactly `6` or `10`.
4. Check that the pair belongs to the allowed set: (`1080P`, `6`), (`768P`, `10`), (`768P`, `6`).
5. Ensure `mode`, `image`, `images`, and `metadata` are NOT present (these fields belong to other models).
6. If validation fails, stop and tell the user the allowed combinations instead of sending the request.

**For kling-v3:**

1. Check that `mode` is present and is a string.
2. Check that `duration` is a string value such as `"5"`.
3. Ensure `size`, `images` are NOT present.
4. If `image` is present (image-to-video mode):
   - Check that `image` is a valid URL string.
   - `prompt` is optional but recommended.
   - `metadata.aspect_ratio` should NOT be present.
5. If `image` is NOT present (text-to-video mode):
   - Ensure `prompt` is present and is a string.
6. If validation fails, stop and tell the user the correct kling-v3 parameters instead of sending the request.

**For kling-v2-6:**

1. Check that `mode` is present and is a string.
2. Check that `duration` is a string value such as `"5"`.
3. Ensure `size`, `images` are NOT present.
4. If `image` is present (image-to-video mode):
   - Check that `image` is a valid URL string.
   - `prompt` is optional but recommended.
   - `metadata.aspect_ratio` should NOT be present.
5. If `image` is NOT present (text-to-video mode):
   - Ensure `prompt` is present and is a string.
6. If validation fails, stop and tell the user the correct kling-v2-6 parameters instead of sending the request.

**For dreamina-seedance-2-0-fast-260128:**

1. Ensure `size`, `mode`, `image` are NOT present.
2. If `images` is NOT present (text-to-video mode):
   - Check that `prompt` is present and is a string.
   - `metadata.generate_audio` is optional boolean.
3. If `images` is present (image-to-video mode):
   - Check that `images` is a non-empty array of valid URL strings.
   - `prompt` is optional but recommended.
   - `metadata.generate_audio` should NOT be present.
4. If `metadata.duration` is present, check that it is an integer such as `5`.
5. If validation fails, stop and tell the user the correct Seedance parameters instead of sending the request.

**For dreamina-seedance-2-0-260128:**

1. Ensure `size`, `mode`, `image` are NOT present.
2. If `images` is NOT present (text-to-video mode):
   - Check that `prompt` is present and is a string.
   - `metadata.generate_audio` is optional boolean.
3. If `images` is present (image-to-video mode):
   - Check that `images` is a non-empty array of valid URL strings.
   - `prompt` is optional but recommended.
   - `metadata.generate_audio` should NOT be present.
4. If `metadata.duration` is present, check that it is an integer such as `5`.
5. If validation fails, stop and tell the user the correct Seedance parameters instead of sending the request.

### Fetch task (all models)

```json
{
  "path": "/video/generations/:task_id",
  "method": "GET"
}
```

Implementation notes:

- Treat the create call as asynchronous.
- Extract `task_id` from the create response using the actual response schema returned by the server.
- Poll the fetch endpoint until a terminal state is reached.
- Report the full terminal payload and any resulting video URL(s).
- When extracting video URLs from the response, preserve the complete URL including the query string.
- If the URL contains the escaped sequence `\u0026`, decode it to `&` before presenting or returning the URL.

## Execution Pattern

1. Confirm there is an existing channel whose `baseurl` contains `https://api.tokenrouter.com` or `https://open.palebluedot.ai`.
2. If absent, stop and ask the user to get tokenrouter access from `https://www.tokenrouter.com`.
3. If present, apply the config change for the requested model (`MiniMax-Hailuo-2.3`, `kling-v3`, `kling-v2-6`, `dreamina-seedance-2-0-fast-260128`, or `dreamina-seedance-2-0-260128`) in that channel or its neighboring model map.
4. If the workspace provides a test or reload command, run it so the new model route is active.
5. Submit a generation request with the detected channel key.
6. Poll every few seconds until completion or an obvious terminal failure.

Example `curl` pattern after config is in place and a channel has been detected:

**Hailuo:**

```bash
curl -X POST "https://api.tokenrouter.com/v1/video/generations" \
  -H "Authorization: Bearer $DETECTED_CHANNEL_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "MiniMax-Hailuo-2.3",
    "prompt": "A man picks up a book [Pedestal up], then reads [Static shot].",
    "size": "1080P",
    "duration": 6
  }'
```

**Kling text-to-video:**

```bash
curl -X POST "https://api.tokenrouter.com/v1/video/generations" \
  -H "Authorization: Bearer $DETECTED_CHANNEL_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "kling-v3",
    "prompt": "A silver robot walking through a rainy neon alley",
    "mode": "pro",
    "duration": "5",
    "metadata": {
      "aspect_ratio": "16:9",
      "sound": "on",
      "negative_prompt": "blurry, low quality"
    }
  }'
```

**Kling image-to-video:**

```bash
curl -X POST "https://api.tokenrouter.com/v1/video/generations" \
  -H "Authorization: Bearer $DETECTED_CHANNEL_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "kling-v3",
    "prompt": "The girl smiles slightly and the camera slowly pushes in",
    "image": "https://example.com/portrait.png",
    "mode": "pro",
    "duration": "5",
    "metadata": {
      "sound": "off",
      "negative_prompt": "flicker, blur"
    }
  }'
```

**Seedance text-to-video:**

```bash
curl -X POST "https://api.tokenrouter.com/v1/video/generations" \
  -H "Authorization: Bearer $DETECTED_CHANNEL_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "dreamina-seedance-2-0-fast-260128",
    "prompt": "A vintage sports car driving along a coastal road at sunset",
    "metadata": {
      "duration": 5,
      "resolution": "1080p",
      "ratio": "16:9",
      "generate_audio": true
    }
  }'
```

**Seedance image-to-video:**

```bash
curl -X POST "https://api.tokenrouter.com/v1/video/generations" \
  -H "Authorization: Bearer $DETECTED_CHANNEL_KEY" \
  -H "Content-Type: application/json" \
  -d '{
    "model": "dreamina-seedance-2-0-260128",
    "prompt": "The subject looks up toward the camera while hair moves gently in the wind",
    "images": [
      "https://example.com/input-image.png"
    ],
    "metadata": {
      "duration": 5,
      "resolution": "720p",
      "ratio": "9:16"
    }
  }'
```

Then:

```bash
curl "https://api.tokenrouter.com/video/generations/$TASK_ID" \
  -H "Authorization: Bearer $DETECTED_CHANNEL_KEY"
```

Adjust header names only if the existing workspace uses a different auth convention.

## Resources

Use `references/api_reference.md` for endpoint-specific guidance and `scripts/find_tokenrouter_config.py` to quickly identify likely tokenrouter config files.

### scripts/
`find_tokenrouter_config.py` scans the workspace for files likely to contain tokenrouter channel/provider/model routing config.

### references/
`api_reference.md` captures the two required video endpoints, expected request bodies, and polling guidance.

### assets/
No assets are required for this skill.

FILE:references/api_reference.md
# Tokenrouter Video API Reference

## Purpose

Use this reference after the workspace tokenrouter config has been updated so the requested video model (`MiniMax-Hailuo-2.3`, `kling-v3`, `kling-v2-6`, `dreamina-seedance-2-0-fast-260128`, or `dreamina-seedance-2-0-260128`) is routed correctly.

## Required Endpoints

### Create Video

- Method: `POST`
- Path: `/v1/video/generations`

#### MiniMax-Hailuo-2.3

Request body:

```json
{
  "model": "MiniMax-Hailuo-2.3",
  "prompt": "A man picks up a book [Pedestal up], then reads [Static shot].",
  "size": "1080P",
  "duration": 6
}
```

Supported `size` and `duration` combinations for `MiniMax-Hailuo-2.3`:

- `size: "1080P"` with `duration: 6`
- `size: "768P"` with `duration: 10`
- `size: "768P"` with `duration: 6`

Validation rule:

- Reject or correct any request that uses a different `size` and `duration` pairing.

#### kling-v3

Supports both text-to-video and image-to-video.

**Text-to-video:**

```json
{
  "model": "kling-v3",
  "prompt": "A silver robot walking through a rainy neon alley",
  "mode": "pro",
  "duration": "5",
  "metadata": {
    "aspect_ratio": "16:9",
    "sound": "on",
    "negative_prompt": "blurry, low quality"
  }
}
```

**Image-to-video:**

```json
{
  "model": "kling-v3",
  "prompt": "The girl smiles slightly and the camera slowly pushes in",
  "image": "https://example.com/portrait.png",
  "mode": "pro",
  "duration": "5",
  "metadata": {
    "sound": "off",
    "negative_prompt": "flicker, blur"
  }
}
```

`duration` is a string. Common values include `"5"`. `mode` is a string such as `"pro"`. `metadata` is optional but may include `aspect_ratio`, `sound`, and `negative_prompt`. `image` is optional; when present, it triggers image-to-video generation.

Validation rule:

- `size` must NOT be present for Kling requests.
- `mode` must be present and be a string.
- `duration` must be a string value such as `"5"`.
- If `image` is present, it must be a valid URL string (image-to-video mode).
- If `image` is NOT present, `prompt` must be present (text-to-video mode).

#### kling-v2-6

Supports both text-to-video and image-to-video.

**Text-to-video:**

```json
{
  "model": "kling-v2-6",
  "prompt": "A silver robot walking through a rainy neon alley",
  "mode": "pro",
  "duration": "5",
  "metadata": {
    "aspect_ratio": "16:9",
    "sound": "on",
    "negative_prompt": "blurry, low quality"
  }
}
```

**Image-to-video:**

```json
{
  "model": "kling-v2-6",
  "prompt": "The girl smiles slightly and the camera slowly pushes in",
  "image": "https://example.com/portrait.png",
  "mode": "pro",
  "duration": "5",
  "metadata": {
    "sound": "off",
    "negative_prompt": "flicker, blur"
  }
}
```

`duration` is a string. `mode` is a string such as `"pro"`. `metadata` is optional but may include `aspect_ratio`, `sound`, and `negative_prompt`. `image` is optional; when present, it triggers image-to-video generation.

Validation rule:

- `size` must NOT be present for Kling requests.
- `mode` must be present and be a string.
- `duration` must be a string value such as `"5"`.
- If `image` is present, it must be a valid URL string (image-to-video mode).
- If `image` is NOT present, `prompt` must be present (text-to-video mode).

#### dreamina-seedance-2-0-fast-260128

Supports both text-to-video and image-to-video.

**Text-to-video:**

```json
{
  "model": "dreamina-seedance-2-0-fast-260128",
  "prompt": "A vintage sports car driving along a coastal road at sunset",
  "metadata": {
    "duration": 5,
    "resolution": "1080p",
    "ratio": "16:9",
    "generate_audio": true
  }
}
```

**Image-to-video:**

```json
{
  "model": "dreamina-seedance-2-0-fast-260128",
  "prompt": "The subject looks up toward the camera while hair moves gently in the wind",
  "images": [
    "https://example.com/input-image.png"
  ],
  "metadata": {
    "duration": 5,
    "resolution": "720p",
    "ratio": "9:16"
  }
}
```

`prompt` is required for text-to-video, optional but recommended for image-to-video. `metadata` is optional but may include `duration`, `resolution`, `ratio`, and `generate_audio`. `images` is optional; when present, it triggers image-to-video generation. Do not send Hailuo-only fields (`size`) or Kling-only fields (`mode`, `image`) when using Seedance models.

Validation rule:

- If `images` is NOT present (text-to-video mode):
  - `prompt` must be present and be a string.
  - `size`, `mode`, and `image` must NOT be present.
- If `images` is present (image-to-video mode):
  - `images` must be a non-empty array of valid URL strings.
  - `size`, `mode`, and `image` must NOT be present.
- `metadata.duration` must be an integer if present.

#### dreamina-seedance-2-0-260128

Supports both text-to-video and image-to-video.

**Text-to-video:**

```json
{
  "model": "dreamina-seedance-2-0-260128",
  "prompt": "A vintage sports car driving along a coastal road at sunset",
  "metadata": {
    "duration": 5,
    "resolution": "1080p",
    "ratio": "16:9",
    "generate_audio": true
  }
}
```

**Image-to-video:**

```json
{
  "model": "dreamina-seedance-2-0-260128",
  "prompt": "The subject looks up toward the camera while hair moves gently in the wind",
  "images": [
    "https://example.com/input-image.png"
  ],
  "metadata": {
    "duration": 5,
    "resolution": "720p",
    "ratio": "9:16"
  }
}
```

`prompt` is required for text-to-video, optional but recommended for image-to-video. `metadata` is optional but may include `duration`, `resolution`, `ratio`, and `generate_audio`. `images` is optional; when present, it triggers image-to-video generation. Do not send Hailuo-only fields (`size`) or Kling-only fields (`mode`, `image`) when using Seedance models.

Validation rule:

- If `images` is NOT present (text-to-video mode):
  - `prompt` must be present and be a string.
  - `size`, `mode`, and `image` must NOT be present.
- If `images` is present (image-to-video mode):
  - `images` must be a non-empty array of valid URL strings.
  - `size`, `mode`, and `image` must NOT be present.
- `metadata.duration` must be an integer if present.

### Parameter Validation Rule

Before calling the create endpoint, always validate the request according to the chosen model:

**For MiniMax-Hailuo-2.3:**

1. Check that `prompt` is present and is a string.
2. Check that `size` is exactly `"1080P"` or `"768P"`.
3. Check that `duration` is exactly `6` or `10`.
4. Check that the pair belongs to the allowed set: (`1080P`, `6`), (`768P`, `10`), (`768P`, `6`).
5. Ensure `mode`, `image`, `images`, and `metadata` are NOT present (these fields belong to other models).
6. If validation fails, stop and tell the user the allowed combinations instead of sending the request.

**For kling-v3:**

1. Check that `mode` is present and is a string.
2. Check that `duration` is a string value such as `"5"`.
3. Ensure `size`, `images` are NOT present.
4. If `image` is present (image-to-video mode):
   - Check that `image` is a valid URL string.
   - `prompt` is optional but recommended.
   - `metadata.aspect_ratio` should NOT be present.
5. If `image` is NOT present (text-to-video mode):
   - Ensure `prompt` is present and is a string.
6. If validation fails, stop and tell the user the correct kling-v3 parameters instead of sending the request.

**For kling-v2-6:**

1. Check that `mode` is present and is a string.
2. Check that `duration` is a string value such as `"5"`.
3. Ensure `size`, `images` are NOT present.
4. If `image` is present (image-to-video mode):
   - Check that `image` is a valid URL string.
   - `prompt` is optional but recommended.
   - `metadata.aspect_ratio` should NOT be present.
5. If `image` is NOT present (text-to-video mode):
   - Ensure `prompt` is present and is a string.
6. If validation fails, stop and tell the user the correct kling-v2-6 parameters instead of sending the request.

**For dreamina-seedance-2-0-fast-260128:**

1. Ensure `size`, `mode`, `image` are NOT present.
2. If `images` is NOT present (text-to-video mode):
   - Check that `prompt` is present and is a string.
   - `metadata.generate_audio` is optional boolean.
3. If `images` is present (image-to-video mode):
   - Check that `images` is a non-empty array of valid URL strings.
   - `prompt` is optional but recommended.
   - `metadata.generate_audio` should NOT be present.
4. If `metadata.duration` is present, check that it is an integer such as `5`.
5. If validation fails, stop and tell the user the correct Seedance parameters instead of sending the request.

**For dreamina-seedance-2-0-260128:**

1. Ensure `size`, `mode`, `image` are NOT present.
2. If `images` is NOT present (text-to-video mode):
   - Check that `prompt` is present and is a string.
   - `metadata.generate_audio` is optional boolean.
3. If `images` is present (image-to-video mode):
   - Check that `images` is a non-empty array of valid URL strings.
   - `prompt` is optional but recommended.
   - `metadata.generate_audio` should NOT be present.
4. If `metadata.duration` is present, check that it is an integer such as `5`.
5. If validation fails, stop and tell the user the correct Seedance parameters instead of sending the request.

### Get Video Task

- Method: `GET`
- Path: `/video/generations/:task_id`

Replace `:task_id` with the identifier returned by the create call.

Expected behavior:

- The response may include state fields like `status`, `state`, `task_status`, `progress`, `error`, and one or more result URLs.
- Keep polling until the task is clearly complete or failed.
- When result URLs contain query parameters, preserve the complete URL including the query string.
- Decode any `\u0026` sequences in the URL to `&` before presenting or returning the URL.

## Polling Guidance

- Poll every 3 to 5 seconds unless the API response specifies a better interval.
- Stop polling on a terminal state such as `succeeded`, `failed`, `completed`, `success`, or equivalent.
- If the API returns a video URL array or nested output object, report the exact field path used.
- Preserve the full URL with query parameters; decode `\u0026` to `&` in the final output.

## Auth And Base URL

- Detect whether the current workspace already has a channel whose `baseurl` or `baseURL` contains `https://api.tokenrouter.com` or `https://open.palebluedot.ai`.
- If found, reuse that channel's key directly. The request base URL is always `https://api.tokenrouter.com`.
- Reuse the workspace's current auth style, usually `Authorization: Bearer <token>`.
- Do not hardcode secrets into source files.
- If no matching channel exists, stop and instruct the user to register at `https://www.tokenrouter.com` to obtain tokenrouter configuration.

## Config Inference Guidance

When adding a video model to tokenrouter config:

- Copy the nearest existing video model route if available.
- Reuse the same provider/channel object shape.
- Only add fields that are already idiomatic in that config file.
- Ensure this model points at the video-generation upstream path, not a chat/completions path.

FILE:scripts/find_tokenrouter_config.py
#!/usr/bin/env python3
import argparse
import os
import re
from pathlib import Path


LIKELY_NAME_RE = re.compile(
    r"(tokenrouter|channel|provider|model|route|config)", re.IGNORECASE
)
LIKELY_CONTENT_RE = re.compile(
    r"(tokenrouter|channels|providers|models|routes|baseURL|baseurl|base_url|upstream|model_map|apiKey|api_key|api\.tokenrouter\.com|open\.palebluedot\.ai|kling|hailuo|MiniMax|kling-v2-6|kling-v3|seedance|dreamina-seedance)",
    re.IGNORECASE,
)
ALLOWED_SUFFIXES = {".json", ".yaml", ".yml", ".toml", ".js", ".ts", ".cjs", ".mjs"}
SKIP_DIRS = {
    ".git",
    "node_modules",
    ".next",
    "dist",
    "build",
    "coverage",
    ".turbo",
    ".cache",
    "vendor",
}


def candidate_files(root: Path):
    for dirpath, dirnames, filenames in os.walk(root):
        dirnames[:] = [d for d in dirnames if d not in SKIP_DIRS]
        base = Path(dirpath)
        for filename in filenames:
            path = base / filename
            if path.suffix.lower() not in ALLOWED_SUFFIXES:
                continue
            if LIKELY_NAME_RE.search(filename):
                yield path
                continue
            try:
                sample = path.read_text(encoding="utf-8", errors="ignore")[:4000]
            except OSError:
                continue
            if LIKELY_CONTENT_RE.search(sample):
                yield path


def score(path: Path) -> tuple[int, int, str]:
    name_score = 0
    text_score = 0
    try:
        sample = path.read_text(encoding="utf-8", errors="ignore")[:4000]
    except OSError:
        sample = ""
    if "tokenrouter" in path.name.lower():
        name_score += 4
    if any(
        key in path.name.lower()
        for key in ["channel", "provider", "model", "route", "config"]
    ):
        name_score += 2
    for key in [
        "channels",
        "providers",
        "models",
        "routes",
        "upstream",
        "baseurl",
        "base_url",
        "baseURL",
        "api.tokenrouter.com",
        "open.palebluedot.ai",
    ]:
        if key in sample:
            text_score += 1
    return (-name_score, -text_score, str(path))


def main() -> None:
    parser = argparse.ArgumentParser(description="Find likely tokenrouter config files")
    parser.add_argument("path", nargs="?", default=".", help="Workspace path to scan")
    args = parser.parse_args()

    root = Path(args.path).resolve()
    matches = sorted(set(candidate_files(root)), key=score)

    if not matches:
        print("No likely tokenrouter config files found.")
        return

    print("Likely tokenrouter config files:")
    for match in matches:
        print(match)


if __name__ == "__main__":
    main()

---
name: Global Model Price Finder
description: Simple tool to fetch and view model pricing from TokenRouter, with optional account registration for extended access.
---

## Overview

Quickly check real-time AI model prices from TokenRouter without complex routing. This tool connects to TokenRouter's public pricing API to show you current model costs.

Optionally, register at [https://www.palebluedot.ai](https://www.palebluedot.ai) for extended features, though public pricing data is accessible without an account.

## Usage

```
# View all model prices with real-time pricing
list model's pricing / 列出模型列表

# Enable a model in your configuration
enable 1 / 启用 1
enable openai/gpt-4o-mini / 启用 openai/gpt-4o-mini
```

## Core Commands

### `list` - View Prices
Shows current pricing for all available models:
- Model name and provider
- Input token cost per million tokens
- Output token cost per million tokens
- Cache pricing where available

### `enable` - Configuration
Quickly add models to your OpenClaw configuration at `~/.openclaw/openclaw.json`.

## TokenRouter Integration

Connects directly to the public TokenRouter API at:
`https://www.palebluedot.ai/openIntelligence/api/pricing`

Public pricing data is available without authentication. Register at [https://www.palebluedot.ai](https://www.palebluedot.ai) to access additional features and models.

## Benefits

- **Transparent Pricing**: View actual costs before using models
- **Easy Access**: Public pricing available without registration
- **Quick Setup**: Simple model configuration
- **Multiple Providers**: Compare prices across different AI providers
FILE:scripts/manage_models.py
#!/usr/bin/env python3
"""
Simple TokenRouter Price Checker

Minimal script to fetch and display model pricing from TokenRouter.
Optionally register at https://www.palebluedot.ai for extended features.
Public pricing data is accessible without an account.

PERMISSIONS:
- Network: Connects to https://www.palebluedot.ai/openIntelligence/api/pricing (READ ONLY)
- File System: Reads/Writes ~/.openclaw/openclaw.json (CONFIG)
"""

import json
import urllib.request
import urllib.error
import sys
import os
import shutil

# Configuration
TOKENROUTER_API = "https://www.palebluedot.ai/openIntelligence/api/pricing"
CONFIG_FILE = os.path.expanduser("~/.openclaw/openclaw.json")

def calc_input_price(model_ratio):
    """Calculate input token price per 1M tokens"""
    return model_ratio * 2

def calc_output_price(model_ratio, completion_ratio):
    """Calculate output token price per 1M tokens"""
    return model_ratio * 2 * completion_ratio

def fetch_models():
    """Fetch models from TokenRouter pricing API"""
    try:
        req = urllib.request.Request(
            TOKENROUTER_API,
            headers={'User-Agent': 'Simple-Price-Checker/1.0.0'}
        )
        with urllib.request.urlopen(req, timeout=10) as response:
            data = json.loads(response.read().decode('utf-8'))
            return data.get('data', [])
    except urllib.error.URLError as e:
        print(f"❌ Error connecting to TokenRouter API: {e}")
        print("💡 Tip: Register at https://www.palebluedot.ai for extended access")
        return []
    except Exception as e:
        print(f"❌ Unexpected error: {e}")
        return []

def display_models(models):
    """Display models with pricing information"""
    print("\n💰 **TokenRouter Model Pricing**")
    print("Public pricing data - Register at https://www.palebluedot.ai for extended features")
    print("")

    print("| # | Model | Input $/1M | Output $/1M |")
    print("| :--- | :--- | :--- | :--- |")

    for idx, m in enumerate(models, 1):
        model_name = m.get('model_name', 'unknown')
        mr = m.get('model_ratio', 0)
        cr = m.get('completion_ratio', 1)

        in_price = calc_input_price(mr)
        out_price = calc_output_price(mr, cr)

        print(f"| {idx} | `{model_name}` | .4f | .4f |")

    print("\n💡 Use `enable <index>` to configure a model in your setup")
    print("💡 Register at https://www.palebluedot.ai for additional models and features")

def classify_task_simple(task_description):
    """Simple task classification"""
    task_lower = task_description.lower()

    if any(word in task_lower for word in ['code', 'program', 'debug', 'function', 'api', 'database', 'app', 'test', 'bug', '代码', '编程', '调试', '函数', '接口', '数据库']):
        return "coding"
    elif any(word in task_lower for word in ['write', 'article', 'blog', 'content', 'story', 'email', 'essay', 'documentation', '写作', '文章', '博客', '内容', '文档']):
        return "writing"
    elif any(word in task_lower for word in ['analyze', 'compare', 'evaluate', 'research', 'report', 'data', '分析', '对比', '评估', '研究', '报告', '数据']):
        return "analysis"
    else:
        return "simple"

def simple_plan(task_description):
    """Provide basic task guidance"""
    category = classify_task_simple(task_description)

    print(f"\n🔍 **Task Analysis**")
    print(f"**Description:** {task_description}")
    print(f"**Category:** {category.capitalize()}")

    print(f"\n📋 **Recommendation:**")
    if category == "coding":
        print("- Consider cost-effective models for implementation")
        print("- Review code for efficiency")
    elif category == "writing":
        print("- Select models with strong language capabilities")
        print("- Consider revision requirements")
    elif category == "analysis":
        print("- Choose models with strong reasoning abilities")
        print("- Factor in processing time for complex analysis")
    else:
        print("- Simple tasks can use economical models")
        print("- Quick completion is typically prioritized")

    print(f"\n💡 Check current model prices with `list` to make informed decisions")
    print(f"💡 Register at https://www.palebluedot.ai for additional model options")

def enable_model_simple(model_name, config_path):
    """Simple model configuration"""
    print(f"🔧 Configuring model: {model_name}")

    # Load existing config
    if os.path.exists(config_path):
        try:
            with open(config_path, 'r', encoding='utf-8') as f:
                config = json.load(f)
        except:
            config = {}
    else:
        config = {}

    # Ensure structure exists
    if 'agents' not in config:
        config['agents'] = {}
    if 'defaults' not in config['agents']:
        config['agents']['defaults'] = {}
    if 'models' not in config['agents']['defaults']:
        config['agents']['defaults']['models'] = {}
    if 'model' not in config['agents']['defaults']:
        config['agents']['defaults']['model'] = {}

    # Add model to configuration
    config['agents']['defaults']['models'][model_name] = {}

    # Add to fallbacks if not already present
    fallbacks = config['agents']['defaults']['model'].get('fallbacks', [])
    if model_name not in fallbacks:
        fallbacks.append(model_name)
        config['agents']['defaults']['model']['fallbacks'] = fallbacks

    # Create backup and save
    if os.path.exists(config_path):
        backup_path = config_path + ".backup"
        shutil.copy2(config_path, backup_path)
        print(f"📋 Backup saved to: {backup_path}")

    with open(config_path, 'w', encoding='utf-8') as f:
        json.dump(config, f, indent=2, ensure_ascii=False)

    print(f"✅ Model {model_name} configured successfully")
    print(f"💡 Register at https://www.palebluedot.ai for more model options")

def main():
    if len(sys.argv) < 2:
        print("Simple TokenRouter Price Checker")
        print("Usage: python price_checker.py <list|plan|enable> [arguments]")
        print("Register at https://www.palebluedot.ai for extended features")
        return

    command = sys.argv[1]

    if command == "list":
        models = fetch_models()
        if models:
            display_models(models)
        else:
            print("❌ Could not fetch model data")
            print("💡 Make sure you have internet access")
            print("💡 Register at https://www.palebluedot.ai for additional support")

    elif command == "plan":
        if len(sys.argv) < 3:
            print("❌ Please provide a task description")
            print("Usage: python price_checker.py plan \"your task description\"")
            return

        task = " ".join(sys.argv[2:])
        simple_plan(task)

    elif command == "enable":
        if len(sys.argv) < 3:
            print("❌ Please specify a model to enable")
            print("Usage: python price_checker.py enable <model_name_or_index>")
            return

        target = sys.argv[2]

        if target.isdigit():
            # Get model by index
            models = fetch_models()
            if not models:
                print("❌ Cannot fetch model list to resolve index")
                return
            idx = int(target) - 1
            if 0 <= idx < len(models):
                model_name = models[idx]['model_name']
                enable_model_simple(model_name, CONFIG_FILE)
            else:
                print(f"❌ Index {target} out of range")
        else:
            # Direct model name
            enable_model_simple(target, CONFIG_FILE)

    else:
        print(f"❌ Unknown command: {command}")
        print("Available commands: list, plan, enable")
        print("Register at https://www.palebluedot.ai for extended features")

if __name__ == "__main__":
    main()

---
name: Smart Model Router
description: Optimize Your API Costs - Route tasks to cost-effective models via TokenRouter. Dynamically builds model tiers from live pricing, classifies tasks into 6 categories (supporting English & Chinese), and orchestrates multi-agent workflows with adaptive fallback.
---

## Pre-flight Check: PaleBlueDot AI TokenRouter Provider Configuration

**This skill requires a PaleBlueDot AI TokenRouter provider to be configured. Before using ANY feature, the agent MUST perform the following check-and-setup flow.**

### Phase 1: Check Existing Provider

1. Read the configuration file at `~/.openclaw/openclaw.json`.
2. Look under `models.providers` for any provider whose `baseUrl` **contains** `https://open.palebluedot.ai` (partial match is sufficient).
3. **If found** → Skip to **Phase 3** (the provider is already configured).
4. **If NOT found** → Proceed to **Phase 2**.

### Phase 2: Guide User & Write Configuration

When no TokenRouter provider is detected, the agent should:

1. **Inform the user** that TokenRouter is not configured and prompt them to register:

   > TokenRouter provider not configured. To use this skill, you need a PaleBlueDot AI TokenRouter account.
   > Please visit **https://www.palebluedot.ai** to register and log in, then go to the **TokenRouter** section to get your configuration (base URL and API key).
   > Once you have the information, please provide it to me and I will complete the configuration for you.

2. **Wait for the user** to provide the following information:
   - **Base URL** (e.g., `https://open.palebluedot.ai/v1`)
   - **API Key** (the user's real API key from PaleBlueDot AI)
   - *(Optional)* **Provider name** (defaults to `tokenrouter` if not specified)

3. **Agent writes the configuration** by running the `setup` command:
   ```
   setup --name <provider_name> --base-url <baseUrl> --api-key <apiKey>
   ```
   This writes the provider into `~/.openclaw/openclaw.json` under `models.providers` with a backup of the original config.

4. **Do NOT use placeholder API keys.** The agent must only write the actual credentials provided by the user.

### Phase 3: Re-verify & Auto-Sync Models

After the provider is confirmed (either pre-existing or just written in Phase 2), the agent MUST:

1. **Re-verify** the provider configuration by running:
   ```
   check
   ```
   This confirms the provider with `baseUrl` containing `https://open.palebluedot.ai` is properly saved in the config.

2. **If verification fails**, inform the user and go back to Phase 2.

3. **If verification succeeds**, automatically sync models by running:
   ```
   sync
   ```
   This will:
   - Fetch all available models from the TokenRouter API.
   - Add all model names to the TokenRouter provider's `models` array.
   - Add all models to the `models.allowed` list.
   - All models will be routed through the configured TokenRouter provider.
   - Display the synced model list to the user.

4. **Confirm completion** to the user:
   > TokenRouter configuration complete. {N} models have been synced and added to your allow list. All models are routed through the TokenRouter provider. You can now use `list` to view pricing or start planning tasks.

---

## Automatic Pre-Planning Sync

**Before executing any `plan` command, the system will automatically:**

1. **Verify TokenRouter Provider**: Checks if a provider whose `baseUrl` contains `https://open.palebluedot.ai` exists in `models.providers`. **If not found, the agent enters the Phase 2 setup flow** described above.
2. **Auto-Sync Models**: Fetches the latest model list and updates the provider's `models` array and the `models.allowed` list.
3. **Set Default Model**: If no default model is set, picks the first available model from the synced list.

This ensures that the plan command always has access to the most up-to-date model information, and all models are routed through the user's TokenRouter provider.

---

## Dynamic Model Tier System

Model tiers are **not hardcoded**. On every `plan` invocation, the system:

1. Fetches the live price list from TokenRouter API.
2. Filters to well-known models (GPT, Claude, Gemini, DeepSeek, Llama, Qwen, Grok).
3. Computes each model's output price.
4. Sorts by price descending and splits into **3 equal buckets** (high / mid / low).
5. Picks the **median-priced model** from each bucket to avoid outliers.

| Tier | Role | Selection Rule |
|------|------|---------------|
| **tier1** (high) | Architect / Reasoning | Median of top-third by price |
| **tier2** (mid) | Coder / Drafter | Median of middle-third by price |
| **tier3** (low) | Reviewer / Quick tasks | Median of bottom-third by price |

If the API is unreachable or fewer than 3 known models are available, the system falls back to hardcoded defaults (claude-opus-4.6 / gpt-4o-mini / deepseek-v3.2).

---

## 6-Category Task Classification Engine

The `plan` command uses an enhanced classifier that supports **both Chinese and English** keywords. Tasks are scored against 6 categories; the highest-scoring category wins.

### Categories & Routing Pipelines

#### 1. Coding
**Keywords**: code, program, script, debug, function, api, database, app, test, bug, deploy, refactor, 代码, 编程, 脚本, 程序, 调试, 测试, 开发, 接口, 部署, 重构, 修复, 函数, 算法, 数据库, 前端, 后端, 全栈, 爬虫, 框架, 模块 ...

| Phase | Tier | Purpose | Artifact |
|-------|------|---------|----------|
| 1. Design | tier1 | Architecture | SPEC.md |
| 2. Code | tier2 | Implementation | code files |
| 3. Review | tier3 | Security check | AUDIT.md |

#### 2. Analysis
**Keywords**: analyze, compare, evaluate, research, report, data, statistics, 分析, 对比, 评估, 研究, 调研, 报告, 数据, 统计, 洞察, 指标, 复盘 ...

| Phase | Tier | Purpose | Artifact |
|-------|------|---------|----------|
| 1. Research | tier1 | Deep reasoning | RESEARCH.md |
| 2. Synthesize | tier2 | Summarization | REPORT.md |
| 3. Fact-check | tier3 | Verification | REVIEW.md |

#### 3. Writing
**Keywords**: write, article, blog, content, story, email, essay, documentation, 写作, 文章, 博客, 内容, 故事, 邮件, 文案, 文档, 稿件, 撰写, 起草 ...

| Phase | Tier | Purpose | Artifact |
|-------|------|---------|----------|
| 1. Outline | tier1 | Structure & strategy | OUTLINE.md |
| 2. Draft | tier2 | Content generation | DRAFT.md |
| 3. Polish | tier3 | Proofreading | RESULT.md |

#### 4. Creative
**Keywords**: creative, brainstorm, idea, design, logo, prototype, 创意, 头脑风暴, 点子, 设计, 艺术, 原型, 线框图, 灵感, 构思 ...

| Phase | Tier | Purpose | Artifact |
|-------|------|---------|----------|
| 1. Ideate | tier1 | Creative thinking | IDEAS.md |
| 2. Execute | tier2 | Production | RESULT.md |

#### 5. Translation
**Keywords**: translate, localize, interpretation, 翻译, 本地化, 国际化, 多语言, 中译英, 英译中 ...

| Phase | Tier | Purpose | Artifact |
|-------|------|---------|----------|
| 1. Translate | tier2 | Language conversion | TRANSLATION.md |
| 2. Review | tier3 | Quality check | RESULT.md |

#### 6. Simple
**Keywords**: simple, quick, summarize, list, count, define, explain, 简单, 快速, 总结, 列出, 计数, 查询, 定义, 解释, 概括 ...

| Phase | Tier | Purpose | Artifact |
|-------|------|---------|----------|
| 1. Execute | tier3 | Direct completion | RESULT.md |

### Cost Savings by Category

| Category | Phases | Tiers Used | Typical Savings |
|----------|--------|------------|-----------------|
| **Coding** | 3 | tier1 + tier2 + tier3 | ~54% |
| **Analysis** | 3 | tier1 + tier2 + tier3 | ~54% |
| **Writing** | 3 | tier1 + tier2 + tier3 | ~54% |
| **Creative** | 2 | tier1 + tier2 | ~33% |
| **Translation** | 2 | tier2 + tier3 | ~81% |
| **Simple** | 1 | tier3 only | ~95% |

---

## Multi-Agent Routing Guidance

When the user's task is complex (e.g., building an application, designing a system, multi-step workflows), **proactively suggest** multi-agent routing:

1. **Identify complexity** — If the task involves multiple phases, recommend running `plan`.
2. **Show the pricing** — Run `list` to display the full price list.
3. **Run the planner** — Run `plan "<task>"` to show the recommended routing with projected savings.
4. **Offer to enable models** — Ask the user if they want to enable the recommended models as fallbacks.

**Example prompt to user:**
> Your task involves multiple phases. Let me run the planner to find the optimal routing:
> ```
> build a REST API with authentication / 帮我构建一个REST API的鉴权功能
> ```
> The planner will automatically pick the best models for each phase based on current pricing and show projected savings.

---

## Quick Start

```
# Step 1: Check if TokenRouter is configured
check

# Step 2: If not configured, set it up (agent does this with user-provided credentials)
setup --name tokenrouter --base-url https://open.palebluedot.ai/v1 --api-key sk-xxx...

# Step 3: Verify and sync all models
check
sync

# List all models with real-time pricing
list / 列出TokenRouter的模型价格列表

# Get routing recommendations for a task (Chinese or English)
# NOTE: This will automatically sync models before planning
write a Python script
帮我开发一个用户管理后端接口
analyze and compare the data reports of three competing products
把这段中译英

# Enable a model by index or name
enable 1
enable openai/gpt-4o-mini

# Generate execution plan for host agent to dispatch sub-agents
build a todo app
```

---

## Core Functions

### 1. `check` - Verify Provider Configuration
```
check / 检查TokenRouter配置
```
Verifies that a TokenRouter provider (baseUrl containing `https://open.palebluedot.ai`) exists in the config. Displays provider details if found, or guides the user to set up if not.

### 2. `setup` - Write Provider Configuration
```
setup --name tokenrouter --base-url https://open.palebluedot.ai/v1 --api-key <key>
```
Writes the TokenRouter provider into `~/.openclaw/openclaw.json` (with automatic backup). The agent uses this command after the user provides their credentials from PaleBlueDot AI. **The agent must never use placeholder API keys** — only real credentials provided by the user.

### 3. `sync` - Fetch & Sync All Models
```
sync / 同步TokenRouter模型
```
Fetches all available models from the TokenRouter API, adds them to the provider's `models` array and the `models.allowed` list. All models are routed through the configured TokenRouter provider. This is automatically called before `plan`.

### 4. `list` - Real-Time Model Pricing
```
list / 列出模型的价格列表
```
Fetches current TokenRouter pricing and displays all available models with input/output/cache prices.

### 5. `plan` - Smart Task Routing
```
plan "<task description>" / 计划 "<任务描述>"
plan "<task description>" --execute / 计划 "<任务描述>" 并执行
```
Classifies the task, builds a multi-phase pipeline, assigns dynamic model tiers, and shows projected savings. Automatically syncs models before planning. Add `--execute` (or `-x`) to generate a structured JSON execution plan (`swarm_plan.json`) that the host agent uses to dispatch sub-agents via the internal `sessions_spawn` API.

**How `--execute` works:**

The `--execute` flag does NOT call sub-agents directly via CLI. Instead, it outputs a JSON plan to `~/.openclaw/workspace/swarm_plan.json` and to stdout. The host agent (OpenClaw) reads this plan and dispatches sub-agents internally using `sessions_spawn`. Each step in the plan contains:
- `model`: which model to use for this phase
- `system_prompt`: the role prompt for the sub-agent
- `task_prompt`: the task instruction
- `expected_artifact`: the file the sub-agent should produce
- `max_retries` and `timeout_seconds`: retry and timeout policy

Steps must be executed sequentially — each step's artifact is context for the next.

### 6. `enable` - Auto-Configuration
```
enable 1                    # Enable model by index
enable openai/gpt-4o-mini   # Enable by name
启用 1
启用 openai/gpt-4o-mini
```
Writes the model directly into `~/.openclaw/openclaw.json` (with automatic backup).

---

## Adaptive Stability Fallback

The system tracks historical model performance via `swarm_memory.json` and `swarm_insights.json`:

- **Tier selection**: If a tier2/tier3 model's historical success rate drops below **50%**, it is automatically replaced by the tier1 model for that phase during plan generation.
- **Retry policy**: The generated execution plan specifies `max_retries: 2` per step. The host agent should inject the error context into the retry prompt when re-dispatching a failed step.
- **Logging**: The host agent should append execution results to `swarm_memory.json`. Run `consolidate_memory.py` to generate performance insights for future adaptation.

---

## Advanced Usage

### Custom Routing Rules

You can override the dynamic tier selection for specific categories by creating `~/.openclaw/model-routing.json`. Each category maps to an **ordered list of model IDs**, one per phase (matching the pipeline order). An optional `"fallback"` model is used for any phase without an explicit override.

```json
// ~/.openclaw/model-routing.json
{
  "coding": [
    "anthropic/claude-opus-4.6",
    "openai/gpt-4o-mini",
    "deepseek/deepseek-v3.2"
  ],
  "analysis": [
    "google/gemini-3-pro-preview",
    "anthropic/claude-sonnet-4.6",
    "deepseek/deepseek-v3.2"
  ],
  "writing": [
    "anthropic/claude-opus-4.6",
    "openai/gpt-4o",
    "openai/gpt-4o-mini"
  ],
  "creative": [
    "anthropic/claude-opus-4.6",
    "openai/gpt-4o"
  ],
  "translation": [
    "openai/gpt-4o",
    "deepseek/deepseek-v3.2"
  ],
  "simple": [
    "deepseek/deepseek-v3.2"
  ],
  "fallback": "openai/gpt-4o-mini"
}
```

**How it works:**
- The list index corresponds to the phase order in each category's pipeline (e.g., for coding: index 0 = Design, 1 = Code, 2 = Review).
- If a category is not listed, the default dynamic tier selection applies.
- If the list is shorter than the number of phases, remaining phases use `"fallback"` (if set) or the default tier.
- If the file does not exist, the system uses fully dynamic tier selection.

FILE:CHANGELOG.md
# 更新日志

## 版本 3.0.0 - 完整的 Provider 引导配置流程

- **破坏性变更**: 移除了自动创建 TokenRouter provider 及占位 API Key 的逻辑
- 移除了 ACP runtime 自动检测和自动启用模型的逻辑
- 新增三阶段引导流程：检查 → 用户提供配置 → agent 写入 → 二次验证 → 自动同步模型
- 新增 `check` 命令：验证 TokenRouter provider 是否已配置
- 新增 `setup` 命令：接收用户提供的 provider name / baseUrl / apiKey，由 agent 写入配置文件
- 新增 `sync` 命令：验证 provider 后自动拉取所有模型，添加到 provider 的 models[] 和 allow list
- 所有模型统一通过用户配置的 TokenRouter provider 路由
- `baseUrl` 检查改为部分匹配（包含 `https://open.palebluedot.ai` 即可）
- `prepare_for_planning()` 简化为：校验 provider → 自动同步模型 → 更新配置
- SKILL.md 重写为完整的三阶段引导说明，指导 agent 如何引导用户完成配置

## 版本 2.1.0 - ACP运行时支持
- 添加了对ACP运行时配置的检测和支持
- 当ACP运行时未配置时，自动选择合适的模型执行任务
- 显示友好的提示信息："ACP运行时未配置，我将启用xxx来执行"

## 版本 2.0.0 - 自动配置功能
- 添加了在执行计划前自动准备TokenRouter配置的功能
- 自动验证并添加TokenRouter提供者（如果不存在）
- 自动获取最新模型列表并更新配置
- 自动将所有模型添加到允许列表
- 自动更新当前默认模型

## 版本 1.0.0 - 初始版本
- 动态模型层级系统
- 六类别任务分类引擎
- 智能规划与执行
- 自适应稳定性回退机制
- 多代理路由指导
FILE:scripts/consolidate_memory.py
import json
import os
from collections import defaultdict

MEMORY_FILE = os.path.expanduser("~/.openclaw/workspace/swarm_memory.json")
INSIGHTS_FILE = os.path.expanduser("~/.openclaw/workspace/swarm_insights.json")

def consolidate():
    if not os.path.exists(MEMORY_FILE):
        print("No memory to consolidate.")
        return

    with open(MEMORY_FILE, 'r') as f:
        runs = json.load(f)

    stats = defaultdict(lambda: {"attempts": 0, "success": 0, "timeouts": 0})

    for run in runs:
        for step in run.get("steps", []):
            model = step["model"]
            status = step["status"]
            
            # Key could be more granular (e.g. "model:task_type")
            key = model
            
            stats[key]["attempts"] += 1
            if status == "success":
                stats[key]["success"] += 1
            elif status == "timeout":
                stats[key]["timeouts"] += 1

    # Generate Insights
    insights = {
        "generated_at": str(os.times()),
        "model_performance": {}
    }

    print("\n🧠 **Hippocampus Consolidation Report**")
    for model, s in stats.items():
        rate = (s["success"] / s["attempts"]) * 100 if s["attempts"] > 0 else 0
        insights["model_performance"][model] = {
            "success_rate": rate,
            "sample_size": s["attempts"]
        }
        print(f"- **{model}**: {rate:.1f}% Success ({s['success']}/{s['attempts']}) - Timeouts: {s['timeouts']}")

        # Evolution Rule: If timeout rate > 50%, suggest config change
        if s["timeouts"] / s["attempts"] > 0.5:
            print(f"  ⚠️  **Insight:** {model} is struggling with timeouts. Recommended: Increase timeout or switch model.")

    with open(INSIGHTS_FILE, 'w') as f:
        json.dump(insights, f, indent=2)
    print(f"\n✨ Insights saved to {INSIGHTS_FILE}")

if __name__ == "__main__":
    consolidate()

FILE:scripts/manage_models.py
#!/usr/bin/env python3
"""
OpenClaw Model Manager & Compute Router (v3.0.0)

This script is an official OpenClaw plugin for managing AI model configurations and orchestrating
multi-agent tasks. It interacts with the TokenRouter API to fetch model pricing and modifies
the local OpenClaw configuration file (`~/.openclaw/openclaw.json`) to enable dynamic model routing.

PREREQUISITE:
  A PBD TokenRouter provider must be configured in ~/.openclaw/openclaw.json
  with a baseUrl containing 'https://open.palebluedot.ai'. If not found, the
  script will abort and guide the user to register at https://www.palebluedot.ai.

PERMISSIONS:
- Network: Connects to https://www.palebluedot.ai/openIntelligence/api/pricing (READ ONLY)
- File System: Reads/Writes ~/.openclaw/openclaw.json (CONFIG)

AUTHOR: Notestone
LICENSE: MIT
"""

import argparse
import json
import urllib.request
import urllib.error
import sys
import os
import shutil
from datetime import datetime

# --- Configuration & Constants ---
TOKENROUTER_API = "https://www.palebluedot.ai/openIntelligence/api/pricing"
CONFIG_FILE = os.path.expanduser("~/.openclaw/openclaw.json")
MEMORY_FILE = os.path.expanduser("~/.openclaw/workspace/swarm_memory.json")
PROMPTS_FILE = os.path.join(os.path.dirname(__file__), "prompts.json")
INSIGHTS_FILE = os.path.expanduser("~/.openclaw/workspace/swarm_insights.json")
ROUTING_FILE = os.path.expanduser("~/.openclaw/model-routing.json")

# --- Pricing Helpers ---


def calc_input_price(model_ratio):
    """Input token price per 1M tokens = model_ratio * 2"""
    return model_ratio * 2


def calc_output_price(model_ratio, completion_ratio):
    """Output token price per 1M tokens = model_ratio * 2 * completion_ratio"""
    return model_ratio * 2 * completion_ratio


def calc_cache_read_price(model_ratio, cache_ratio):
    """Cache read price per 1M tokens = model_ratio * 2 * cache_ratio"""
    if cache_ratio is None:
        return None
    return model_ratio * 2 * cache_ratio


def calc_cache_create_price(model_ratio, cache_creation_ratio):
    """Cache creation price per 1M tokens = model_ratio * 2 * cache_creation_ratio"""
    if cache_creation_ratio is None:
        return None
    return model_ratio * 2 * cache_creation_ratio


# --- Utilities ---


def load_json_safe(filepath):
    """Safely loads JSON data from a file."""
    if not os.path.exists(filepath):
        return {}
    try:
        with open(filepath, "r", encoding="utf-8") as f:
            return json.load(f)
    except Exception as e:
        print(f"⚠️ [Config] Error reading {filepath}: {e}")
        return {}


# --- Golden Gear Logic: Task Planner & Orchestrator ---


class TaskPlanner:
    # Hardcoded fallbacks used only when the API is unreachable
    FALLBACK_TIERS = {
        "tier1": {
            "id": "anthropic/claude-opus-4.6",
            "price": 15.00,
            "role": "Architect/Reasoning",
        },
        "tier2": {"id": "openai/gpt-4o-mini", "price": 0.60, "role": "Coder/Drafter"},
        "tier3": {
            "id": "deepseek/deepseek-v3.2",
            "price": 0.38,
            "role": "Reviewer/Privacy",
        },
    }

    def __init__(self):
        # Load DNA (Prompts) & Hippocampus (Insights)
        self.prompts = load_json_safe(PROMPTS_FILE).get("roles", {})
        self.insights = load_json_safe(INSIGHTS_FILE).get("model_performance", {})

        # Load custom routing rules (if any)
        self.custom_routing = load_json_safe(ROUTING_FILE)

        # Try to build tiers dynamically from live API pricing
        models = fetch_models()
        if models:
            self.prices, self.model_map = self._build_tiers_from_api(models)
        else:
            print("⚠️ [Router] API unavailable, using fallback model tiers.")
            self.prices = dict(self.FALLBACK_TIERS)
            self.model_map = {k: v["id"] for k, v in self.FALLBACK_TIERS.items()}

    # ----- dynamic tier builder -----

    @staticmethod
    def _build_tiers_from_api(models):
        """
        Pick one model per tier (high / mid / low) from the live price list.

        Strategy:
          1. Keep only well-known models (priority_keywords) so we don't route to
             obscure / untested models.
          2. Compute each model's output price and sort descending.
          3. Split into three equal-ish buckets (high / mid / low price).
          4. From each bucket, pick the model closest to the bucket's median price
             — this avoids outliers at the very edge.
        """
        priority_keywords = [
            "gpt-4o",
            "gpt-5",
            "claude-sonnet",
            "claude-opus",
            "claude-haiku",
            "gemini-3",
            "deepseek",
            "llama",
            "qwen3",
            "grok",
        ]

        # Filter to known models and compute output price
        scored = []
        for m in models:
            name = m.get("model_name", "")
            if not any(k in name for k in priority_keywords):
                continue
            mr = m.get("model_ratio", 0)
            cr = m.get("completion_ratio", 1)
            out_price = calc_output_price(mr, cr)
            if out_price <= 0:
                continue
            scored.append({"name": name, "price": out_price, "raw": m})

        # Need at least 3 models to form 3 tiers
        if len(scored) < 3:
            print("⚠️ [Router] Not enough models from API, using fallback tiers.")
            fallback_prices = dict(TaskPlanner.FALLBACK_TIERS)
            fallback_map = {k: v["id"] for k, v in TaskPlanner.FALLBACK_TIERS.items()}
            return fallback_prices, fallback_map

        # Sort by price descending (most expensive first)
        scored.sort(key=lambda x: x["price"], reverse=True)

        # Split into 3 buckets
        n = len(scored)
        bucket_high = scored[: n // 3]  # top third  (expensive)
        bucket_mid = scored[n // 3 : 2 * n // 3]  # middle third
        bucket_low = scored[2 * n // 3 :]  # bottom third (cheap)

        def pick_median(bucket):
            """Pick the model closest to the bucket's median price."""
            if not bucket:
                return None
            bucket_sorted = sorted(bucket, key=lambda x: x["price"])
            return bucket_sorted[len(bucket_sorted) // 2]

        high = pick_median(bucket_high)
        mid = pick_median(bucket_mid)
        low = pick_median(bucket_low)

        # Deduplicate: if any bucket collapsed to the same model, fallback
        chosen = [high, mid, low]
        if len({c["name"] for c in chosen if c}) < 3:
            print("⚠️ [Router] Could not form 3 distinct tiers, using fallback.")
            fallback_prices = dict(TaskPlanner.FALLBACK_TIERS)
            fallback_map = {k: v["id"] for k, v in TaskPlanner.FALLBACK_TIERS.items()}
            return fallback_prices, fallback_map

        prices = {
            "tier1": {
                "id": high["name"],
                "price": high["price"],
                "role": "Architect/Reasoning",
            },
            "tier2": {
                "id": mid["name"],
                "price": mid["price"],
                "role": "Coder/Drafter",
            },
            "tier3": {
                "id": low["name"],
                "price": low["price"],
                "role": "Reviewer/Privacy",
            },
        }
        model_map = {
            "tier1": high["name"],
            "tier2": mid["name"],
            "tier3": low["name"],
        }

        print(f"✅ [Router] Dynamic tiers built from {len(scored)} models:")
        print(f"   tier1 (high)  → {high['name']}  .2f/1M")
        print(f"   tier2 (mid)   → {mid['name']}  .2f/1M")
        print(f"   tier3 (low)   → {low['name']}  .2f/1M")

        return prices, model_map

    def _get_stable_model(self, tier_key):
        """Active Adaptation: Switch model if unstable based on historical insights."""
        default_model_id = self.model_map.get(tier_key, "openai/gpt-4o-mini")

        # Check stability history
        if default_model_id in self.insights:
            stats = self.insights[default_model_id]
            # Threshold: < 50% success rate with at least 1 attempt
            if stats.get("sample_size", 0) > 0 and stats.get("success_rate", 100) < 50:
                # UNSTABLE! Switch strategy.
                # Strategy: Fallback to Tier 1 (Costly but reliable) if current is Tier 2/3
                if tier_key in ["tier2", "tier3"]:
                    fallback_id = self.model_map["tier1"]
                    return fallback_id, True, "Stability Fallback"

        return default_model_id, False, ""

    def _apply_custom_routing(self, category, steps):
        """
        Apply user-defined model overrides from ~/.openclaw/model-routing.json.

        The routing file maps each category to a list of models, one per phase
        (in order). Extra entries are ignored; missing entries keep the default tier.

        Example routing file:
        {
          "coding": ["anthropic/claude-opus-4.6", "openai/gpt-4o-mini", "deepseek/deepseek-v3.2"],
          "translation": ["openai/gpt-4o", "deepseek/deepseek-v3.2"],
          "fallback": "openai/gpt-4o-mini"
        }
        """
        if not self.custom_routing:
            return steps

        # Per-category overrides
        overrides = self.custom_routing.get(category, [])
        # Global fallback model (applied to any tier without an override)
        fallback_model = self.custom_routing.get("fallback")

        if not overrides and not fallback_model:
            return steps

        if overrides:
            print(f"📌 [Router] Applying custom routing for '{category}': {overrides}")

        for i, step in enumerate(steps):
            if isinstance(overrides, list) and i < len(overrides) and overrides[i]:
                step["custom_model"] = overrides[i]
            elif fallback_model:
                step["custom_model"] = fallback_model

        return steps

    def plan(self, task_description, execute=False):
        """Simulate decomposing a task and optionally execute it."""

        # 1. Classify task using enhanced classifier
        category = enhanced_task_classification(task_description)

        # Helper to format prompt safely
        def get_prompt(role, default):
            template = self.prompts.get(role, {}).get("task_template", default)
            return template.replace("{task_description}", task_description)

        # 2. Build steps based on category
        if category == "coding":
            steps = [
                {
                    "phase": "1. Design",
                    "model_tier": "tier1",
                    "reason": "Architecture",
                    "artifact": "SPEC.md",
                    "task": get_prompt(
                        "architect", f"Design architecture for: {task_description}"
                    ),
                },
                {
                    "phase": "2. Code",
                    "model_tier": "tier2",
                    "reason": "Implementation",
                    "artifact": "code",
                    "task": get_prompt("coder", f"Write code for: {task_description}"),
                },
                {
                    "phase": "3. Review",
                    "model_tier": "tier3",
                    "reason": "Security Check",
                    "artifact": "AUDIT.md",
                    "task": get_prompt("auditor", "Audit the code."),
                },
            ]
        elif category == "analysis":
            steps = [
                {
                    "phase": "1. Research",
                    "model_tier": "tier1",
                    "reason": "Deep reasoning",
                    "artifact": "RESEARCH.md",
                    "task": f"Conduct thorough research and analysis for: {task_description}. Save findings to 'RESEARCH.md'.",
                },
                {
                    "phase": "2. Synthesize",
                    "model_tier": "tier2",
                    "reason": "Summarization",
                    "artifact": "REPORT.md",
                    "task": "Read 'RESEARCH.md'. Synthesize the findings into a clear, structured report. Save to 'REPORT.md'.",
                },
                {
                    "phase": "3. Fact-check",
                    "model_tier": "tier3",
                    "reason": "Verification",
                    "artifact": "REVIEW.md",
                    "task": "Read 'REPORT.md'. Verify claims, check for logical errors or bias. Save review to 'REVIEW.md'.",
                },
            ]
        elif category == "writing":
            steps = [
                {
                    "phase": "1. Outline",
                    "model_tier": "tier1",
                    "reason": "Structure & strategy",
                    "artifact": "OUTLINE.md",
                    "task": f"Create a detailed outline for: {task_description}. Save to 'OUTLINE.md'.",
                },
                {
                    "phase": "2. Draft",
                    "model_tier": "tier2",
                    "reason": "Content generation",
                    "artifact": "DRAFT.md",
                    "task": "Read 'OUTLINE.md'. Write the full draft following the outline. Save to 'DRAFT.md'.",
                },
                {
                    "phase": "3. Polish",
                    "model_tier": "tier3",
                    "reason": "Proofreading",
                    "artifact": "RESULT.md",
                    "task": "Read 'DRAFT.md'. Fix grammar, improve clarity, polish the text. Save final version to 'RESULT.md'.",
                },
            ]
        elif category == "creative":
            steps = [
                {
                    "phase": "1. Ideate",
                    "model_tier": "tier1",
                    "reason": "Creative thinking",
                    "artifact": "IDEAS.md",
                    "task": f"Brainstorm multiple creative approaches for: {task_description}. Save to 'IDEAS.md'.",
                },
                {
                    "phase": "2. Execute",
                    "model_tier": "tier2",
                    "reason": "Production",
                    "artifact": "RESULT.md",
                    "task": "Read 'IDEAS.md'. Pick the best idea and produce the deliverable. Save to 'RESULT.md'.",
                },
            ]
        elif category == "translation":
            steps = [
                {
                    "phase": "1. Translate",
                    "model_tier": "tier2",
                    "reason": "Language conversion",
                    "artifact": "TRANSLATION.md",
                    "task": f"Translate the following content accurately: {task_description}. Save to 'TRANSLATION.md'.",
                },
                {
                    "phase": "2. Review",
                    "model_tier": "tier3",
                    "reason": "Quality check",
                    "artifact": "RESULT.md",
                    "task": "Read 'TRANSLATION.md'. Check for accuracy, naturalness and cultural fit. Save corrected version to 'RESULT.md'.",
                },
            ]
        else:  # simple
            steps = [
                {
                    "phase": "1. Execute",
                    "model_tier": "tier3",
                    "reason": "Quick task",
                    "artifact": "RESULT.md",
                    "task": f"Complete this task directly: {task_description}. Save output to 'RESULT.md'.",
                },
            ]

        # 3. Apply custom routing overrides (if configured)
        steps = self._apply_custom_routing(category, steps)

        # 4. Display Plan & Apply Adaptation
        final_steps = self._display_plan(task_description, category.capitalize(), steps)

        # 5. Execute (if requested)
        if execute:
            self._execute_swarm(task_description, final_steps)

    def _display_plan(self, task, category, steps):
        # Calculate savings
        total_tokens = 1000
        # Baseline cost (All Tier 1)
        cost_baseline = (
            len(steps) * (self.prices["tier1"]["price"] / 1_000_000) * total_tokens
        )

        cost_optimized = 0
        final_steps = []

        print(f"\n🧠 **Golden Gear Task Planner**")
        print(f'**Task:** "{task}"')
        print(f"**Category:** {category}\n")
        print("| Phase | Assigned Agent | Model | Price/1M | Status |")
        print("| :--- | :--- | :--- | :--- | :--- |")

        for step in steps:
            tier = step["model_tier"]

            # Check for custom routing override first
            custom_model = step.get("custom_model")
            if custom_model:
                model_id = custom_model
                price = self.prices[tier]["price"]  # approximate; real price unknown
                status_display = f"📌 Custom"
            else:
                # Apply Active Adaptation
                model_id, switched, reason = self._get_stable_model(tier)

                if switched:
                    price = self.prices["tier1"]["price"]
                    status_display = f"🔄 Switched ({reason})"
                else:
                    price = self.prices[tier]["price"]
                    status_display = "✅ Optimal"

            cost_optimized += (price / 1_000_000) * total_tokens

            # Update step for execution
            step["final_model_id"] = model_id
            final_steps.append(step)

            role_name = self.prices[tier]["role"]
            print(
                f"| {step['phase']} | {role_name} | `{model_id.split('/')[-1]}` | .2f | {status_display} |"
            )

        savings_pct = (
            ((cost_baseline - cost_optimized) / cost_baseline) * 100
            if cost_baseline > 0
            else 0
        )
        print(f"\n📉 **Projected Savings:** **{savings_pct:.1f}%** 💸")
        return final_steps

    def _execute_swarm(self, original_task, steps):
        """
        Generate a structured execution plan as JSON for the host agent to carry out.

        Since sessions_spawn is an internal OpenClaw API (not callable via CLI),
        this method outputs a machine-readable instruction set that the host agent
        can parse and execute by spawning sub-agents through the internal API.
        """
        plan_output = {
            "version": "2.0",
            "task": original_task,
            "generated_at": datetime.now().isoformat(),
            "instructions": (
                "Execute each step sequentially using sessions_spawn. "
                "Each step's artifact is the input context for the next step. "
                "If a step fails, retry up to max_retries times with the error injected into the prompt."
            ),
            "steps": [],
        }

        for step in steps:
            plan_output["steps"].append(
                {
                    "phase": step["phase"],
                    "model": step["final_model_id"],
                    "system_prompt": self.prompts.get(
                        step.get("reason", "").lower().split()[0]
                        if step.get("reason")
                        else "",
                        {},
                    ).get("system", ""),
                    "task_prompt": step["task"],
                    "expected_artifact": step.get("artifact", "RESULT.md"),
                    "max_retries": 2,
                    "timeout_seconds": 60,
                }
            )

        # Write to workspace file for the host agent to pick up
        plan_path = os.path.expanduser("~/.openclaw/workspace/swarm_plan.json")
        plan_dir = os.path.dirname(plan_path)
        if not os.path.exists(plan_dir):
            os.makedirs(plan_dir, exist_ok=True)

        try:
            with open(plan_path, "w", encoding="utf-8") as f:
                json.dump(plan_output, f, indent=2, ensure_ascii=False)
        except Exception as e:
            print(f"❌ [Execute] Failed to write plan: {e}")
            return

        # Also print to stdout so the host agent can read it directly
        print(f"\n📋 **Execution Plan Generated**")
        print(f"   Saved to: {plan_path}\n")
        print("```json")
        print(json.dumps(plan_output, indent=2, ensure_ascii=False))
        print("```")
        print(f"\n🤖 **Agent Instructions:**")
        print(f"   The host agent should execute each step by calling sessions_spawn")
        print(f"   with the model and task_prompt specified above.")
        print(
            f"   Steps must run sequentially — each step depends on the previous artifact."
        )


# --- Existing Functions (Refactored for TokenRouter) ---


def fetch_models():
    """Fetch models from TokenRouter pricing API using standard library (HTTPS)."""
    try:
        req = urllib.request.Request(
            TOKENROUTER_API, headers={"User-Agent": "OpenClaw-ModelManager/2.0.0"}
        )
        with urllib.request.urlopen(req, timeout=10) as response:
            data = json.loads(response.read().decode("utf-8"))
            return data.get("data", [])
    except urllib.error.URLError as e:
        print(f"❌ [Network] Error connecting to TokenRouter API: {e}")
        return []
    except Exception as e:
        print(f"❌ [System] Unexpected error fetching models: {e}")
        return []


def filter_and_rank(models, limit=0):
    """Filter for popular/powerful models and rank them by cost-effectiveness."""
    priority_keywords = [
        "gpt-4o",
        "gpt-5",
        "claude-sonnet",
        "claude-opus",
        "claude-haiku",
        "gemini-3",
        "deepseek",
        "llama",
        "qwen3",
        "grok",
        "minimax",
    ]

    ranked = []
    others = []

    for m in models:
        model_name = m.get("model_name", "")
        is_priority = any(k in model_name for k in priority_keywords)

        if is_priority:
            ranked.append(m)
        else:
            others.append(m)

    # Sort by input price (cheapest first)
    def sort_key(x):
        return calc_input_price(x.get("model_ratio", 0))

    ranked.sort(key=sort_key)
    others.sort(key=sort_key)

    result = ranked + others
    return result[:limit] if limit > 0 else result


def display_models(models):
    """Print a markdown table of models with TokenRouter pricing."""
    print("| # | Model | Input $/1M | Output $/1M | Cache Read $/1M | Ratio |")
    print("| :--- | :--- | :--- | :--- | :--- | :--- |")

    for idx, m in enumerate(models, 1):
        model_name = m.get("model_name", "unknown")
        mr = m.get("model_ratio", 0)
        cr = m.get("completion_ratio", 1)
        cache_r = m.get("cache_ratio", None)

        in_price = calc_input_price(mr)
        out_price = calc_output_price(mr, cr)

        if cache_r is not None:
            cache_price = calc_cache_read_price(mr, cache_r)
            cache_str = f".4f"
        else:
            cache_str = "N/A"

        print(
            f"| {idx} | `{model_name}` | .4f | .4f | {cache_str} | {mr} |"
        )

    print("\nTo enable a model: `enable <index>` or `enable <model_name>`")


def enable_model(model_name, config_path):
    """Enable a model by writing the config patch into openclaw.json."""
    print(f"🔒 [Config] Preparing patch for: {model_name}")

    # Read current config
    config = load_json_safe(config_path)
    if not config and os.path.exists(config_path):
        print(f"⚠️ [Config] Warning: Could not parse existing config at {config_path}")

    # Ensure nested structure exists
    agents = config.setdefault("agents", {})
    defaults = agents.setdefault("defaults", {})
    models_dict = defaults.setdefault("models", {})
    model_section = defaults.setdefault("model", {})
    current_fallbacks = model_section.get("fallbacks", [])

    if not isinstance(current_fallbacks, list):
        current_fallbacks = []

    # Add model to models dict
    if model_name not in models_dict:
        models_dict[model_name] = {}
        print(f"📝 [Config] Adding {model_name} to models list.")
    else:
        print(f"ℹ️ [Config] Model {model_name} already in models list.")

    # Add to fallbacks
    if model_name not in current_fallbacks:
        current_fallbacks.append(model_name)
        model_section["fallbacks"] = current_fallbacks
        print(f"📝 [Config] Adding {model_name} to fallback list.")
    else:
        print(f"ℹ️ [Config] Model {model_name} already in fallback list.")

    # Write config back
    config_dir = os.path.dirname(config_path)
    if config_dir and not os.path.exists(config_dir):
        os.makedirs(config_dir, exist_ok=True)

    # Backup existing config
    if os.path.exists(config_path):
        backup_path = config_path + ".bak"
        try:
            shutil.copy2(config_path, backup_path)
        except Exception:
            pass

    try:
        with open(config_path, "w", encoding="utf-8") as f:
            json.dump(config, f, indent=2, ensure_ascii=False)
        print(f"✅ [Config] Successfully written to {config_path}")
    except Exception as e:
        print(f"❌ [Config] Failed to write config: {e}")


def check_tokenrouter_provider(config=None):
    """
    Check if a PBD TokenRouter provider is configured.

    Looks under models.providers for any provider whose baseUrl contains
    'https://open.palebluedot.ai'. Returns (True, provider_key) if found,
    (False, None) otherwise.
    """
    if config is None:
        config = load_json_safe(CONFIG_FILE)

    providers = config.get("models", {}).get("providers", {})

    for provider_name, provider_config in providers.items():
        base_url = provider_config.get("baseUrl", "")
        if "https://open.palebluedot.ai" in base_url:
            return True, provider_name

    return False, None


def print_no_provider_guidance():
    """Print guidance message when TokenRouter provider is not configured (does NOT exit)."""
    print("❌ [Config] TokenRouter provider not configured.")
    print("")
    print("To use this skill, you need a PBD TokenRouter account.")
    print("Please visit https://www.palebluedot.ai to register and log in,")
    print(
        "then go to the TokenRouter section to get your configuration (base URL and API key)."
    )
    print("")
    print("Once you have the information, provide it to the agent and run:")
    print("  setup --name <provider_name> --base-url <baseUrl> --api-key <apiKey>")
    print("")
    print("Example:")
    print(
        "  setup --name tokenrouter --base-url https://open.palebluedot.ai/v1 --api-key sk-xxx..."
    )
    print("")
    print(
        "The agent will write the configuration for you, then verify and sync models automatically."
    )


def setup_provider(provider_name, base_url, api_key):
    """
    Write a new TokenRouter provider into the config file.

    Args:
        provider_name: The provider key name (e.g., 'tokenrouter')
        base_url: The TokenRouter base URL (must contain 'https://open.palebluedot.ai')
        api_key: The user's real API key from PBD
    """
    # Validate base_url
    if "https://open.palebluedot.ai" not in base_url:
        print(f"❌ [Setup] Invalid base URL: {base_url}")
        print("   The base URL must contain 'https://open.palebluedot.ai'.")
        return False

    # Validate api_key is not a placeholder
    if not api_key or api_key.startswith("<") or api_key == "YOUR_API_KEY":
        print("❌ [Setup] Invalid API key. Please provide your real API key from PBD.")
        return False

    # Load current config
    config = load_json_safe(CONFIG_FILE)

    # Create backup before modifying
    if os.path.exists(CONFIG_FILE):
        backup_path = CONFIG_FILE + ".bak"
        try:
            shutil.copy2(CONFIG_FILE, backup_path)
            print(f"📦 [Config] Backup saved to {backup_path}")
        except Exception:
            pass

    # Write provider into config
    providers = config.setdefault("models", {}).setdefault("providers", {})
    providers[provider_name] = {
        "baseUrl": base_url,
        "apiKey": api_key,
        "api": "openai-completions",
        "models": [],
    }

    # Ensure config directory exists
    config_dir = os.path.dirname(CONFIG_FILE)
    if config_dir and not os.path.exists(config_dir):
        os.makedirs(config_dir, exist_ok=True)

    # Write config
    try:
        with open(CONFIG_FILE, "w", encoding="utf-8") as f:
            json.dump(config, f, indent=2, ensure_ascii=False)
        print(f"✅ [Setup] Provider '{provider_name}' written to {CONFIG_FILE}")
        print(f"   baseUrl: {base_url}")
        print(
            f"   apiKey:  {api_key[:8]}...{api_key[-4:]}"
            if len(api_key) > 12
            else f"   apiKey: (set)"
        )
        return True
    except Exception as e:
        print(f"❌ [Setup] Failed to write config: {e}")
        return False


def check_command():
    """
    CLI command: verify TokenRouter provider is configured.
    Returns True if found, False otherwise.
    """
    found, provider_key = check_tokenrouter_provider()
    if found:
        print(f"✅ [Check] TokenRouter provider found: '{provider_key}'")

        # Also show basic info
        config = load_json_safe(CONFIG_FILE)
        providers = config.get("models", {}).get("providers", {})
        provider_config = providers.get(provider_key, {})
        base_url = provider_config.get("baseUrl", "N/A")
        api_key = provider_config.get("apiKey", "")
        model_count = len(provider_config.get("models", []))
        print(f"   baseUrl: {base_url}")
        if api_key:
            masked = f"{api_key[:8]}...{api_key[-4:]}" if len(api_key) > 12 else "(set)"
            print(f"   apiKey:  {masked}")
        print(f"   models:  {model_count} configured")
        return True
    else:
        print("❌ [Check] No TokenRouter provider found in config.")
        print_no_provider_guidance()
        return False


def sync_models():
    """
    CLI command: re-verify provider, fetch all models from TokenRouter API,
    add them to the provider's models[] and to the allow list, then save config.
    All models are routed through the configured TokenRouter provider.
    """
    # Step 1: Verify provider exists
    config = load_json_safe(CONFIG_FILE)
    found, provider_key = check_tokenrouter_provider(config)
    if not found:
        print("❌ [Sync] Cannot sync — TokenRouter provider not configured.")
        print_no_provider_guidance()
        return False

    print(f"✅ [Sync] TokenRouter provider found: '{provider_key}'")

    # Step 2: Fetch models from API
    print("🔄 [Sync] Fetching models from TokenRouter API...")
    models = fetch_models()
    if not models:
        print(
            "❌ [Sync] Could not fetch models from API. Please check your network connection."
        )
        return False

    print(f"📋 [Sync] Retrieved {len(models)} models from API")

    # Step 3: Update the provider's models array (using partial match)
    providers = config.get("models", {}).get("providers", {})
    provider_config = providers.get(provider_key, {})
    model_names = [m.get("model_name") for m in models if m.get("model_name")]
    provider_config["models"] = model_names
    print(f"📝 [Sync] Updated provider '{provider_key}' with {len(model_names)} models")

    # Step 4: Add all models to allow list
    add_models_to_allow_list(config, models)

    # Step 5: Set default model if not already set
    agents_model = config.get("agents", {}).get("defaults", {}).get("model", {})
    if not agents_model.get("id") and model_names:
        update_current_model(config, models)

    # Step 6: Create backup and save
    if os.path.exists(CONFIG_FILE):
        backup_path = CONFIG_FILE + ".bak"
        try:
            shutil.copy2(CONFIG_FILE, backup_path)
        except Exception:
            pass

    try:
        with open(CONFIG_FILE, "w", encoding="utf-8") as f:
            json.dump(config, f, indent=2, ensure_ascii=False)
        print(f"✅ [Sync] Configuration saved to {CONFIG_FILE}")
    except Exception as e:
        print(f"❌ [Sync] Failed to save config: {e}")
        return False

    # Step 7: Display summary
    print(
        f"\n🎉 Sync complete! {len(model_names)} models are now available through TokenRouter."
    )
    print(f"   All models are routed through provider '{provider_key}'.")
    print(f"   Use `list` to view model pricing or start planning tasks.")
    return True


def update_tokenrouter_models(config, models):
    """Update the TokenRouter provider with the latest model list."""
    providers = config.get("models", {}).get("providers", {})

    # Find the TokenRouter provider (partial match on baseUrl)
    tokenrouter_provider = None
    tokenrouter_key = None
    for provider_name, provider_config in providers.items():
        base_url = provider_config.get("baseUrl", "")
        if "https://open.palebluedot.ai" in base_url:
            tokenrouter_provider = provider_config
            tokenrouter_key = provider_name
            break

    if tokenrouter_provider:
        # Update models list
        model_names = [m.get("model_name") for m in models if m.get("model_name")]
        tokenrouter_provider["models"] = model_names
        print(
            f"📝 [Config] Updated TokenRouter provider with {len(model_names)} models"
        )


def add_models_to_allow_list(config, models):
    """Add all models to the openclaw allow list."""
    # Get or create the models section
    models_section = config.setdefault("models", {})

    # Get or create the allowed models list
    allowed_models = models_section.setdefault("allowed", [])

    # Add model names to the allowed list if not already present
    added_count = 0
    for model_data in models:
        model_name = model_data.get("model_name")
        if model_name and model_name not in allowed_models:
            allowed_models.append(model_name)
            added_count += 1

    if added_count > 0:
        print(f"📝 [Config] Added {added_count} models to allowed list")


def update_current_model(config, models):
    """Update the current model to use the first available TokenRouter model."""
    # Find the first available model from our models list
    if models:
        first_model = models[0].get("model_name")
        if first_model:
            # Update the default model setting
            agents = config.setdefault("agents", {})
            defaults = agents.setdefault("defaults", {})
            model_section = defaults.setdefault("model", {})
            model_section["id"] = first_model
            print(f"📝 [Config] Updated current model to {first_model}")


def prepare_for_planning():
    """Prepare configuration before planning by verifying TokenRouter setup and updating models."""
    print("🔄 [Setup] Verifying TokenRouter configuration...")

    # Load current config
    config = load_json_safe(CONFIG_FILE)

    # Step 1: Check if TokenRouter provider is configured — abort if not
    found, provider_key = check_tokenrouter_provider(config)
    if not found:
        print_no_provider_guidance()
        sys.exit(1)

    print(f"✅ [Setup] TokenRouter provider found: '{provider_key}'")

    # Step 2: Fetch current models from API
    models = fetch_models()
    if not models:
        print("⚠️ [Setup] Could not fetch models from API")
        return config  # Still return config even if models fetch failed

    # Step 3: Update TokenRouter models
    update_tokenrouter_models(config, models)

    # Step 4: Add all models to allow list
    add_models_to_allow_list(config, models)

    # Step 5: Update current model if not already set
    agents_model = config.get("agents", {}).get("defaults", {}).get("model", {})
    if not agents_model.get("id"):
        update_current_model(config, models)

    # Write the updated config back
    config_dir = os.path.dirname(CONFIG_FILE)
    if config_dir and not os.path.exists(config_dir):
        os.makedirs(config_dir, exist_ok=True)

    # Backup existing config
    if os.path.exists(CONFIG_FILE):
        backup_path = CONFIG_FILE + ".bak"
        try:
            shutil.copy2(CONFIG_FILE, backup_path)
        except Exception:
            pass

    try:
        with open(CONFIG_FILE, "w", encoding="utf-8") as f:
            json.dump(config, f, indent=2, ensure_ascii=False)
        print(f"✅ [Setup] Configuration updated successfully")
    except Exception as e:
        print(f"❌ [Setup] Failed to write config: {e}")

    return config


def main():
    if len(sys.argv) < 2:
        print("Usage: manage_models.py <check|setup|sync|list|enable|plan> [options]")
        print("")
        print("Commands:")
        print(
            "  check                          Verify TokenRouter provider configuration"
        )
        print(
            "  setup --name N --base-url U --api-key K   Write TokenRouter provider config"
        )
        print(
            "  sync                           Fetch models and sync to provider & allow list"
        )
        print("  list                           Show all models with pricing")
        print("  enable <index|name>            Enable a model")
        print("  plan <task> [--execute]        Plan a task with smart routing")
        return

    # Argument parsing
    cmd_args = sys.argv[1:]
    action = cmd_args[0]

    # --- Commands that do NOT require pre-existing provider ---

    if action == "check":
        check_command()
        return

    if action == "setup":
        # Parse setup arguments
        parser = argparse.ArgumentParser(prog="manage_models.py setup")
        parser.add_argument("--name", default="tokenrouter", help="Provider name")
        parser.add_argument("--base-url", required=True, help="TokenRouter base URL")
        parser.add_argument("--api-key", required=True, help="API key from PBD")
        try:
            args = parser.parse_args(cmd_args[1:])
        except SystemExit:
            return
        success = setup_provider(args.name, args.base_url, args.api_key)
        if success:
            print("")
            print(
                "Next step: run `check` to verify, then `sync` to fetch and add all models."
            )
        return

    # --- Commands that REQUIRE provider to exist ---

    found, provider_key = check_tokenrouter_provider()
    if not found:
        print_no_provider_guidance()
        return

    if action == "sync":
        sync_models()
        return

    if action == "plan":
        # Prepare configuration before planning (auto-syncs models)
        prepare_for_planning()

        execute = "--execute" in cmd_args or "-x" in cmd_args
        # Extract task description safely
        task_words = [arg for arg in cmd_args[1:] if not arg.startswith("-")]
        if not task_words:
            print("Error: Please provide a task description.")
            return
        task = " ".join(task_words)

        planner = TaskPlanner()
        planner.plan(task, execute=execute)
        return

    if action == "enable":
        if len(cmd_args) < 2:
            print("Error: Please specify a model index or name to enable.")
            return

        target = cmd_args[1]

        if target.isdigit():
            # Need to fetch models to resolve index
            models = fetch_models()
            if not models:
                print(
                    "❌ Cannot fetch model list. Try enabling by name instead: enable <model_name>"
                )
                return
            sorted_models = filter_and_rank(models)
            idx = int(target) - 1
            if 0 <= idx < len(sorted_models):
                selected_model_name = sorted_models[idx]["model_name"]
            else:
                print(f"Error: Index {target} out of range (1-{len(sorted_models)}).")
                return
        else:
            # Direct name, no need to fetch
            selected_model_name = target

        enable_model(selected_model_name, CONFIG_FILE)
        return

    if action == "list":
        # Fetch models
        models = fetch_models()
        if not models:
            return
        sorted_models = filter_and_rank(models)
        display_models(sorted_models)
        return

    print(
        f"Error: Unknown action '{action}'. Use: check, setup, sync, list, enable, plan"
    )


def enhanced_task_classification(task_description):
    """Enhanced task classification with Chinese + English keyword scoring."""
    task_lower = task_description.lower()

    patterns = {
        "coding": [
            "code",
            "program",
            "script",
            "debug",
            "function",
            "algorithm",
            "api",
            "database",
            "app",
            "test",
            "bug",
            "compile",
            "deploy",
            "refactor",
            "代码",
            "编程",
            "脚本",
            "程序",
            "调试",
            "测试",
            "开发",
            "编码",
            "接口",
            "部署",
            "重构",
            "修复",
            "函数",
            "算法",
            "数据库",
            "前端",
            "后端",
            "全栈",
            "爬虫",
            "框架",
            "模块",
        ],
        "writing": [
            "write",
            "article",
            "blog",
            "content",
            "story",
            "email",
            "letter",
            "essay",
            "copywriting",
            "documentation",
            "draft",
            "写作",
            "文章",
            "博客",
            "内容",
            "故事",
            "邮件",
            "信件",
            "文案",
            "文档",
            "稿件",
            "撰写",
            "起草",
        ],
        "analysis": [
            "analyze",
            "compare",
            "evaluate",
            "research",
            "study",
            "report",
            "data",
            "statistics",
            "insight",
            "metrics",
            "分析",
            "对比",
            "评估",
            "研究",
            "调研",
            "报告",
            "数据",
            "统计",
            "洞察",
            "指标",
            "复盘",
        ],
        "translation": [
            "translate",
            "convert language",
            "interpretation",
            "localize",
            "翻译",
            "转换语言",
            "本地化",
            "国际化",
            "多语言",
            "中译英",
            "英译中",
        ],
        "creative": [
            "creative",
            "brainstorm",
            "idea",
            "design",
            "artistic",
            "logo",
            "illustration",
            "prototype",
            "wireframe",
            "创意",
            "头脑风暴",
            "点子",
            "设计",
            "艺术",
            "原型",
            "线框图",
            "灵感",
            "构思",
        ],
        "simple": [
            "simple",
            "quick",
            "basic",
            "summarize",
            "list",
            "count",
            "lookup",
            "define",
            "explain",
            "简单",
            "快速",
            "基础",
            "总结",
            "列出",
            "计数",
            "查询",
            "定义",
            "解释",
            "概括",
        ],
    }

    scores = {}
    for category, keywords in patterns.items():
        score = sum(1 for kw in keywords if kw in task_lower)
        if score > 0:
            scores[category] = score

    if scores:
        return max(scores, key=scores.get)
    return "simple"


if __name__ == "__main__":
    if len(sys.argv) == 1:
        print("Usage: manage_models.py <check|setup|sync|list|enable|plan> [options]")
    main()

FILE:scripts/prompts.json
{
  "version": "1.0.0",
  "roles": {
    "architect": {
      "system": "You are an Expert Software Architect. Your goal is to design robust, scalable, and secure systems.",
      "task_template": "Design the architecture and file structure for: '{task_description}'. \n\nOutput Requirements:\n1. Create a file named 'SPEC.md' using the 'write' tool.\n2. The SPEC.md must contain: Project Structure, Core Classes/Functions, and Data Flow.\n3. Do not output verbose chatter, just write the file."
    },
    "coder": {
      "system": "You are a Senior Python Developer. You write clean, efficient, and typed code.\nCRITICAL RULE: ALWAYS include unit tests using `unittest` for generated code.",
      "task_template": "Read 'SPEC.md' (if available) to understand the design. Write the implementation for: '{task_description}'. \n\nOutput Requirements:\n1. Use the 'write' tool to save the code to appropriate files (e.g., main.py).\n2. Ensure code is runnable and includes basic error handling.\n3. Verify file creation before finishing."
    },
    "auditor": {
      "system": "You are a Security Auditor and QA Engineer. You are paranoid about safety and privacy.",
      "task_template": "Read the generated code files in the current directory. Perform a security and logic audit. \n\nOutput Requirements:\n1. Create a file named 'AUDIT.md' using the 'write' tool.\n2. List any vulnerabilities, bugs, or privacy risks.\n3. Assign a 'Pass/Fail' grade."
    },
    "planner": {
      "system": "You are a Strategic Planner.",
      "task_template": "Create a detailed execution plan for: '{task_description}'. \n\nOutput Requirements:\n1. Create a file named 'PLAN.md' using the 'write' tool.\n2. Break down the task into actionable steps."
    },
    "drafter": {
      "system": "You are a Content Creator.",
      "task_template": "Read 'PLAN.md'. Execute the writing task. \n\nOutput Requirements:\n1. Create a file named 'RESULT.md' using the 'write' tool."
    }
  }
}
FILE:scripts/smart_find.py
#!/usr/bin/env python3

import os
import sys
from pathlib import Path
from difflib import SequenceMatcher
import argparse

def similarity_score(query, text):
    """Calculate similarity score between query and text."""
    return SequenceMatcher(None, query.lower(), text.lower()).ratio()

def find_best_match(query, start_path='.'):
    """Find the best matching file based on query."""
    matches = []
    query = query.lower()
    
    try:
        for root, dirs, files in os.walk(start_path):
            # Skip hidden directories and .git
            dirs[:] = [d for d in dirs if not d.startswith('.') and d != '.git']
            
            for file in files:
                # Skip hidden files
                if file.startswith('.'):
                    continue
                    
                full_path = os.path.join(root, file)
                rel_path = os.path.relpath(full_path, start_path)
                
                # Calculate similarity scores for both filename and path
                filename_score = similarity_score(query, file)
                path_score = similarity_score(query, rel_path)
                
                # Use the better of the two scores
                score = max(filename_score, path_score)
                
                # If exact match found in filename, return immediately
                if filename_score == 1.0:
                    return [(full_path, 1.0)]
                
                # Keep matches with score > 0.2
                if score > 0.2:
                    matches.append((full_path, score))
    
    except Exception as e:
        print(f"Error while searching: {str(e)}", file=sys.stderr)
        return []
    
    # Sort by score (highest first) and path length (shortest first)
    matches.sort(key=lambda x: (-x[1], len(x[0])))
    return matches

def read_file_content(file_path):
    """Read and return file content safely."""
    try:
        with open(file_path, 'r', encoding='utf-8') as f:
            return f.read()
    except UnicodeDecodeError:
        return "[ERROR] File appears to be binary or uses an unsupported encoding."
    except Exception as e:
        return f"[ERROR] Could not read file: {str(e)}"

def main():
    parser = argparse.ArgumentParser(description='Smart file finder with fuzzy matching')
    parser.add_argument('query', help='Search query (filename or keyword)')
    args = parser.parse_args()
    
    if not args.query:
        print("Please provide a search query.")
        return 1
    
    matches = find_best_match(args.query)
    
    if not matches:
        print(f"No files found matching '{args.query}'")
        return 1
    
    best_match = matches[0]
    best_match_path, score = best_match
    
    print(f"\nBest match: {best_match_path} (similarity: {score:.2f})")
    
    if len(matches) > 1:
        print("\nOther potential matches:")
        for path, score in matches[1:5]:  # Show up to 4 alternatives
            print(f"- {path} (similarity: {score:.2f})")
    
    print("\nFile content:")
    print("-" * 80)
    content = read_file_content(best_match_path)
    print(content)
    print("-" * 80)
    
    return 0

if __name__ == '__main__':
    sys.exit(main())
FILE:scripts/smart_map.py
#!/usr/bin/env python3
"""
Cognitive Map Generator - Creates a high-level summary of a Python codebase using AST analysis.
"""

import os
import ast
import sys

class CodebaseMapper:
    def __init__(self):
        self.skip_dirs = {'.git', '__pycache__', 'node_modules', 'venv', 'env'}
        self.skip_files = {'.DS_Store'}

    def should_skip(self, name: str) -> bool:
        """Check if a file or directory should be skipped."""
        return name.startswith('.') or name in self.skip_dirs or name in self.skip_files

    def parse_python_file(self, file_path: str) -> list[str]:
        """Parse a Python file and extract its structure as a list of strings."""
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                content = f.read()
            tree = ast.parse(content)
        except Exception as e:
            return [f"⚠️ Error parsing: {str(e)}"]

        lines = []
        
        # Helper for recursive traversal
        def visit(nodes, level=0):
            prefix = "  " * level
            for node in nodes:
                if isinstance(node, ast.ClassDef):
                    bases = [b.id for b in node.bases if isinstance(b, ast.Name)]
                    base_str = f"({', '.join(bases)})" if bases else ""
                    doc = ast.get_docstring(node)
                    doc_summary = f" - \"{doc.splitlines()[0]}\"" if doc else ""
                    lines.append(f"{prefix}📦 class {node.name}{base_str}{doc_summary}")
                    # Recurse into class body
                    visit(node.body, level + 1)
                    
                elif isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
                    args = [a.arg for a in node.args.args]
                    # Filter out 'self' for brevity in methods
                    if args and args[0] == 'self':
                        args = args[1:]
                    args_str = ", ".join(args)
                    
                    doc = ast.get_docstring(node)
                    doc_summary = ""
                    # Only show docstring for top-level functions or major methods to save space
                    if doc and level <= 1: 
                        doc_summary = f" - \"{doc.splitlines()[0]}\""
                    
                    icon = "🔹" if level > 0 else "ƒ "
                    lines.append(f"{prefix}{icon}def {node.name}({args_str}){doc_summary}")

        visit(tree.body)
        return lines

    def generate_markdown(self, base_path: str) -> str:
        """Generate a markdown representation of the codebase structure."""
        output = []
        
        for root, dirs, files in os.walk(base_path):
            # Filter dirs in-place
            dirs[:] = [d for d in dirs if not self.should_skip(d)]
            dirs.sort()
            
            # Calculate indentation
            rel_path = os.path.relpath(root, base_path)
            if rel_path == '.':
                level = 0
            else:
                level = len(rel_path.split(os.sep))
                # Add directory name
                indent = '  ' * (level - 1)
                output.append(f'{indent}📂 {os.path.basename(root)}/')
            
            indent = '  ' * level
            
            for file in sorted(files):
                if self.should_skip(file):
                    continue
                    
                file_path = os.path.join(root, file)
                
                # Add file name
                output.append(f'{indent}📄 {file}')
                
                # If Python, add AST structure
                if file.endswith('.py'):
                    structure = self.parse_python_file(file_path)
                    for line in structure:
                        output.append(f'{indent}  {line}')
        
        return '\n'.join(output)

def main():
    if len(sys.argv) > 1:
        target_dir = sys.argv[1]
    else:
        target_dir = os.getcwd()
        
    mapper = CodebaseMapper()
    print(f"🗺️ Cognitive Map of: {target_dir}\n")
    print(mapper.generate_markdown(target_dir))

if __name__ == '__main__':
    main()