scripts.ask_openrouter

Reference › scripts › ask_openrouter

A command-line tool, built on typer, that sends a question to any model hosted on OpenRouter and optionally attaches scoped context. A human runs it; no module or test imports it. It reuses the OpenRouterAdapter from the provider layer (client and the protocols in core.protocols) for non-streaming calls.

Invocation: main takes the prompt as an optional argument and reads standard input when it is omitted and input is piped. Options are --model (required unless listing), --system, --context-file and --context (both repeatable), --max-context-tokens (default 8000), --temperature (0.2), --max-tokens (4000), --stream, --json, --list-models and a free-only filter for the catalog. Results go to stdout; status lines go to stderr.

run_cli holds the flow and returns an exit code, with 1 for configuration or usage errors. The key comes from the OPENROUTER_API_KEY variable, falling back to a dotenv file (resolve_api_key). --list-models fetches the catalog and prints a table (parse_models, format_models_table). Otherwise assemble_context reads the named files, passes the text through redact_context, caps it by approximate tokens (characters divided by four) and reports how many lines were redacted and how many tokens were sent. run_completion makes the call, or iter_sse_chunks parses the streamed events. Each run appends an entry to an audit log (build_audit_entry, append_audit_log, default_audit_log_path) recording the model, prompt, context file names and byte and token counts.

At a glance

Class diagram

Dependency diagram

API

Classes

ContextMeta (dataclass)

Decorators: dataclass · Line 171

Metadata about an assembled context payload (for the stderr notice and the audit log — never the context text itself).

ModelRow (dataclass)

Decorators: dataclass · Line 263

One row of the OpenRouter model catalog.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/ask_openrouter.py

#!/usr/bin/env python3
"""Ask any OpenRouter-hosted model a question, with opt-in scoped context.

Design: ``plans/openrouter-tool-design.md`` (task #26). Distinct from the
internal volk OpenRouter routing (``orchestrata_ultimate.llm.client`` /
``_register_openrouter``) — this is a Claude-Code-invokable CLI for
cross-vendor second opinions, judge panels, and cheap delegation. It reuses
the settled :class:`OpenRouterAdapter` for the actual chat call rather than
reimplementing the HTTP client.

Usage:
    uv run python scripts/ask_openrouter.py "PROMPT" --model anthropic/claude-sonnet-4-6
    echo "PROMPT" | uv run python scripts/ask_openrouter.py --model deepseek/deepseek-chat
    uv run python scripts/ask_openrouter.py --list-models [--free]

Context is OPT-IN ONLY: by default no project context is sent. Passing
``--context-file``/``--context`` runs a mandatory redaction + token-cap pass
before anything leaves the process — see ``redact_context``/``assemble_context``
below and the ``.claude/skills/ask-openrouter`` skill for the full contract.

Every source/network boundary in this module is dependency-injected
(``run_cli``'s ``env``/``dotenv_loader``/``complete_call``/``models_fetch``/
``stream_fetch``/``audit_log_path`` keywords) so tests exercise the real
control flow without live API calls or mocking — the only sanctioned test
double in this project is the LLM/network boundary itself.
"""

import asyncio
import hashlib
import json
import os
import re
import sys
from collections.abc import Awaitable, Callable, Iterable, Iterator, Mapping
from dataclasses import dataclass
from datetime import UTC, datetime
from pathlib import Path
from typing import Any, TextIO

import httpx
import typer

from orchestrata_ultimate.core.protocols import (
    Message,
    ProviderConfig,
    RawProviderResponse,
)
from orchestrata_ultimate.providers.adapters.openrouter_adapter import (
    OPENROUTER_BASE_URL,
    OpenRouterAdapter,
)

MODELS_URL = f"{OPENROUTER_BASE_URL}/models"
CHAT_URL = f"{OPENROUTER_BASE_URL}/chat/completions"
AUDIT_LOG_REL_PATH = Path(".orchestrata") / "openrouter_audit.log"

CompleteCall = Callable[
    [list[Message], str, dict[str, Any]], Awaitable[RawProviderResponse]
]
StreamFetch = Callable[[list[Message], str, dict[str, Any], str], Iterator[str]]
ModelsFetch = Callable[[str], dict[str, Any]]
DotenvLoader = Callable[[], dict[str, str]]

app = typer.Typer(add_completion=False, no_args_is_help=True)


# ============================================================================
# Credential resolution — process env first, then .env (repo root)
# ============================================================================


_DEFAULT_DOTENV_PATHS = (".env", "/opt/orchestrata/.env")


def _fallback_parse_dotenv(
    paths: Iterable[str] = _DEFAULT_DOTENV_PATHS,
) -> dict[str, str]:
    """Minimal .env parser mirroring ``llm.client._parse_dotenv_file``.

    Used only if ``orchestrata_ultimate.llm.client`` becomes unimportable —
    the primary path reuses that module's real ``_load_dotenv`` directly.
    ``paths`` defaults to the real lookup locations; tests pass explicit
    paths instead.
    """
    loaded: dict[str, str] = {}
    for env_path in paths:
        path = Path(env_path)
        if not path.exists():
            continue
        for raw_line in path.read_text(encoding="utf-8").splitlines():
            line = raw_line.strip()
            if not line or line.startswith("#") or "=" not in line:
                continue
            key, value = line.split("=", 1)
            value = value.strip().strip('"').strip("'")
            key = key.strip()
            if value and key not in os.environ:
                loaded[key] = value
    return loaded


def _default_dotenv_loader() -> dict[str, str]:
    """Reuse ``llm.client._load_dotenv`` (repo root + /opt/orchestrata) when
    importable; otherwise fall back to a minimal inline parser."""
    try:
        from orchestrata_ultimate.llm.client import _load_dotenv
    except ImportError:
        return _fallback_parse_dotenv()
    return _load_dotenv()


def resolve_api_key(
    env: Mapping[str, str] | None = None,
    dotenv_loader: DotenvLoader | None = None,
) -> str:
    """Resolve ``OPENROUTER_API_KEY``: process env first, then ``.env``.

    Both sources are injectable so tests can exercise "key missing" and
    "key found via .env" deterministically, independent of the real
    process environment or the real repo-root ``.env`` file.
    """
    active_env = env if env is not None else os.environ
    key = active_env.get("OPENROUTER_API_KEY", "")
    if key:
        return key
    loader = dotenv_loader or _default_dotenv_loader
    return loader().get("OPENROUTER_API_KEY", "")


# ============================================================================
# Context redaction, capping, and assembly (security-critical — opt-in only)
# ============================================================================

_REDACT_PATTERN = re.compile(r"(API_KEY|SECRET|TOKEN|PASSWORD|Bearer\s)", re.IGNORECASE)


def redact_context(text: str) -> tuple[str, int]:
    """Strip any line that looks like it carries a credential.

    Case-insensitive match on ``API_KEY|SECRET|TOKEN|PASSWORD|Bearer `` (a
    literal space after Bearer, so only header-shaped tokens match) anywhere
    in the line; the whole line is replaced with ``[REDACTED]``.

    Returns:
        (redacted_text, n_lines_redacted).
    """
    n_redacted = 0
    out_lines: list[str] = []
    for line in text.splitlines():
        if _REDACT_PATTERN.search(line):
            out_lines.append("[REDACTED]")
            n_redacted += 1
        else:
            out_lines.append(line)
    return "\n".join(out_lines), n_redacted


def validate_context_path(path: Path) -> None:
    """Refuse .env-like files and anything under a ``.git/`` directory.

    Raises:
        ValueError: if the path is disallowed as a context source.
    """
    if path.name.startswith(".env"):
        raise ValueError(f"refusing to read env file as context: {path}")
    if ".git" in path.parts:
        raise ValueError(f"refusing to read .git-tracked path as context: {path}")


@dataclass
class ContextMeta:
    """Metadata about an assembled context payload (for the stderr notice
    and the audit log — never the context text itself)."""

    n_files: int
    n_lines_redacted: int
    n_tokens: int
    truncated: bool
    truncated_chars: int = 0


def assemble_context(
    files: Iterable[Path],
    inline: Iterable[str],
    max_tokens: int,
) -> tuple[str, ContextMeta]:
    """Read, redact, and token-cap opt-in context from files + inline text.

    Raises:
        ValueError: if any file fails :func:`validate_context_path`.
    """
    file_list = list(files)
    for f in file_list:
        validate_context_path(f)

    sections: list[str] = []
    n_lines_redacted = 0
    for f in file_list:
        raw = f.read_text(encoding="utf-8", errors="replace")
        clean, n = redact_context(raw)
        n_lines_redacted += n
        sections.append(f"--- {f} ---\n{clean}")
    for text in inline:
        clean, n = redact_context(text)
        n_lines_redacted += n
        sections.append(clean)

    combined = "\n\n".join(sections)
    truncated = False
    truncated_chars = 0
    if max_tokens > 0:
        max_chars = max_tokens * 4
        if len(combined) > max_chars:
            truncated_chars = len(combined) - max_chars
            combined = (
                combined[:max_chars] + f"\n[...truncated {truncated_chars} chars...]"
            )
            truncated = True

    meta = ContextMeta(
        n_files=len(file_list),
        n_lines_redacted=n_lines_redacted,
        n_tokens=len(combined) // 4,
        truncated=truncated,
        truncated_chars=truncated_chars,
    )
    return combined, meta


def build_messages(system: str | None, user: str, context: str | None) -> list[Message]:
    """Build the OpenAI-compatible message list for the completion call."""
    messages: list[Message] = []
    if system:
        messages.append(Message(role="system", content=system))
    content = user if not context else f"{user}\n\n---\nContext:\n{context}"
    messages.append(Message(role="user", content=content))
    return messages


def resolve_prompt(
    prompt_arg: str | None,
    list_models: bool,
    stdin_is_tty: bool,
    stdin_reader: Callable[[], str],
) -> str | None:
    """Resolve the prompt: the positional arg wins; otherwise read stdin
    (unless it's a tty or we're only listing models). Returns None if no
    prompt could be resolved."""
    if prompt_arg is not None:
        return prompt_arg
    if list_models or stdin_is_tty:
        return None
    data = stdin_reader().strip()
    return data or None


# ============================================================================
# Model catalog (--list-models)
# ============================================================================


@dataclass
class ModelRow:
    """One row of the OpenRouter model catalog."""

    id: str
    context_length: int
    price_prompt: str
    price_completion: str


def parse_models(payload: dict[str, Any]) -> list[ModelRow]:
    """Extract id/context_length/pricing from a ``GET /models`` JSON body."""
    rows: list[ModelRow] = []
    for entry in payload.get("data", []) or []:
        pricing = entry.get("pricing") or {}
        rows.append(
            ModelRow(
                id=str(entry.get("id", "")),
                context_length=int(entry.get("context_length") or 0),
                price_prompt=str(pricing.get("prompt", "0")),
                price_completion=str(pricing.get("completion", "0")),
            )
        )
    return rows


def _is_free(row: ModelRow) -> bool:
    if row.id.endswith(":free"):
        return True
    try:
        return float(row.price_prompt) == 0.0 and float(row.price_completion) == 0.0
    except ValueError:
        return False


def filter_free_models(rows: list[ModelRow]) -> list[ModelRow]:
    """Keep only free-tier models (``:free`` suffix or zero pricing)."""
    return [r for r in rows if _is_free(r)]


def format_models_table(rows: list[ModelRow]) -> str:
    """Render model rows as an aligned plain-text table."""
    header = f"{'MODEL':50} {'CONTEXT':>10} {'PROMPT $/1M':>14} {'COMPLETION $/1M':>16}"
    lines = [header]
    for r in rows:
        lines.append(
            f"{r.id:50} {r.context_length:>10} {r.price_prompt:>14} {r.price_completion:>16}"
        )
    return "\n".join(lines)


def _fetch_models_payload(api_key: str) -> dict[str, Any]:
    """Real network call: GET the OpenRouter model catalog."""
    headers = {"Authorization": f"Bearer {api_key}"}
    response = httpx.get(MODELS_URL, headers=headers, timeout=30.0)
    response.raise_for_status()
    payload: dict[str, Any] = response.json()
    return payload


# ============================================================================
# Completion (non-streaming) — reuses OpenRouterAdapter
# ============================================================================


def run_completion(
    adapter_or_call: OpenRouterAdapter | CompleteCall,
    messages: list[Message],
    model: str,
    params: dict[str, Any],
) -> RawProviderResponse:
    """Execute a completion via an :class:`OpenRouterAdapter` (uses its
    ``.complete``) or a raw async callable with the same signature.

    Tests inject a plain async function matching that signature — the
    sanctioned LLM-boundary double — so no live API call is needed.
    """
    call: CompleteCall = (
        adapter_or_call.complete
        if isinstance(adapter_or_call, OpenRouterAdapter)
        else adapter_or_call
    )
    return asyncio.run(call(messages, model, params))


def _default_complete_call(api_key: str) -> OpenRouterAdapter:
    cfg = ProviderConfig(
        api_key=api_key, base_url=OPENROUTER_BASE_URL, timeout_seconds=180.0
    )
    return OpenRouterAdapter(cfg)


# ============================================================================
# Streaming (SSE) — the adapter has no streaming path; written directly here
# ============================================================================


def iter_sse_chunks(lines: Iterable[str]) -> Iterator[str]:
    """Parse OpenAI-compatible SSE lines into content-delta strings.

    Stops at ``data: [DONE]``; ignores non-``data:`` lines (blank
    keep-alives) and malformed JSON chunks.
    """
    for line in lines:
        if not line.startswith("data: "):
            continue
        data = line[len("data: ") :].strip()
        if data == "[DONE]":
            return
        try:
            payload = json.loads(data)
        except json.JSONDecodeError:
            continue
        choices = payload.get("choices") or []
        if not choices:
            continue
        delta = choices[0].get("delta") or {}
        content = delta.get("content")
        if content:
            yield content


def _default_stream_fetch(
    messages: list[Message],
    model: str,
    params: dict[str, Any],
    api_key: str,
) -> Iterator[str]:
    """Real network call: POST with ``stream: true`` and yield token deltas."""
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json",
    }
    payload = {
        "model": model,
        "messages": [{"role": m.role, "content": m.content} for m in messages],
        "stream": True,
        **params,
    }
    with httpx.stream(
        "POST", CHAT_URL, headers=headers, json=payload, timeout=180.0
    ) as response:
        response.raise_for_status()
        yield from iter_sse_chunks(response.iter_lines())


# ============================================================================
# Audit log — sha256 + sizes only, NEVER the prompt/context text or the key
# ============================================================================


def build_audit_entry(
    model: str,
    prompt: str,
    context_files: list[str],
    bytes_sent: int,
    input_tokens: int,
    output_tokens: int,
) -> dict[str, Any]:
    """Build one audit record. Only the prompt's sha256 is logged — never
    the prompt or context text, and never the API key."""
    return {
        "ts": datetime.now(UTC).isoformat(),
        "model": model,
        "prompt_sha256": hashlib.sha256(prompt.encode("utf-8")).hexdigest(),
        "context_files": context_files,
        "bytes_sent": bytes_sent,
        "input_tokens": input_tokens,
        "output_tokens": output_tokens,
    }


def append_audit_log(log_path: Path, entry: dict[str, Any]) -> None:
    """Append one JSON line to the audit log, creating its directory if
    needed."""
    log_path.parent.mkdir(parents=True, exist_ok=True)
    with log_path.open("a", encoding="utf-8") as fh:
        fh.write(json.dumps(entry) + "\n")


def default_audit_log_path() -> Path:
    return Path.cwd() / AUDIT_LOG_REL_PATH


# ============================================================================
# Core orchestration — every I/O boundary is an injectable keyword so tests
# drive the full control flow without live network calls or mocking.
# ============================================================================


def run_cli(
    prompt: str | None,
    *,
    model: str | None,
    system: str | None,
    context_files: list[Path],
    context_inline: list[str],
    max_context_tokens: int,
    temperature: float,
    max_tokens: int,
    stream: bool,
    json_output: bool,
    list_models: bool,
    free_only: bool,
    env: Mapping[str, str] | None = None,
    dotenv_loader: DotenvLoader | None = None,
    complete_call: OpenRouterAdapter | CompleteCall | None = None,
    models_fetch: ModelsFetch | None = None,
    stream_fetch: StreamFetch | None = None,
    audit_log_path: Path | None = None,
    stdout: TextIO = sys.stdout,
    stderr: TextIO = sys.stderr,
) -> int:
    """Resolve credentials, assemble context, run the completion (or catalog
    listing), print output, and write the audit log. Returns a process exit
    code (0 success, 1 user/config error)."""
    api_key = resolve_api_key(env, dotenv_loader)
    if not api_key:
        print(
            "Error: OPENROUTER_API_KEY not set (checked process env and .env)",
            file=stderr,
        )
        return 1

    if list_models:
        fetch = models_fetch or _fetch_models_payload
        rows = parse_models(fetch(api_key))
        if free_only:
            rows = filter_free_models(rows)
        print(format_models_table(rows), file=stdout)
        return 0

    if model is None:
        print("Error: --model is required (use --list-models to browse)", file=stderr)
        return 1
    if prompt is None:
        print("Error: no prompt provided (pass PROMPT or pipe via stdin)", file=stderr)
        return 1

    context_text: str | None = None
    context_file_names: list[str] = [str(p) for p in context_files]
    if context_files or context_inline:
        try:
            context_text, meta = assemble_context(
                context_files, context_inline, max_context_tokens
            )
        except ValueError as exc:
            print(f"Error: {exc}", file=stderr)
            return 1
        print(
            f"redacted {meta.n_lines_redacted} lines, sent {meta.n_tokens} tokens "
            f"of context from {meta.n_files} files",
            file=stderr,
        )

    messages = build_messages(system, prompt, context_text)
    params: dict[str, Any] = {"temperature": temperature, "max_tokens": max_tokens}
    bytes_sent = sum(len(m.content.encode("utf-8")) for m in messages)
    log_path = audit_log_path or default_audit_log_path()

    if stream:
        fetch_stream = stream_fetch or _default_stream_fetch
        parts: list[str] = []
        for chunk in fetch_stream(messages, model, params, api_key):
            print(chunk, end="", file=stdout, flush=True)
            parts.append(chunk)
        print(file=stdout)
        full_text = "".join(parts)
        entry = build_audit_entry(
            model,
            prompt,
            context_file_names,
            bytes_sent,
            len(prompt) // 4,
            len(full_text) // 4,
        )
        append_audit_log(log_path, entry)
        return 0

    call = (
        complete_call if complete_call is not None else _default_complete_call(api_key)
    )
    resp = run_completion(call, messages, model, params)
    entry = build_audit_entry(
        model,
        prompt,
        context_file_names,
        bytes_sent,
        resp.input_tokens,
        resp.output_tokens,
    )
    append_audit_log(log_path, entry)
    if json_output:
        print(
            json.dumps(
                {
                    "content": resp.content,
                    "model": resp.model,
                    "input_tokens": resp.input_tokens,
                    "output_tokens": resp.output_tokens,
                }
            ),
            file=stdout,
        )
    else:
        print(resp.content, file=stdout)
    return 0


# ============================================================================
# Typer CLI — thin wiring only; all logic lives in run_cli() above
# ============================================================================


@app.command()
def main(
    prompt: str | None = typer.Argument(
        None, help="The prompt to send. Reads stdin if omitted (and not a tty)."
    ),
    model: str | None = typer.Option(
        None, "--model", help="OpenRouter model id, e.g. anthropic/claude-sonnet-4-6."
    ),
    system: str | None = typer.Option(
        None, "--system", help="Optional system message."
    ),
    context_file: list[Path] = typer.Option(
        [],
        "--context-file",
        help="Path to include as redacted, capped context (repeatable).",
    ),
    context: list[str] = typer.Option(
        [], "--context", help="Inline context text (repeatable)."
    ),
    max_context_tokens: int = typer.Option(
        8000,
        "--max-context-tokens",
        help="Hard cap on context tokens (approx chars/4).",
    ),
    temperature: float = typer.Option(0.2, "--temperature"),
    max_tokens: int = typer.Option(4000, "--max-tokens"),
    stream: bool = typer.Option(
        False, "--stream", help="Stream tokens to stdout via SSE."
    ),
    json_output: bool = typer.Option(
        False,
        "--json",
        help="Print {content, model, input_tokens, output_tokens} JSON.",
    ),
    list_models: bool = typer.Option(
        False,
        "--list-models",
        help="Print the OpenRouter catalog (id/context/pricing) and exit.",
    ),
    free: bool = typer.Option(
        False, "--free", help="With --list-models, only show free-tier models."
    ),
) -> None:
    """Ask any OpenRouter-hosted model (GPT/Gemini/Kimi/Qwen/DeepSeek/...) a
    question, with opt-in, redacted, token-capped project context."""
    resolved_prompt = resolve_prompt(
        prompt, list_models, sys.stdin.isatty(), sys.stdin.read
    )
    exit_code = run_cli(
        resolved_prompt,
        model=model,
        system=system,
        context_files=list(context_file),
        context_inline=list(context),
        max_context_tokens=max_context_tokens,
        temperature=temperature,
        max_tokens=max_tokens,
        stream=stream,
        json_output=json_output,
        list_models=list_models,
        free_only=free,
    )
    raise typer.Exit(code=exit_code)


if __name__ == "__main__":
    app()