feat(config): persist Stytch session + account id per profile

README.md

@@ -94,6 +94,10 @@ and `aai whoami` / `aai doctor` confirm which source is active without printing
 | `aai llm <prompt>` | Prompt AssemblyAI's LLM Gateway (over a past transcript with `--transcript-id`, or a live streamed transcript with `--follow`). |
 | `aai claude install` | Wire Claude Code up to AssemblyAI's docs + skill. |
 | `aai samples create <name>` | Scaffold a runnable starter script (reads your key from `ASSEMBLYAI_API_KEY`). |
+| `aai keys list` / `create` / `rename` | Manage your API keys (browser login). |
+| `aai balance` / `usage` / `limits` | Account billing, usage, and rate limits (browser login). |
+| `aai sessions list` / `get <id>` | Browse past streaming (real-time) sessions (browser login). |
+| `aai audit` | View your account's audit log (browser login). |
 
 Add `--json` to any command for machine-readable output (it's also the default when
 output is piped or run by an agent). Errors always go to **stderr**, so stdout stays
@@ -108,6 +112,31 @@ across every command.
 > aai transcribe "https://www.youtube.com/watch?v=VIDEO_ID"
 > ```
 
+## Account self-service
+
+These commands use your browser login session (run `aai login` without
+`--api-key`), not your API key:
+
+```sh
+aai keys list                       # list API keys (masked) across projects
+aai keys create --name ci-pipeline  # mint a new key (printed once)
+aai keys rename 123 "prod"          # relabel a key
+
+aai balance                         # remaining account balance
+aai usage --start 2026-05-01 --end 2026-06-01
+aai limits                          # rate limits per service
+
+aai sessions list --status completed
+aai sessions get <session-id>       # one streaming session's details
+
+aai audit --limit 20                # recent account audit-log entries
+aai audit --action token.create     # filter by action
+```
+
+If a command reports it needs a browser login, your session has expired — run
+`aai login` again. (AMS sessions are short-lived and cannot be refreshed
+silently.)
+
 ## Transcribe options
 
 `aai transcribe` exposes the full `TranscriptionConfig` surface as curated flags,

aai_cli/auth/ams.py

@@ -20,11 +20,15 @@ def _detail(resp: httpx.Response) -> str:
     return resp.text or f"HTTP {resp.status_code}"
 
 
-def _json_or_raise(resp: httpx.Response) -> Any:
+def _raise_for_error(resp: httpx.Response) -> None:
     if resp.status_code in (401, 403):
         raise NotAuthenticated(f"AMS rejected the login ({resp.status_code}): {_detail(resp)}")
     if resp.status_code >= 400:
         raise APIError(f"AMS request failed ({resp.status_code}): {_detail(resp)}")
+
+
+def _json_or_raise(resp: httpx.Response) -> Any:
+    _raise_for_error(resp)
     return resp.json()
 
 
@@ -74,3 +78,65 @@ def create_token(
             json={"project_id": project_id, "token_name": token_name},
         )
     return cast(dict[str, Any], _json_or_raise(resp))
+
+
+def get_balance(session_jwt: str) -> dict[str, Any]:
+    """GET /v2/billing/balance -> {account_id, balance_in_cents, ...}."""
+    with _client(session_jwt) as client:
+        resp = client.get("/v2/billing/balance")
+    return cast(dict[str, Any], _json_or_raise(resp))
+
+
+def get_usage(
+    session_jwt: str,
+    starting_on: str,
+    ending_before: str,
+    window_size: str | None = None,
+) -> dict[str, Any]:
+    """POST /v2/billing/usage (ISO dates) -> {usage_items: [...]}."""
+    body: dict[str, Any] = {"starting_on": starting_on, "ending_before": ending_before}
+    if window_size:
+        body["window_size"] = window_size
+    with _client(session_jwt) as client:
+        resp = client.post("/v2/billing/usage", json=body)
+    return cast(dict[str, Any], _json_or_raise(resp))
+
+
+def get_rate_limits(account_id: int, session_jwt: str) -> dict[str, Any]:
+    """GET /v1/users/accounts/{id}/rate-limits -> {rate_limits: [...]}."""
+    with _client(session_jwt) as client:
+        resp = client.get(f"/v1/users/accounts/{account_id}/rate-limits")
+    return cast(dict[str, Any], _json_or_raise(resp))
+
+
+def rename_token(account_id: int, token_id: int, token_name: str, session_jwt: str) -> None:
+    """PUT /v1/users/accounts/{id}/tokens/{token_id}; 200 returns an empty body."""
+    with _client(session_jwt) as client:
+        resp = client.put(
+            f"/v1/users/accounts/{account_id}/tokens/{token_id}",
+            json={"token_name": token_name},
+        )
+    _raise_for_error(resp)
+
+
+def list_streaming(session_jwt: str, **filters: Any) -> dict[str, Any]:
+    """GET /v1/users/streaming -> {page_details, data: [StreamingSessionSchema]}."""
+    params = {k: v for k, v in filters.items() if v is not None}
+    with _client(session_jwt) as client:
+        resp = client.get("/v1/users/streaming", params=params)
+    return cast(dict[str, Any], _json_or_raise(resp))
+
+
+def get_streaming(session_id: str, session_jwt: str) -> dict[str, Any]:
+    """GET /v1/users/streaming/{session_id} -> StreamingSessionSchema."""
+    with _client(session_jwt) as client:
+        resp = client.get(f"/v1/users/streaming/{session_id}")
+    return cast(dict[str, Any], _json_or_raise(resp))
+
+
+def list_audit_logs(session_jwt: str, **filters: Any) -> dict[str, Any]:
+    """GET /v2/user/audit-logs -> {page_details, data: [AuditLogResponse]}."""
+    params = {k: v for k, v in filters.items() if v is not None}
+    with _client(session_jwt) as client:
+        resp = client.get("/v2/user/audit-logs", params=params)
+    return cast(dict[str, Any], _json_or_raise(resp))

aai_cli/auth/flow.py

@@ -2,13 +2,25 @@
 
 import webbrowser
 from collections.abc import Mapping
+from dataclasses import dataclass
 from typing import Any
 
 from aai_cli import output
 from aai_cli.auth import ams, discovery, endpoints, loopback
 from aai_cli.errors import APIError
 
 
+@dataclass
+class LoginResult:
+    """Everything a successful browser login yields: a usable API key plus the
+    Stytch session the AMS self-service commands authenticate with."""
+
+    api_key: str
+    session_jwt: str
+    session_token: str
+    account_id: int
+
+
 def _require(mapping: Mapping[str, Any], key: str, what: str) -> Any:
     """Pull a required field out of an AMS response, or raise a clean APIError.
 
@@ -73,8 +85,8 @@ def find_or_create_cli_key(account_id: int, session_jwt: str) -> str:
     return str(_require(created, "api_key", "an API key"))
 
 
-def run_login_flow() -> str:
-    """Drive the full browser + AMS login and return a usable AssemblyAI API key."""
+def run_login_flow() -> LoginResult:
+    """Drive the full browser + AMS login and return a LoginResult."""
     _open_browser(discovery.build_start_url())
     result = _capture()
 
@@ -109,7 +121,14 @@ def run_login_flow() -> str:
     intermediate_session_token = _require(disc, "intermediate_session_token", "a session token")
     signed_in = ams.exchange(intermediate_session_token, organization_id)
     session_jwt = _require(signed_in, "session_jwt", "a session token")
+    session_token = _require(signed_in, "session_token", "a session token")
     # `exchange` already returns the signed-in account, so read the id from it
     # rather than making a second GET /v1/auth round-trip.
-    account_id = _require(_require(signed_in, "account", "an account"), "id", "an account id")
-    return find_or_create_cli_key(int(account_id), session_jwt)
+    account_id = int(_require(_require(signed_in, "account", "an account"), "id", "an account id"))
+    api_key = find_or_create_cli_key(account_id, session_jwt)
+    return LoginResult(
+        api_key=api_key,
+        session_jwt=session_jwt,
+        session_token=session_token,
+        account_id=account_id,
+    )

aai_cli/commands/account.py

@@ -0,0 +1,107 @@
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+
+import typer
+from rich.markup import escape
+from rich.table import Table
+
+from aai_cli import output
+from aai_cli.auth import ams
+from aai_cli.context import AppState, resolve_session, run_command
+from aai_cli.help_text import examples_epilog
+
+app = typer.Typer(help="Account billing, usage, and limits.")
+
+
+@app.command(
+    epilog=examples_epilog(
+        [
+            ("Show your remaining balance", "aai balance"),
+        ]
+    )
+)
+def balance(
+    ctx: typer.Context,
+    json_out: bool = typer.Option(False, "--json", help="Output raw JSON."),
+) -> None:
+    """Show your remaining account balance."""
+
+    def body(state: AppState, json_mode: bool) -> None:
+        _, jwt = resolve_session(state)
+        data = ams.get_balance(jwt)
+        cents = data.get("balance_in_cents", 0) or 0
+        output.emit(
+            data,
+            lambda _d: f"Balance: [aai.success]${cents / 100:,.2f}[/aai.success]",
+            json_mode=json_mode,
+        )
+
+    run_command(ctx, body, json=json_out)
+
+
+@app.command(
+    epilog=examples_epilog(
+        [
+            ("Usage over the last 30 days", "aai usage"),
+            ("A specific date range", "aai usage --start 2026-05-01 --end 2026-06-01"),
+        ]
+    )
+)
+def usage(
+    ctx: typer.Context,
+    start: str = typer.Option(None, "--start", help="Start date (YYYY-MM-DD). Default: 30d ago."),
+    end: str = typer.Option(None, "--end", help="End date (YYYY-MM-DD). Default: today."),
+    window: str = typer.Option(None, "--window", help="Window size, e.g. 'day' or 'month'."),
+    json_out: bool = typer.Option(False, "--json", help="Output raw JSON."),
+) -> None:
+    """Show usage over a date range (defaults to the last 30 days)."""
+
+    def body(state: AppState, json_mode: bool) -> None:
+        _, jwt = resolve_session(state)
+        today = datetime.now(UTC).date()
+        end_date = end or today.isoformat()
+        start_date = start or (today - timedelta(days=30)).isoformat()
+        data = ams.get_usage(jwt, start_date, end_date, window)
+
+        def render(d: dict) -> Table:
+            table = Table("window start", "window end", "total", header_style="aai.heading")
+            for item in d.get("usage_items", []):
+                table.add_row(
+                    escape(str(item["start_timestamp"])),
+                    escape(str(item["end_timestamp"])),
+                    f"{item['total']:,}",
+                )
+            return table
+
+        output.emit(data, render, json_mode=json_mode)
+
+    run_command(ctx, body, json=json_out)
+
+
+@app.command(
+    epilog=examples_epilog(
+        [
+            ("Show rate limits per service", "aai limits"),
+        ]
+    )
+)
+def limits(
+    ctx: typer.Context,
+    json_out: bool = typer.Option(False, "--json", help="Output raw JSON."),
+) -> None:
+    """Show your account's rate limits per service."""
+
+    def body(state: AppState, json_mode: bool) -> None:
+        account_id, jwt = resolve_session(state)
+        data = ams.get_rate_limits(account_id, jwt)
+
+        def render(d: dict) -> Table:
+            table = Table("service", "limit", header_style="aai.heading")
+            for limit in d.get("rate_limits", []):
+                table.add_row(escape(str(limit["service"])), f"{limit['magnitude']:,}")
+            return table
+
+        output.emit(data, render, json_mode=json_mode)
+
+    run_command(ctx, body, json=json_out)

aai_cli/commands/audit.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+import typer
+from rich.markup import escape
+from rich.table import Table
+
+from aai_cli import output
+from aai_cli.auth import ams
+from aai_cli.context import AppState, resolve_session, run_command
+from aai_cli.help_text import examples_epilog
+
+app = typer.Typer(help="View your account's audit log.")
+
+
+@app.command(
+    epilog=examples_epilog(
+        [
+            ("Recent audit-log entries", "aai audit --limit 20"),
+            ("Filter by action", "aai audit --action token.create"),
+        ]
+    )
+)
+def audit(
+    ctx: typer.Context,
+    limit: int = typer.Option(20, "--limit", help="How many entries to show."),
+    action: str = typer.Option(None, "--action", help="Filter by action_taken."),
+    resource: str = typer.Option(None, "--resource", help="Filter by resource_type."),
+    json_out: bool = typer.Option(False, "--json", help="Output raw JSON."),
+) -> None:
+    """List recent audit-log entries for your account."""
+
+    def body(state: AppState, json_mode: bool) -> None:
+        _, jwt = resolve_session(state)
+        payload = ams.list_audit_logs(jwt, limit=limit, action_taken=action, resource_type=resource)
+        rows = payload.get("data", [])
+
+        def render(data: list[dict]) -> Table:
+            table = Table("time", "actor", "action", "resource", header_style="aai.heading")
+            for entry in data:
+                actor = str(entry["actor_type"])
+                actor_id = entry.get("actor_id")
+                if actor_id is not None:
+                    actor = f"{actor}:{actor_id}"
+                resource_label = entry.get("resource_type") or ""
+                resource_id = entry.get("resource_id")
+                if resource_label and resource_id:
+                    resource_label = f"{resource_label}:{resource_id}"
+                table.add_row(
+                    escape(str(entry["log_time"])),
+                    escape(actor),
+                    escape(str(entry["action_taken"])),
+                    escape(str(resource_label)),
+                )
+            return table
+
+        output.emit(rows, render, json_mode=json_mode)
+
+    run_command(ctx, body, json=json_out)