From 9cc2a41b74364a3042083e4a57e51d5c41d192a7 Mon Sep 17 00:00:00 2001 From: binichallein <132759743+binichallein@users.noreply.github.com> Date: Fri, 15 May 2026 13:46:04 +0800 Subject: [PATCH] feat(cli): add non-interactive usage command --- docs/en/faq.md | 2 +- docs/en/reference/kimi-command.md | 13 +++ docs/zh/faq.md | 2 +- docs/zh/reference/kimi-command.md | 13 +++ src/kimi_cli/cli/_lazy_group.py | 2 + src/kimi_cli/cli/usage.py | 139 +++++++++++++++++++++++++++++ src/kimi_cli/ui/shell/usage.py | 16 ++-- tests/core/test_startup_imports.py | 5 +- tests/core/test_usage_cli.py | 74 +++++++++++++++ 9 files changed, 255 insertions(+), 11 deletions(-) create mode 100644 src/kimi_cli/cli/usage.py create mode 100644 tests/core/test_usage_cli.py diff --git a/docs/en/faq.md b/docs/en/faq.md index 4cb33d388..f04bfdc97 100644 --- a/docs/en/faq.md +++ b/docs/en/faq.md @@ -19,7 +19,7 @@ Possible reasons for an invalid API key: ### Membership expired or quota exhausted -If you're using the Kimi Code platform, you can check your current quota and membership status with the `/usage` command. If the quota is exhausted or membership expired, you need to renew or upgrade at [Kimi Code](https://kimi.com/coding). +If you're using the Kimi Code platform, you can check your current quota and membership status with the `/usage` command. For scripts or CI, run `kimi usage --json` to get JSON output. If the quota is exhausted or membership expired, you need to renew or upgrade at [Kimi Code](https://kimi.com/coding). ## Interaction issues diff --git a/docs/en/reference/kimi-command.md b/docs/en/reference/kimi-command.md index cf5250311..99aa71a6e 100644 --- a/docs/en/reference/kimi-command.md +++ b/docs/en/reference/kimi-command.md @@ -161,6 +161,7 @@ When not specified, Kimi Code CLI automatically discovers user-level and project | [`kimi login`](#kimi-login) | Log in to your Kimi account | | [`kimi logout`](#kimi-logout) | Log out from your Kimi account | | [`kimi info`](./kimi-info.md) | Display version and protocol information | +| [`kimi usage`](#kimi-usage) | Show Kimi Code quota usage | | [`kimi acp`](./kimi-acp.md) | Start multi-session ACP server | | [`kimi mcp`](./kimi-mcp.md) | Manage MCP server configuration | | [`kimi plugin`](../customization/plugins.md) | Manage plugins (Beta) | @@ -203,6 +204,18 @@ kimi export [] [-o ] [--yes] Added in version 1.20. ::: +### `kimi usage` + +Check Kimi Code platform quota usage non-interactively. By default, it prints the same terminal panel as `/usage` in shell mode; add `--json` to emit the raw JSON response for scripts, CI jobs, or monitoring widgets. + +```sh +kimi usage [--json] +``` + +| Option | Description | +|--------|-------------| +| `--json` | Output the raw usage API response as JSON | + ### `kimi vis` ::: warning Note diff --git a/docs/zh/faq.md b/docs/zh/faq.md index a82d0b9b0..4b89ae384 100644 --- a/docs/zh/faq.md +++ b/docs/zh/faq.md @@ -19,7 +19,7 @@ API 密钥无效可能的原因: ### 会员过期或配额用尽 -如果你使用 Kimi Code 平台,可以通过 `/usage` 命令查看当前的配额和会员状态。如果配额用尽或会员过期,需要在 [Kimi Code](https://kimi.com/coding) 续费或升级。 +如果你使用 Kimi Code 平台,可以通过 `/usage` 命令查看当前的配额和会员状态。需要在脚本或 CI 中读取时,可以运行 `kimi usage --json` 获取 JSON 输出。如果配额用尽或会员过期,需要在 [Kimi Code](https://kimi.com/coding) 续费或升级。 ## 交互问题 diff --git a/docs/zh/reference/kimi-command.md b/docs/zh/reference/kimi-command.md index 7de8bdd86..bb87ce7ee 100644 --- a/docs/zh/reference/kimi-command.md +++ b/docs/zh/reference/kimi-command.md @@ -161,6 +161,7 @@ Thinking 模式需要模型支持。如果不指定,使用上次会话的设 | [`kimi login`](#kimi-login) | 登录 Kimi 账号 | | [`kimi logout`](#kimi-logout) | 登出 Kimi 账号 | | [`kimi info`](./kimi-info.md) | 显示版本和协议信息 | +| [`kimi usage`](#kimi-usage) | 查看 Kimi Code 配额使用情况 | | [`kimi acp`](./kimi-acp.md) | 启动多会话 ACP 服务器 | | [`kimi mcp`](./kimi-mcp.md) | 管理 MCP 服务器配置 | | [`kimi plugin`](../customization/plugins.md) | 管理插件(Beta) | @@ -203,6 +204,18 @@ kimi export [] [-o ] [--yes] 新增于 1.20 版本。 ::: +### `kimi usage` + +非交互式查看 Kimi Code 平台的配额使用情况。默认输出与 Shell 模式中的 `/usage` 相同的终端面板;添加 `--json` 可输出原始 JSON,方便脚本、CI 或监控组件读取。 + +```sh +kimi usage [--json] +``` + +| 选项 | 说明 | +|------|------| +| `--json` | 输出原始 usage API 响应的 JSON | + ### `kimi vis` ::: warning 注意 diff --git a/src/kimi_cli/cli/_lazy_group.py b/src/kimi_cli/cli/_lazy_group.py index eac315ece..bc514b55e 100644 --- a/src/kimi_cli/cli/_lazy_group.py +++ b/src/kimi_cli/cli/_lazy_group.py @@ -18,6 +18,7 @@ class LazySubcommandGroup(typer.core.TyperGroup): "export": ("kimi_cli.cli.export", "cli", "Export session data."), "mcp": ("kimi_cli.cli.mcp", "cli", "Manage MCP server configurations."), "plugin": ("kimi_cli.cli.plugin", "cli", "Manage plugins."), + "usage": ("kimi_cli.cli.usage", "cli", "Display Kimi Code quota usage."), "vis": ("kimi_cli.cli.vis", "cli", "Run Kimi Agent Tracing Visualizer."), "web": ("kimi_cli.cli.web", "cli", "Run Kimi Code CLI web interface."), } @@ -26,6 +27,7 @@ class LazySubcommandGroup(typer.core.TyperGroup): "export", "mcp", "plugin", + "usage", "vis", "web", ) diff --git a/src/kimi_cli/cli/usage.py b/src/kimi_cli/cli/usage.py new file mode 100644 index 000000000..237ce5ba7 --- /dev/null +++ b/src/kimi_cli/cli/usage.py @@ -0,0 +1,139 @@ +from __future__ import annotations + +import asyncio +import json +from pathlib import Path +from typing import Annotated, Any, NoReturn, cast + +import aiohttp +import typer + +from kimi_cli.config import Config, LLMModel, LLMProvider +from kimi_cli.exception import ConfigError + +cli = typer.Typer(help="Display Kimi Code quota usage.") + + +def _exit_error(message: str) -> NoReturn: + typer.echo(f"Error: {message}", err=True) + raise typer.Exit(code=1) + + +def _load_config_from_root_options(ctx: typer.Context) -> Config: + from kimi_cli.config import load_config, load_config_from_string + + root_params = ctx.find_root().params + config_string = cast(str | None, root_params.get("config_string")) + config_file_raw = root_params.get("config_file") + config_file = Path(config_file_raw) if config_file_raw is not None else None + + if config_string is not None and config_file is not None: + raise typer.BadParameter( + "Cannot combine --config, --config-file.", + param_hint="--config", + ) + + if config_string is not None: + config_string = config_string.strip() + if not config_string: + raise typer.BadParameter("Config cannot be empty", param_hint="--config") + try: + return load_config_from_string(config_string) + except ConfigError as exc: + raise typer.BadParameter(str(exc), param_hint="--config") from exc + + try: + return load_config(config_file) + except ConfigError as exc: + param_hint = "--config-file" if config_file is not None else None + raise typer.BadParameter(str(exc), param_hint=param_hint) from exc + + +def _select_usage_target(config: Config, *, model_name: str | None) -> tuple[LLMModel, LLMProvider]: + selected_model = model_name or config.default_model + if not selected_model: + _exit_error("No model configured. Run `kimi login` or pass --model.") + + model = config.models.get(selected_model) + if model is None: + _exit_error(f"Model '{selected_model}' not found in configuration.") + + provider = config.providers.get(model.provider) + if provider is None: + _exit_error(f"Provider '{model.provider}' not found in configuration.") + + return model, provider + + +def _format_usage_fetch_error(exc: aiohttp.ClientResponseError) -> str: + if exc.status == 401: + return "Authorization failed. Please check your API key." + if exc.status == 404: + return "Usage endpoint not available. Try Kimi for Coding." + return "Failed to fetch usage." + + +async def _fetch_usage_payload(ctx: typer.Context) -> dict[str, Any]: + from kimi_cli.auth.oauth import OAuthError, OAuthManager + from kimi_cli.llm import augment_provider_with_env_vars + from kimi_cli.ui.shell import usage as shell_usage + + config = _load_config_from_root_options(ctx) + model_name = cast(str | None, ctx.find_root().params.get("model_name")) + model, provider = _select_usage_target(config, model_name=model_name) + + augment_provider_with_env_vars(provider, model) + usage_url = shell_usage.get_usage_url(model) + if usage_url is None: + _exit_error("Usage is available on Kimi Code platform only.") + + oauth = OAuthManager(config) + if provider.oauth is not None: + try: + await oauth.ensure_fresh() + except OAuthError as exc: + if not provider.api_key.get_secret_value(): + _exit_error(f"Failed to refresh OAuth token: {exc}") + typer.echo(f"Warning: failed to refresh OAuth token: {exc}", err=True) + + api_key = oauth.resolve_api_key(provider.api_key, provider.oauth) + try: + return dict(await shell_usage.fetch_usage(usage_url, api_key)) + except aiohttp.ClientResponseError as exc: + _exit_error(_format_usage_fetch_error(exc)) + except TimeoutError: + _exit_error("Failed to fetch usage: request timed out.") + except aiohttp.ClientError as exc: + _exit_error(f"Failed to fetch usage: {exc}") + + +def _emit_usage(payload: dict[str, Any], *, json_output: bool) -> None: + from kimi_cli.ui.shell import usage as shell_usage + from kimi_cli.ui.shell.console import console + + if json_output: + typer.echo(json.dumps(payload, ensure_ascii=False)) + return + + summary, limits = shell_usage.parse_usage_payload(payload) + if summary is None and not limits: + console.print("[yellow]No usage data available.[/yellow]") + return + + console.print(shell_usage.build_usage_panel(summary, limits)) + + +@cli.callback(invoke_without_command=True) +def usage( + ctx: typer.Context, + json_output: Annotated[ + bool, + typer.Option( + "--json", + help="Output the raw usage API response as JSON.", + ), + ] = False, +): + """Display Kimi Code quota usage.""" + payload = asyncio.run(_fetch_usage_payload(ctx)) + _emit_usage(payload, json_output=json_output) diff --git a/src/kimi_cli/ui/shell/usage.py b/src/kimi_cli/ui/shell/usage.py index 0c46a31dc..a7ce67f11 100644 --- a/src/kimi_cli/ui/shell/usage.py +++ b/src/kimi_cli/ui/shell/usage.py @@ -47,7 +47,7 @@ async def usage(app: Shell, args: str): console.print("[red]LLM provider configuration not found.[/red]") return - usage_url = _usage_url(app.soul.runtime.llm.model_config) + usage_url = get_usage_url(app.soul.runtime.llm.model_config) if usage_url is None: console.print("[yellow]Usage is available on Kimi Code platform only.[/yellow]") return @@ -55,7 +55,7 @@ async def usage(app: Shell, args: str): with console.status("[cyan]Fetching usage...[/cyan]"): api_key = app.soul.runtime.oauth.resolve_api_key(provider.api_key, provider.oauth) try: - payload = await _fetch_usage(usage_url, api_key) + payload = await fetch_usage(usage_url, api_key) except aiohttp.ClientResponseError as e: message = "Failed to fetch usage." if e.status == 401: @@ -71,15 +71,15 @@ async def usage(app: Shell, args: str): console.print(f"[red]Failed to fetch usage: {e}[/red]") return - summary, limits = _parse_usage_payload(payload) + summary, limits = parse_usage_payload(payload) if summary is None and not limits: console.print("[yellow]No usage data available.[/yellow]") return - console.print(_build_usage_panel(summary, limits)) + console.print(build_usage_panel(summary, limits)) -def _usage_url(model: LLMModel | None) -> str | None: +def get_usage_url(model: LLMModel | None) -> str | None: if model is None: return None platform_id = parse_managed_provider_key(model.provider) @@ -92,7 +92,7 @@ def _usage_url(model: LLMModel | None) -> str | None: return f"{base_url}/usages" -async def _fetch_usage(url: str, api_key: str) -> Mapping[str, Any]: +async def fetch_usage(url: str, api_key: str) -> Mapping[str, Any]: async with ( new_client_session() as session, session.get( @@ -104,7 +104,7 @@ async def _fetch_usage(url: str, api_key: str) -> Mapping[str, Any]: return await resp.json() -def _parse_usage_payload( +def parse_usage_payload( payload: Mapping[str, Any], ) -> tuple[UsageRow | None, list[UsageRow]]: summary: UsageRow | None = None @@ -228,7 +228,7 @@ def _to_int(value: Any) -> int | None: return None -def _build_usage_panel(summary: UsageRow | None, limits: list[UsageRow]) -> Panel: +def build_usage_panel(summary: UsageRow | None, limits: list[UsageRow]) -> Panel: rows = ([summary] if summary else []) + limits if not rows: return Panel( diff --git a/tests/core/test_startup_imports.py b/tests/core/test_startup_imports.py index 538d66524..8301f1ab9 100644 --- a/tests/core/test_startup_imports.py +++ b/tests/core/test_startup_imports.py @@ -76,6 +76,7 @@ def test_root_help_lists_lazy_subcommands_without_importing_them() -> None: "kimi_cli.cli.info", "kimi_cli.cli.export", "kimi_cli.cli.mcp", + "kimi_cli.cli.usage", "kimi_cli.cli.vis", "kimi_cli.cli.web", ] @@ -88,7 +89,7 @@ def test_root_help_lists_lazy_subcommands_without_importing_them() -> None: result = CliRunner().invoke(cli, ["--help"]) assert result.exit_code == 0, result.output -for name in ("info", "export", "mcp", "vis", "web"): +for name in ("info", "export", "mcp", "usage", "vis", "web"): assert name in result.output assert all(name not in sys.modules for name in lazy_modules) print("ok") @@ -107,6 +108,7 @@ def test_info_subcommand_loads_on_demand() -> None: "kimi_cli.cli.info", "kimi_cli.cli.export", "kimi_cli.cli.mcp", + "kimi_cli.cli.usage", "kimi_cli.cli.vis", "kimi_cli.cli.web", ] @@ -121,6 +123,7 @@ def test_info_subcommand_loads_on_demand() -> None: assert "kimi_cli.cli.info" in sys.modules assert "kimi_cli.cli.export" not in sys.modules assert "kimi_cli.cli.mcp" not in sys.modules +assert "kimi_cli.cli.usage" not in sys.modules assert "kimi_cli.cli.vis" not in sys.modules assert "kimi_cli.cli.web" not in sys.modules print("ok") diff --git a/tests/core/test_usage_cli.py b/tests/core/test_usage_cli.py new file mode 100644 index 000000000..ac7dd7acb --- /dev/null +++ b/tests/core/test_usage_cli.py @@ -0,0 +1,74 @@ +from __future__ import annotations + +import json +from collections.abc import Mapping +from pathlib import Path +from typing import Any + +from typer.testing import CliRunner + +from kimi_cli.cli import cli +from kimi_cli.ui.shell import usage as shell_usage + + +def _write_kimi_code_config(path: Path) -> None: + path.write_text( + """ +default_model = "kimi-code/kimi-k2" + +[providers."managed:kimi-code"] +type = "kimi" +base_url = "https://api.kimi.com/coding/v1" +api_key = "test-key" + +[models."kimi-code/kimi-k2"] +provider = "managed:kimi-code" +model = "kimi-k2" +max_context_size = 262144 +""".lstrip(), + encoding="utf-8", + ) + + +def test_usage_json_fetches_quota_for_configured_kimi_code_model( + tmp_path: Path, monkeypatch +) -> None: + config_file = tmp_path / "config.toml" + _write_kimi_code_config(config_file) + payload = {"usage": {"limit": 100, "remaining": 75}} + observed: dict[str, str] = {} + + async def fake_fetch_usage(url: str, api_key: str) -> Mapping[str, Any]: + observed["url"] = url + observed["api_key"] = api_key + return payload + + monkeypatch.setattr(shell_usage, "fetch_usage", fake_fetch_usage) + + result = CliRunner().invoke(cli, ["--config-file", str(config_file), "usage", "--json"]) + + assert result.exit_code == 0, result.output + assert json.loads(result.output) == payload + assert observed == { + "url": "https://api.kimi.com/coding/v1/usages", + "api_key": "test-key", + } + + +def test_usage_renders_interactive_usage_panel_non_interactively( + tmp_path: Path, monkeypatch +) -> None: + config_file = tmp_path / "config.toml" + _write_kimi_code_config(config_file) + + async def fake_fetch_usage(url: str, api_key: str) -> Mapping[str, Any]: + return {"usage": {"limit": 100, "remaining": 75}} + + monkeypatch.setattr(shell_usage, "fetch_usage", fake_fetch_usage) + + result = CliRunner().invoke(cli, ["--config-file", str(config_file), "usage"]) + + assert result.exit_code == 0, result.output + assert "API Usage" in result.output + assert "Weekly limit" in result.output + assert "75% left" in result.output