Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion docs/en/faq.md
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand Down
13 changes: 13 additions & 0 deletions docs/en/reference/kimi-command.md
Original file line number Diff line number Diff line change
Expand Up @@ -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) |
Expand Down Expand Up @@ -203,6 +204,18 @@ kimi export [<session_id>] [-o <output_path>] [--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
Expand Down
2 changes: 1 addition & 1 deletion docs/zh/faq.md
Original file line number Diff line number Diff line change
Expand Up @@ -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) 续费或升级。

## 交互问题

Expand Down
13 changes: 13 additions & 0 deletions docs/zh/reference/kimi-command.md
Original file line number Diff line number Diff line change
Expand Up @@ -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) |
Expand Down Expand Up @@ -203,6 +204,18 @@ kimi export [<session_id>] [-o <output_path>] [--yes]
新增于 1.20 版本。
:::

### `kimi usage`

非交互式查看 Kimi Code 平台的配额使用情况。默认输出与 Shell 模式中的 `/usage` 相同的终端面板;添加 `--json` 可输出原始 JSON,方便脚本、CI 或监控组件读取。

```sh
kimi usage [--json]
```

| 选项 | 说明 |
|------|------|
| `--json` | 输出原始 usage API 响应的 JSON |

### `kimi vis`

::: warning 注意
Expand Down
2 changes: 2 additions & 0 deletions src/kimi_cli/cli/_lazy_group.py
Original file line number Diff line number Diff line change
Expand Up @@ -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."),
}
Expand All @@ -26,6 +27,7 @@ class LazySubcommandGroup(typer.core.TyperGroup):
"export",
"mcp",
"plugin",
"usage",
"vis",
"web",
)
Expand Down
139 changes: 139 additions & 0 deletions src/kimi_cli/cli/usage.py
Original file line number Diff line number Diff line change
@@ -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)
16 changes: 8 additions & 8 deletions src/kimi_cli/ui/shell/usage.py
Original file line number Diff line number Diff line change
Expand Up @@ -47,15 +47,15 @@ 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

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:
Expand All @@ -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)
Expand All @@ -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(
Expand All @@ -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
Expand Down Expand Up @@ -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(
Expand Down
5 changes: 4 additions & 1 deletion tests/core/test_startup_imports.py
Original file line number Diff line number Diff line change
Expand Up @@ -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",
]
Expand All @@ -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")
Expand All @@ -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",
]
Expand All @@ -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")
Expand Down
74 changes: 74 additions & 0 deletions tests/core/test_usage_cli.py
Original file line number Diff line number Diff line change
@@ -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
Loading