diff --git a/CLAUDE.md b/CLAUDE.md index 71a44c6..8c00593 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -79,6 +79,11 @@ src/keboola_agent_cli/ changelog.py # LAYER 1: Changelog display context.py # LAYER 1: Agent usage instructions doctor.py # LAYER 1: Health check command + hints/ + __init__.py # HintRegistry + render_hint() public API + models.py # HintMode, ClientCall, ServiceCall, HintStep, CommandHint + renderer.py # ClientRenderer + ServiceRenderer (Python code generation) + definitions/ # One file per command group (config.py, storage.py, job.py, ...) services/ base.py # LAYER 2: BaseService - shared parallel execution infrastructure project_service.py # LAYER 2: Business logic for projects @@ -219,7 +224,7 @@ Note: `SKILL.md` instructs Claude to run `kbagent context` as its first step, wh ## All CLI Commands ``` -# Global options: --json, --verbose, --no-color, --config-dir +# Global options: --json, --verbose, --no-color, --config-dir, --hint client|service kbagent project add --project NAME --url URL --token TOKEN kbagent project list diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index a77dd33..87ac3bc 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -188,6 +188,8 @@ When adding a new command (e.g., `kbagent storage create-foo`), you must update - [ ] **Client method** in `client.py` (or `manage_client.py`) -- HTTP layer - [ ] **Service method** in `services/` -- business logic, validation, orchestration - [ ] **Command function** in `commands/` -- Typer options, formatter, error handling +- [ ] **Hint definition** in `hints/definitions/` -- register a `CommandHint` for `--hint` code generation (see existing files for pattern) +- [ ] **Hint short-circuit** in the command function -- add `if should_hint(ctx): emit_hint(...)` before service call - [ ] **Permission registration** in `permissions.py` (`OPERATION_REGISTRY` dict) - [ ] **Service wiring** in `cli.py` if adding a new service class diff --git a/docs/hint-mode.md b/docs/hint-mode.md new file mode 100644 index 0000000..2aaff2f --- /dev/null +++ b/docs/hint-mode.md @@ -0,0 +1,114 @@ +# --hint Mode: Use kbagent as a Python SDK + +The `--hint` flag generates equivalent Python code for any CLI command, without +executing it. This lets you use `kbagent` not just as a CLI tool, but as a +programming reference and SDK. + +## Quick start + +```bash +# Show how to call the API directly (client layer) +kbagent --hint client config list --project myproj + +# Show how to use the service layer (with CLI config) +kbagent --hint service config list --project myproj +``` + +## Two modes + +### `--hint client` (direct API calls) + +Generates code using `KeboolaClient` with explicit URL and token. +Best for: standalone scripts, CI/CD, when you don't want CLI config dependency. + +```python +import os +from keboola_agent_cli.client import KeboolaClient + +client = KeboolaClient( + base_url="https://connection.eu-central-1.keboola.com", + token=os.environ["KBC_STORAGE_TOKEN"], +) +try: + components = client.list_components() +finally: + client.close() +``` + +### `--hint service` (service layer with CLI config) + +Generates code using the service layer, which reads project configuration +from the same `config.json` that `kbagent` uses. Best for: scripts that +work with multiple projects, need branch resolution, or want error accumulation. + +```python +from pathlib import Path +from keboola_agent_cli.config_store import ConfigStore +from keboola_agent_cli.services.config_service import ConfigService + +store = ConfigStore(config_dir=Path("/path/to/.kbagent")) +service = ConfigService(config_store=store) +result = service.list_configs(aliases=["myproj"]) +``` + +The `config_dir` path is always explicit -- no hidden CWD resolution. +The generated code uses the actual path from your current CLI configuration. + +## What it does NOT do + +- Does NOT execute any API calls +- Does NOT include real tokens (always uses `os.environ[...]`) +- Does NOT trigger auto-update checks +- Does NOT require a valid token (only needs project config for URL resolution) + +## Supported commands + +All API-backed commands support `--hint` (45 commands total), including: +- `config list/detail/search` +- `storage buckets/tables/files` and all CRUD operations +- `job list/detail/run` (including poll loop pattern for `--wait`) +- `branch list/create/delete` +- `workspace create/list/detail/delete/load/query` +- `sharing list/share/unshare/link/unlink` +- `component list/detail` +- `encrypt values` +- `lineage show` +- `org setup` + +Local-only commands (`project add/list/remove`, `branch use/reset`, etc.) do +not support `--hint` because they don't make API calls. + +## Examples + +### Multi-step command (job run with polling) + +```bash +kbagent --hint client job run --project myproj \ + --component-id keboola.ex-http --config-id 123 --wait +``` + +Generates code with a polling loop: + +```python +job = client.create_job(component_id="keboola.ex-http", config_id="123") +while not job.get("isFinished"): + time.sleep(5.0) + job = client.get_job_detail(job_id=str(job["id"])) +``` + +### Manage API command + +```bash +kbagent --hint client org setup --org-id 42 --url https://connection.keboola.com +``` + +Generates code with `ManageClient` and the correct token env var: + +```python +from keboola_agent_cli.manage_client import ManageClient + +manage_client = ManageClient( + base_url="https://connection.keboola.com", + token=os.environ["KBC_MANAGE_API_TOKEN"], +) +``` diff --git a/plugins/kbagent/skills/kbagent/SKILL.md b/plugins/kbagent/skills/kbagent/SKILL.md index c3637e9..ce9f345 100644 --- a/plugins/kbagent/skills/kbagent/SKILL.md +++ b/plugins/kbagent/skills/kbagent/SKILL.md @@ -41,6 +41,7 @@ If kbagent is not installed or you need the full standalone reference, run `kbag 3. **Multi-project by default**: read commands query ALL connected projects in parallel -- no need to loop 4. **Write commands need `--project`**: specify the target project alias 5. **Tokens are always masked** in output -- this is expected, not an error +6. **Use `--hint` for Python code generation**: `kbagent --hint client ` generates Python code using `KeboolaClient` (direct API), `kbagent --hint service ` generates code using the service layer with CLI config. See [programming-with-cli.md](references/programming-with-cli.md) for details. ## Choosing the right approach diff --git a/plugins/kbagent/skills/kbagent/references/commands-reference.md b/plugins/kbagent/skills/kbagent/references/commands-reference.md index 0dcbf67..f0ee02f 100644 --- a/plugins/kbagent/skills/kbagent/references/commands-reference.md +++ b/plugins/kbagent/skills/kbagent/references/commands-reference.md @@ -101,6 +101,7 @@ All commands support `--json` for structured output. Multi-project flags (`--pro | `--verbose / -v` | Verbose output | | `--no-color` | Disable colors | | `--config-dir` | Override config directory | +| `--hint client\|service` | Generate Python code instead of executing (see [programming-with-cli.md](programming-with-cli.md)) | ## Environment Variables | Variable | Purpose | diff --git a/plugins/kbagent/skills/kbagent/references/gotchas.md b/plugins/kbagent/skills/kbagent/references/gotchas.md index 00a8cc9..2dfb7de 100644 --- a/plugins/kbagent/skills/kbagent/references/gotchas.md +++ b/plugins/kbagent/skills/kbagent/references/gotchas.md @@ -258,6 +258,24 @@ scope to that branch: This means you can have production and dev branch configs side by side on disk without them overwriting each other. +## --hint mode: generate Python code + +Use `--hint` to generate equivalent Python code instead of executing a command: + +```bash +kbagent --hint client config list --project myproj # direct API calls +kbagent --hint service config list --project myproj # service layer with CLI config +``` + +Two modes: +- **`--hint client`**: generates code using `KeboolaClient` with explicit URL + token +- **`--hint service`**: generates code using the service layer with `ConfigStore` + +Important: `--hint` requires a value (`client` or `service`). Writing just `--hint` +without a value will cause a parsing error. + +See [docs/hint-mode.md](../../../../../docs/hint-mode.md) for full documentation. + ## Common mistakes - **Forgetting `--json`**: without it, output is human-formatted Rich text, not parseable diff --git a/plugins/kbagent/skills/kbagent/references/programming-with-cli.md b/plugins/kbagent/skills/kbagent/references/programming-with-cli.md new file mode 100644 index 0000000..e7758ff --- /dev/null +++ b/plugins/kbagent/skills/kbagent/references/programming-with-cli.md @@ -0,0 +1,172 @@ +# Programming with kbagent CLI as a Python SDK + +kbagent is not just a command-line tool -- it is also a Python SDK. When you +install kbagent (`uv tool install keboola-agent-cli`), you get importable +Python modules that you can use directly in your scripts. + +Use `--hint` on any command to see how to do the same thing in Python: + +```bash +kbagent --hint client config list --project myproj +kbagent --hint service job run --project myproj --component-id keboola.ex-http --config-id 123 +``` + +## Two layers: client vs service + +### Client layer (`--hint client`) + +Direct API calls with explicit URL and token. No dependency on CLI config. + +```python +import os +from keboola_agent_cli.client import KeboolaClient + +client = KeboolaClient( + base_url="https://connection.eu-central-1.keboola.com", + token=os.environ["KBC_STORAGE_TOKEN"], +) +try: + # List all components with their configurations + components = client.list_components() + + # Get a specific config + detail = client.get_config_detail("keboola.ex-db-snowflake", "12345") + + # List tables in a bucket + tables = client.list_tables(bucket_id="in.c-demo") + + # Run a job via Queue API + job = client.create_job( + component_id="keboola.ex-http", + config_id="456", + ) +finally: + client.close() +``` + +**When to use**: standalone scripts, CI/CD pipelines, Lambda functions, +or any context where you manage tokens and URLs yourself. + +**Available clients**: +- `KeboolaClient` -- Storage API + Queue API + Encryption API +- `ManageClient` -- Manage API (organization operations, token management) +- `AiServiceClient` -- AI Service API (component search, schema summaries) + +### Service layer (`--hint service`) + +Higher-level abstraction that uses CLI config for project resolution. + +```python +from pathlib import Path +from keboola_agent_cli.config_store import ConfigStore +from keboola_agent_cli.services.config_service import ConfigService + +# Explicit path to config directory (no CWD magic) +store = ConfigStore(config_dir=Path("/Users/you/.kbagent")) +service = ConfigService(config_store=store) + +# Use project aliases -- service resolves URL + token from config +result = service.list_configs(aliases=["myproj"]) +# Returns: {"configs": [...], "errors": [...]} + +# Query all projects at once (parallel execution) +result = service.list_configs() # aliases=None means all projects +``` + +**When to use**: scripts that work with multiple projects, need automatic +branch resolution, or want error accumulation (one project fails, others +continue). + +**Key difference from client layer**: you pass project **aliases** (like +"myproj") instead of URLs and tokens. The service resolves them from the +same config file that `kbagent` uses. + +## Comparison table + +| Feature | Client layer | Service layer | +|---------|-------------|---------------| +| Authentication | Explicit URL + token | Project alias from config | +| Multi-project | Manual loop | Built-in parallel execution | +| Branch resolution | Explicit branch_id | Auto-resolves active branch | +| Error handling | Exceptions | Error accumulation (partial success) | +| Dependencies | None (just URL + token) | Requires CLI config (`kbagent project add`) | +| Config path | N/A | Explicit `config_dir=Path(...)` | + +## Available services + +| Service class | Import from | Key methods | +|---------------|-------------|-------------| +| `ConfigService` | `services.config_service` | `list_configs`, `get_config_detail`, `search_configs` | +| `StorageService` | `services.storage_service` | `list_buckets`, `list_tables`, `upload_table`, `download_table` | +| `JobService` | `services.job_service` | `list_jobs`, `get_job_detail`, `run_job` | +| `BranchService` | `services.branch_service` | `list_branches`, `create_branch`, `delete_branch` | +| `WorkspaceService` | `services.workspace_service` | `create_workspace`, `execute_query`, `load_tables` | +| `SharingService` | `services.sharing_service` | `list_shared`, `share`, `link`, `unlink` | +| `LineageService` | `services.lineage_service` | `get_lineage` | +| `ComponentService` | `services.component_service` | `list_components`, `get_component_detail` | +| `EncryptService` | `services.encrypt_service` | `encrypt` | +| `OrgService` | `services.org_service` | `setup_organization` | + +## Common patterns + +### Poll a job until completion + +```python +import time +job = client.create_job(component_id="keboola.ex-http", config_id="123") +while not job.get("isFinished"): + time.sleep(5) + job = client.get_job_detail(str(job["id"])) +print(f"Job finished with status: {job['status']}") +``` + +Or with the service layer (handles polling internally): + +```python +result = job_service.run_job( + alias="myproj", + component_id="keboola.ex-http", + config_id="123", + wait=True, + timeout=300.0, +) +``` + +### Upload a CSV file to a table + +```python +result = storage_service.upload_table( + alias="myproj", + table_id="in.c-demo.my-table", + file_path="/tmp/data.csv", + incremental=True, +) +``` + +### Execute SQL in a workspace + +```python +result = workspace_service.execute_query( + alias="myproj", + workspace_id=12345, + sql="SELECT * FROM my_table LIMIT 10", +) +# result contains query results as dicts +``` + +### Search across all project configurations + +```python +result = config_service.search_configs( + query="snowflake", + ignore_case=True, +) +# result = {"matches": [...], "errors": [...], "stats": {...}} +``` + +## Security notes + +- Never hardcode tokens in scripts. Use `os.environ["KBC_STORAGE_TOKEN"]`. +- The client automatically masks tokens in error messages. +- Built-in retry logic handles 429/5xx errors with exponential backoff. +- ConfigStore files use 0600 permissions to protect stored tokens. diff --git a/src/keboola_agent_cli/cli.py b/src/keboola_agent_cli/cli.py index c57490f..384b9dd 100644 --- a/src/keboola_agent_cli/cli.py +++ b/src/keboola_agent_cli/cli.py @@ -112,11 +112,32 @@ def main( "--config-dir", help="Override config directory path.", ), + hint: str | None = typer.Option( + None, + "--hint", + help="Show equivalent Python code instead of executing. " + "Values: 'client' (direct API usage, default) or 'service' (uses CLI config).", + ), ) -> None: """Global options applied to all commands.""" from .auto_update import maybe_auto_update, show_post_update_changelog - maybe_auto_update() + # Skip auto-update in hint mode (code generation only) + if hint: + from .hints.models import HintMode + + valid_modes = [m.value for m in HintMode] + if hint not in valid_modes: + typer.echo( + f"Error: Invalid --hint value '{hint}'.\n" + f"Usage: kbagent --hint client (direct API calls)\n" + f" kbagent --hint service (uses CLI config)\n" + f"Valid values: {', '.join(valid_modes)}", + err=True, + ) + raise typer.Exit(code=2) from None + else: + maybe_auto_update() show_post_update_changelog() # If no subcommand given, launch REPL on TTY or show help otherwise @@ -175,9 +196,17 @@ def main( # Config may be invalid (e.g. corrupted JSON) -- skip permission check permission_engine = PermissionEngine(None) + # Resolve hint mode + hint_mode = None + if hint: + from .hints.models import HintMode + + hint_mode = HintMode(hint) + ctx.ensure_object(dict) ctx.obj["formatter"] = formatter ctx.obj["json_output"] = json_output + ctx.obj["hint_mode"] = hint_mode ctx.obj["permission_engine"] = permission_engine ctx.obj["verbose"] = verbose ctx.obj["no_color"] = effective_no_color @@ -223,6 +252,16 @@ def main( # Enforce permissions for top-level commands (sub-app commands use callbacks) _top_level_commands = {"init", "doctor", "version", "update", "changelog", "context", "repl"} _is_help = "--help" in sys.argv or "-h" in sys.argv + + # Hint mode on top-level commands — these are all local, no hints available + if hint_mode and ctx.invoked_subcommand in _top_level_commands: + typer.echo( + f"No --hint available for '{ctx.invoked_subcommand}'. " + f"This command operates locally and does not make API calls.", + err=True, + ) + raise typer.Exit(0) + if ctx.invoked_subcommand in _top_level_commands and not _is_help: try: permission_engine.check_or_raise(ctx.invoked_subcommand) diff --git a/src/keboola_agent_cli/commands/_helpers.py b/src/keboola_agent_cli/commands/_helpers.py index 593ed7f..700063a 100644 --- a/src/keboola_agent_cli/commands/_helpers.py +++ b/src/keboola_agent_cli/commands/_helpers.py @@ -5,6 +5,7 @@ - Exit code mapping for API errors - Warning emission for multi-project operations - Branch resolution for --branch flag +- Hint mode detection and code generation """ import os @@ -107,7 +108,7 @@ def check_cli_permission(ctx: typer.Context, group_name: str) -> None: Called from sub-app callbacks. Constructs operation name as '{group_name}.{subcommand}' and checks against the permission engine. - Always allows --help through so users can read docs for blocked commands. + Always allows --help and --hint through (no API calls made). Args: ctx: Typer context (must have permission_engine in obj). @@ -116,6 +117,24 @@ def check_cli_permission(ctx: typer.Context, group_name: str) -> None: if _is_help_request(ctx): return + # Hint mode: skip permission check + catch commands without hint definitions + if should_hint(ctx): + subcommand = ctx.invoked_subcommand + if subcommand: + cli_command = f"{group_name}.{subcommand}" + from ..hints import HintRegistry + from ..hints import definitions as _defs # noqa: F401 + + if HintRegistry.get(cli_command) is not None: + return # Hint exists — let the command function handle it + # No hint registered for this command + typer.echo( + f"No --hint available for '{group_name} {subcommand}'.", + err=True, + ) + raise typer.Exit(0) + return + engine = ctx.obj.get("permission_engine") if engine is None or not engine.active: return @@ -206,3 +225,64 @@ def resolve_branch( return alias, proj.active_branch_id return project, None + + +# ── Hint mode helpers ────────────────────────────────────────────── + + +def should_hint(ctx: typer.Context) -> bool: + """Check if --hint mode is active.""" + return ctx.obj.get("hint_mode") is not None + + +def emit_hint(ctx: typer.Context, cli_command: str, **params: Any) -> None: + """Render Python hint code, print to stdout, and exit. + + Resolves the project alias to a stack_url and config_dir, then + delegates to the hint renderer. + + Args: + ctx: Typer context (must have hint_mode and config_store). + cli_command: Dot-separated command key, e.g. 'config.list'. + **params: CLI parameters passed to the command. + """ + from ..hints import render_hint + + hint_mode = ctx.obj["hint_mode"] + config_store: ConfigStore = ctx.obj["config_store"] + + # Resolve project alias -> stack_url + stack_url = _resolve_hint_stack_url(config_store, params.get("project")) + + # Actual config_dir path (for service layer hints) + config_dir = config_store.config_path.parent + + # Branch ID from params + branch_id = params.get("branch") + + output = render_hint(cli_command, hint_mode, params, stack_url, config_dir, branch_id) + sys.stdout.write(output + "\n") + raise typer.Exit(0) + + +def _resolve_hint_stack_url( + config_store: ConfigStore, + project: str | list[str] | None, +) -> str | None: + """Resolve a project alias to its stack_url for hint rendering. + + Returns None if the project cannot be resolved (hint will use a placeholder). + """ + if project is None: + return None + + alias = project[0] if isinstance(project, list) else project + + try: + proj = config_store.get_project(alias) + if proj: + return proj.stack_url + except Exception: + pass + + return None diff --git a/src/keboola_agent_cli/commands/branch.py b/src/keboola_agent_cli/commands/branch.py index 56d3d33..cee3720 100644 --- a/src/keboola_agent_cli/commands/branch.py +++ b/src/keboola_agent_cli/commands/branch.py @@ -10,10 +10,12 @@ from ..output import format_branches_table from ._helpers import ( check_cli_permission, + emit_hint, emit_project_warnings, get_formatter, get_service, map_error_to_exit_code, + should_hint, ) branch_app = typer.Typer(help="Manage development branches") @@ -34,6 +36,9 @@ def branch_list( ), ) -> None: """List development branches from connected projects.""" + if should_hint(ctx): + emit_hint(ctx, "branch.list", project=project) + return formatter = get_formatter(ctx) service = get_service(ctx, "branch_service") @@ -74,6 +79,9 @@ def branch_create( The created branch becomes the active branch for the project, so subsequent tool calls will automatically use it. """ + if should_hint(ctx): + emit_hint(ctx, "branch.create", project=project, name=name, description=description) + return formatter = get_formatter(ctx) service = get_service(ctx, "branch_service") @@ -184,6 +192,9 @@ def branch_delete( If the deleted branch was the active branch, it is automatically reset to main/production. """ + if should_hint(ctx): + emit_hint(ctx, "branch.delete", project=project, branch=branch) + return formatter = get_formatter(ctx) service = get_service(ctx, "branch_service") diff --git a/src/keboola_agent_cli/commands/component.py b/src/keboola_agent_cli/commands/component.py index 4c6b48b..273d2dd 100644 --- a/src/keboola_agent_cli/commands/component.py +++ b/src/keboola_agent_cli/commands/component.py @@ -13,10 +13,12 @@ from ..errors import ConfigError, KeboolaApiError from ._helpers import ( check_cli_permission, + emit_hint, emit_project_warnings, get_formatter, get_service, map_error_to_exit_code, + should_hint, ) component_app = typer.Typer(help="Discover and inspect Keboola components") @@ -149,6 +151,9 @@ def component_list( ), ) -> None: """List available components from connected projects.""" + if should_hint(ctx): + emit_hint(ctx, "component.list", project=project, type=component_type, query=query) + return formatter = get_formatter(ctx) service = get_service(ctx, "component_service") @@ -193,6 +198,9 @@ def component_detail( ), ) -> None: """Show detailed information about a specific component.""" + if should_hint(ctx): + emit_hint(ctx, "component.detail", project=project, component_id=component_id) + return formatter = get_formatter(ctx) service = get_service(ctx, "component_service") diff --git a/src/keboola_agent_cli/commands/config.py b/src/keboola_agent_cli/commands/config.py index c20cbe2..0f58e27 100644 --- a/src/keboola_agent_cli/commands/config.py +++ b/src/keboola_agent_cli/commands/config.py @@ -18,11 +18,13 @@ from ..output import format_config_detail, format_configs_table, format_search_results from ._helpers import ( check_cli_permission, + emit_hint, emit_project_warnings, get_formatter, get_service, map_error_to_exit_code, resolve_branch, + should_hint, ) logger = logging.getLogger(__name__) @@ -93,6 +95,16 @@ def config_list( If a dev branch is active (via 'branch use'), configs from that branch are listed. Use --branch to override. """ + if should_hint(ctx): + emit_hint( + ctx, + "config.list", + project=project, + component_type=component_type, + component_id=component_id, + branch=branch, + ) + formatter = get_formatter(ctx) service = get_service(ctx, "config_service") config_store: ConfigStore = ctx.obj["config_store"] @@ -158,6 +170,16 @@ def config_detail( If a dev branch is active (via 'branch use'), the detail is fetched from that branch. Use --branch to override. """ + if should_hint(ctx): + emit_hint( + ctx, + "config.detail", + project=project, + component_id=component_id, + config_id=config_id, + branch=branch, + ) + formatter = get_formatter(ctx) service = get_service(ctx, "config_service") config_store: ConfigStore = ctx.obj["config_store"] @@ -231,6 +253,19 @@ def config_search( If a dev branch is active (via 'branch use'), configs from that branch are searched. Use --branch to override. """ + if should_hint(ctx): + emit_hint( + ctx, + "config.search", + query=query, + project=project, + component_type=component_type, + component_id=component_id, + ignore_case=ignore_case, + regex=use_regex, + branch=branch, + ) + formatter = get_formatter(ctx) service = get_service(ctx, "config_service") config_store: ConfigStore = ctx.obj["config_store"] diff --git a/src/keboola_agent_cli/commands/context.py b/src/keboola_agent_cli/commands/context.py index c39ec73..0f1040c 100644 --- a/src/keboola_agent_cli/commands/context.py +++ b/src/keboola_agent_cli/commands/context.py @@ -47,6 +47,7 @@ --verbose / -v Verbose output --no-color Disable colors (auto-disabled in non-TTY) --config-dir Override config directory path + --hint MODE Generate Python code instead of executing (MODE: client or service) ## All Commands @@ -393,6 +394,15 @@ kbagent --json tool call get_configs --project prod --branch 456 \\ --input '{{"configs": [{{"component_id": "keboola.snowflake-transformation", "configuration_id": "12345"}}]}}' +10. Python code generation with --hint: + kbagent --hint client config list --project prod # Direct API calls (KeboolaClient) + kbagent --hint service config list --project prod # Service layer (uses CLI config) + Two modes: + client -- generates code with explicit URL + token, no CLI config dependency + service -- generates code using ConfigStore with explicit config_dir path + Works on all API-backed commands (45 total). No API calls are made. + Use this when building Python scripts that automate Keboola operations. + ## Exit Codes 0 Success diff --git a/src/keboola_agent_cli/commands/encrypt.py b/src/keboola_agent_cli/commands/encrypt.py index 04929e7..5ad7c9f 100644 --- a/src/keboola_agent_cli/commands/encrypt.py +++ b/src/keboola_agent_cli/commands/encrypt.py @@ -13,9 +13,11 @@ from ..errors import ConfigError, KeboolaApiError from ._helpers import ( check_cli_permission, + emit_hint, get_formatter, get_service, map_error_to_exit_code, + should_hint, ) encrypt_app = typer.Typer( @@ -65,6 +67,11 @@ def encrypt_values( echo '{"#token": "abc"}' | kbagent encrypt values --project my-proj --component-id keboola.ex-db-snowflake --input - kbagent encrypt values --project my-proj --component-id keboola.ex-db-snowflake --input @secrets.json --output-file encrypted.json """ + if should_hint(ctx): + emit_hint( + ctx, "encrypt.values", project=project, component_id=component_id, input=input_data + ) + return formatter = get_formatter(ctx) service = get_service(ctx, "encrypt_service") diff --git a/src/keboola_agent_cli/commands/job.py b/src/keboola_agent_cli/commands/job.py index 74dbf45..20a4537 100644 --- a/src/keboola_agent_cli/commands/job.py +++ b/src/keboola_agent_cli/commands/job.py @@ -11,10 +11,12 @@ from ..output import format_job_detail, format_jobs_table from ._helpers import ( check_cli_permission, + emit_hint, emit_project_warnings, get_formatter, get_service, map_error_to_exit_code, + should_hint, ) job_app = typer.Typer(help="Browse job history and run jobs") @@ -55,6 +57,17 @@ def job_list( ), ) -> None: """List jobs from connected projects.""" + if should_hint(ctx): + emit_hint( + ctx, + "job.list", + project=project, + component_id=component_id, + config_id=config_id, + status=status, + limit=limit, + ) + return formatter = get_formatter(ctx) service = get_service(ctx, "job_service") @@ -108,6 +121,9 @@ def job_detail( job_id: str = typer.Option(..., "--job-id", help="Job ID"), ) -> None: """Show detailed information about a specific job.""" + if should_hint(ctx): + emit_hint(ctx, "job.detail", project=project, job_id=job_id) + return formatter = get_formatter(ctx) service = get_service(ctx, "job_service") @@ -167,6 +183,18 @@ def job_run( Creates a Queue API job and optionally waits for completion. Use --row-id to run specific configuration rows. """ + if should_hint(ctx): + emit_hint( + ctx, + "job.run", + project=project, + component_id=component_id, + config_id=config_id, + row_id=row_id, + wait=wait, + timeout=timeout, + ) + return formatter = get_formatter(ctx) service = get_service(ctx, "job_service") diff --git a/src/keboola_agent_cli/commands/lineage.py b/src/keboola_agent_cli/commands/lineage.py index 59f24bc..ce24c72 100644 --- a/src/keboola_agent_cli/commands/lineage.py +++ b/src/keboola_agent_cli/commands/lineage.py @@ -8,7 +8,14 @@ from ..errors import ConfigError from ..output import format_lineage_table -from ._helpers import check_cli_permission, emit_project_warnings, get_formatter, get_service +from ._helpers import ( + check_cli_permission, + emit_hint, + emit_project_warnings, + get_formatter, + get_service, + should_hint, +) lineage_app = typer.Typer(help="Analyze cross-project data lineage via bucket sharing") @@ -28,6 +35,9 @@ def lineage_show( ), ) -> None: """Show cross-project data lineage via bucket sharing.""" + if should_hint(ctx): + emit_hint(ctx, "lineage.show", project=project) + return formatter = get_formatter(ctx) service = get_service(ctx, "lineage_service") diff --git a/src/keboola_agent_cli/commands/org.py b/src/keboola_agent_cli/commands/org.py index d20c2f3..29769a5 100644 --- a/src/keboola_agent_cli/commands/org.py +++ b/src/keboola_agent_cli/commands/org.py @@ -12,10 +12,12 @@ from ..errors import KeboolaApiError from ._helpers import ( check_cli_permission, + emit_hint, get_formatter, get_service, map_error_to_exit_code, resolve_manage_token, + should_hint, ) org_app = typer.Typer(help="Organization management") @@ -203,6 +205,9 @@ def org_setup( The token is read from KBC_MANAGE_API_TOKEN env var or prompted interactively (never passed as a CLI argument for security). """ + if should_hint(ctx): + emit_hint(ctx, "org.setup", org_id=org_id, url=url, dry_run=dry_run) + return formatter = get_formatter(ctx) service = get_service(ctx, "org_service") diff --git a/src/keboola_agent_cli/commands/sharing.py b/src/keboola_agent_cli/commands/sharing.py index 7ce9b4a..690a0a5 100644 --- a/src/keboola_agent_cli/commands/sharing.py +++ b/src/keboola_agent_cli/commands/sharing.py @@ -11,10 +11,12 @@ from ..errors import ConfigError, KeboolaApiError from ._helpers import ( check_cli_permission, + emit_hint, emit_project_warnings, get_formatter, get_service, map_error_to_exit_code, + should_hint, ) sharing_app = typer.Typer( @@ -46,6 +48,9 @@ def sharing_list( Shows buckets shared within your organization that can be linked into your projects. Uses the regular project token. """ + if should_hint(ctx): + emit_hint(ctx, "sharing.list", project=project) + return formatter = get_formatter(ctx) service = get_service(ctx, "sharing_service") @@ -130,6 +135,17 @@ def sharing_share( Falls back to the project's configured token if no master token is set. """ + if should_hint(ctx): + emit_hint( + ctx, + "sharing.share", + project=project, + bucket_id=bucket_id, + type=sharing_type, + target_project_ids=target_project_ids, + target_users=target_users, + ) + return formatter = get_formatter(ctx) service = get_service(ctx, "sharing_service") @@ -196,6 +212,9 @@ def sharing_unshare( Requires a master token (see 'sharing share --help' for env var details). """ + if should_hint(ctx): + emit_hint(ctx, "sharing.unshare", project=project, bucket_id=bucket_id) + return formatter = get_formatter(ctx) service = get_service(ctx, "sharing_service") @@ -246,6 +265,16 @@ def sharing_link( Use 'sharing list' to discover available shared buckets. """ + if should_hint(ctx): + emit_hint( + ctx, + "sharing.link", + project=project, + source_project_id=source_project_id, + bucket_id=bucket_id, + name=name, + ) + return formatter = get_formatter(ctx) service = get_service(ctx, "sharing_service") @@ -288,6 +317,9 @@ def sharing_unlink( Deletes the linked bucket. Does not affect the source bucket or other projects that have linked it. Uses the regular project token. """ + if should_hint(ctx): + emit_hint(ctx, "sharing.unlink", project=project, bucket_id=bucket_id) + return formatter = get_formatter(ctx) service = get_service(ctx, "sharing_service") diff --git a/src/keboola_agent_cli/commands/storage.py b/src/keboola_agent_cli/commands/storage.py index 1aceefd..80a1979 100644 --- a/src/keboola_agent_cli/commands/storage.py +++ b/src/keboola_agent_cli/commands/storage.py @@ -12,11 +12,13 @@ from ..errors import ConfigError, KeboolaApiError from ._helpers import ( check_cli_permission, + emit_hint, emit_project_warnings, get_formatter, get_service, map_error_to_exit_code, resolve_branch, + should_hint, ) storage_app = typer.Typer(help="Browse and manage storage buckets, tables, and files") @@ -52,6 +54,9 @@ def storage_buckets( source project ID and name. This information is not available via MCP tools. """ + if should_hint(ctx): + emit_hint(ctx, "storage.buckets", project=project, branch=branch) + formatter = get_formatter(ctx) service = get_service(ctx, "storage_service") config_store: ConfigStore = ctx.obj["config_store"] @@ -140,6 +145,9 @@ def storage_bucket_detail( and schema from the source project. Each table includes a ready-to-use fully-qualified Snowflake path with proper quoting. """ + if should_hint(ctx): + emit_hint(ctx, "storage.bucket-detail", project=project, bucket_id=bucket_id, branch=branch) + formatter = get_formatter(ctx) service = get_service(ctx, "storage_service") config_store: ConfigStore = ctx.obj["config_store"] @@ -220,6 +228,9 @@ def storage_tables( ), ) -> None: """List storage tables from a project.""" + if should_hint(ctx): + emit_hint(ctx, "storage.tables", project=project, bucket_id=bucket_id, branch=branch) + formatter = get_formatter(ctx) service = get_service(ctx, "storage_service") config_store: ConfigStore = ctx.obj["config_store"] @@ -289,6 +300,9 @@ def storage_table_detail( ), ) -> None: """Show detailed table info including columns and types.""" + if should_hint(ctx): + emit_hint(ctx, "storage.table-detail", project=project, table_id=table_id, branch=branch) + formatter = get_formatter(ctx) service = get_service(ctx, "storage_service") config_store: ConfigStore = ctx.obj["config_store"] @@ -375,6 +389,18 @@ def storage_create_bucket( ), ) -> None: """Create a new storage bucket.""" + if should_hint(ctx): + emit_hint( + ctx, + "storage.create-bucket", + project=project, + stage=stage, + name=name, + description=description, + backend=backend, + branch=branch, + ) + formatter = get_formatter(ctx) service = get_service(ctx, "storage_service") config_store: ConfigStore = ctx.obj["config_store"] @@ -447,6 +473,18 @@ def storage_create_table( Column types: STRING, INTEGER, NUMERIC, FLOAT, BOOLEAN, DATE, TIMESTAMP. """ + if should_hint(ctx): + emit_hint( + ctx, + "storage.create-table", + project=project, + bucket_id=bucket_id, + name=name, + column=column, + primary_key=primary_key, + branch=branch, + ) + formatter = get_formatter(ctx) service = get_service(ctx, "storage_service") config_store: ConfigStore = ctx.obj["config_store"] @@ -531,6 +569,17 @@ def storage_upload_table( STRING from the CSV header). Use --no-auto-create to require the table to already exist. """ + if should_hint(ctx): + emit_hint( + ctx, + "storage.upload-table", + project=project, + table_id=table_id, + file=file, + incremental=incremental, + branch=branch, + ) + formatter = get_formatter(ctx) service = get_service(ctx, "storage_service") config_store: ConfigStore = ctx.obj["config_store"] @@ -630,6 +679,18 @@ def storage_download_table( decompression transparently. Use --columns to select specific columns and --limit to cap row count. """ + if should_hint(ctx): + emit_hint( + ctx, + "storage.download-table", + project=project, + table_id=table_id, + output=output, + columns=columns, + limit=limit, + branch=branch, + ) + formatter = get_formatter(ctx) service = get_service(ctx, "storage_service") config_store: ConfigStore = ctx.obj["config_store"] @@ -708,6 +769,16 @@ def storage_delete_table( Supports batch deletion with multiple --table-id flags. All deletes are async and wait for completion. """ + if should_hint(ctx): + emit_hint( + ctx, + "storage.delete-table", + project=project, + table_id=table_id, + dry_run=dry_run, + branch=branch, + ) + formatter = get_formatter(ctx) service = get_service(ctx, "storage_service") config_store: ConfigStore = ctx.obj["config_store"] @@ -805,6 +876,17 @@ def storage_delete_bucket( With --force, cascade-deletes all tables in the bucket. Linked and shared buckets are protected (use sharing unlink/unshare). """ + if should_hint(ctx): + emit_hint( + ctx, + "storage.delete-bucket", + project=project, + bucket_id=bucket_id, + force=force, + dry_run=dry_run, + branch=branch, + ) + formatter = get_formatter(ctx) service = get_service(ctx, "storage_service") config_store: ConfigStore = ctx.obj["config_store"] @@ -899,6 +981,18 @@ def storage_file_list( Lists files from the project's Storage Files API. Use --tag to filter by tags (AND logic - all specified tags must match). """ + if should_hint(ctx): + emit_hint( + ctx, + "storage.files", + project=project, + tag=tag, + limit=limit, + offset=offset, + query=query, + branch=branch, + ) + formatter = get_formatter(ctx) service = get_service(ctx, "storage_service") config_store: ConfigStore = ctx.obj["config_store"] @@ -969,6 +1063,9 @@ def storage_file_info( ), ) -> None: """Show Storage File metadata (without downloading).""" + if should_hint(ctx): + emit_hint(ctx, "storage.file-detail", project=project, file_id=file_id) + formatter = get_formatter(ctx) service = get_service(ctx, "storage_service") @@ -1040,6 +1137,18 @@ def storage_file_upload( Uploads any file (CSV, JSON, ZIP, etc.) to Keboola Storage Files. Use --tag to assign tags and --permanent to prevent auto-deletion. """ + if should_hint(ctx): + emit_hint( + ctx, + "storage.file-upload", + project=project, + file=file, + name=name, + tag=tag, + permanent=permanent, + branch=branch, + ) + formatter = get_formatter(ctx) service = get_service(ctx, "storage_service") config_store: ConfigStore = ctx.obj["config_store"] @@ -1115,6 +1224,16 @@ def storage_file_download( Download by file ID (--file-id) or by tags (--tag, downloads the latest matching file). Handles both sliced and non-sliced files transparently. """ + if should_hint(ctx): + emit_hint( + ctx, + "storage.file-download", + project=project, + file_id=file_id, + tag=tag, + output=output, + ) + formatter = get_formatter(ctx) service = get_service(ctx, "storage_service") @@ -1186,6 +1305,9 @@ def storage_file_tag( Use --add and --remove to modify tags in a single operation. """ + if should_hint(ctx): + emit_hint(ctx, "storage.file-tag", project=project, file_id=file_id, add=add, remove=remove) + formatter = get_formatter(ctx) service = get_service(ctx, "storage_service") @@ -1252,6 +1374,9 @@ def storage_file_delete( ), ) -> None: """Delete one or more Storage Files.""" + if should_hint(ctx): + emit_hint(ctx, "storage.file-delete", project=project, file_id=file_id, dry_run=dry_run) + formatter = get_formatter(ctx) service = get_service(ctx, "storage_service") @@ -1330,6 +1455,19 @@ def storage_load_file( Imports an already-uploaded file (from file-upload or component output) into a storage table. Use --incremental to append rows. """ + if should_hint(ctx): + emit_hint( + ctx, + "storage.load-file", + project=project, + file_id=file_id, + table_id=table_id, + incremental=incremental, + delimiter=delimiter, + enclosure=enclosure, + branch=branch, + ) + formatter = get_formatter(ctx) service = get_service(ctx, "storage_service") config_store: ConfigStore = ctx.obj["config_store"] @@ -1422,6 +1560,20 @@ def storage_unload_table( components. Use --tag to tag the output file and --download to also save it locally. """ + if should_hint(ctx): + emit_hint( + ctx, + "storage.unload-table", + project=project, + table_id=table_id, + columns=columns, + limit=limit, + tag=tag, + download=download, + output=output, + branch=branch, + ) + formatter = get_formatter(ctx) service = get_service(ctx, "storage_service") config_store: ConfigStore = ctx.obj["config_store"] diff --git a/src/keboola_agent_cli/commands/tool.py b/src/keboola_agent_cli/commands/tool.py index b014cbe..0e8b0b5 100644 --- a/src/keboola_agent_cli/commands/tool.py +++ b/src/keboola_agent_cli/commands/tool.py @@ -15,10 +15,12 @@ from ..output import OutputFormatter, format_tool_result, format_tools_table from ._helpers import ( check_cli_permission, + emit_hint, emit_project_warnings, get_formatter, get_service, resolve_branch, + should_hint, validate_branch_requires_project, ) @@ -80,6 +82,9 @@ def tool_list( ), ) -> None: """List available MCP tools from the keboola-mcp-server.""" + if should_hint(ctx): + emit_hint(ctx, "tool.list", project=project, branch=branch) + formatter = get_formatter(ctx) service = get_service(ctx, "mcp_service") config_store: ConfigStore = ctx.obj["config_store"] @@ -142,6 +147,16 @@ def tool_call( If an active branch is set for the project, it is used automatically. Use --branch to override or scope the call to a specific development branch. """ + if should_hint(ctx): + emit_hint( + ctx, + "tool.call", + tool_name=tool_name, + project=project, + input=tool_input, + branch=branch, + ) + formatter = get_formatter(ctx) service = get_service(ctx, "mcp_service") config_store: ConfigStore = ctx.obj["config_store"] diff --git a/src/keboola_agent_cli/commands/workspace.py b/src/keboola_agent_cli/commands/workspace.py index 4b159e2..0be72a0 100644 --- a/src/keboola_agent_cli/commands/workspace.py +++ b/src/keboola_agent_cli/commands/workspace.py @@ -12,10 +12,12 @@ from ..output import format_query_results, format_workspaces_table from ._helpers import ( check_cli_permission, + emit_hint, emit_project_warnings, get_formatter, get_service, map_error_to_exit_code, + should_hint, ) workspace_app = typer.Typer(help="Workspace lifecycle for SQL debugging") @@ -60,6 +62,16 @@ def workspace_create( Default: fast headless mode via Storage API (~1s). With --ui: creates via Queue job (~15s), visible in Keboola UI Workspaces tab. """ + if should_hint(ctx): + emit_hint( + ctx, + "workspace.create", + project=project, + name=name, + backend=backend, + read_only=read_only, + ) + return formatter = get_formatter(ctx) service = get_service(ctx, "workspace_service") @@ -105,6 +117,9 @@ def workspace_list( ), ) -> None: """List workspaces from connected projects.""" + if should_hint(ctx): + emit_hint(ctx, "workspace.list", project=project) + return formatter = get_formatter(ctx) service = get_service(ctx, "workspace_service") @@ -136,6 +151,9 @@ def workspace_detail( ), ) -> None: """Show workspace details (password NOT included).""" + if should_hint(ctx): + emit_hint(ctx, "workspace.detail", project=project, workspace_id=workspace_id) + return formatter = get_formatter(ctx) service = get_service(ctx, "workspace_service") @@ -183,6 +201,9 @@ def workspace_delete( ), ) -> None: """Delete a workspace.""" + if should_hint(ctx): + emit_hint(ctx, "workspace.delete", project=project, workspace_id=workspace_id) + return formatter = get_formatter(ctx) service = get_service(ctx, "workspace_service") @@ -220,6 +241,9 @@ def workspace_password( ), ) -> None: """Reset workspace password and show the new one.""" + if should_hint(ctx): + emit_hint(ctx, "workspace.password", project=project, workspace_id=workspace_id) + return formatter = get_formatter(ctx) service = get_service(ctx, "workspace_service") @@ -273,6 +297,16 @@ def workspace_load( Waits for the async load job to complete. """ + if should_hint(ctx): + emit_hint( + ctx, + "workspace.load", + project=project, + workspace_id=workspace_id, + tables=tables, + preserve=preserve, + ) + return formatter = get_formatter(ctx) service = get_service(ctx, "workspace_service") @@ -332,6 +366,16 @@ def workspace_query( Provide SQL via --sql or --file (exactly one required). """ + if should_hint(ctx): + emit_hint( + ctx, + "workspace.query", + project=project, + workspace_id=workspace_id, + sql=sql, + transactional=transactional, + ) + return formatter = get_formatter(ctx) service = get_service(ctx, "workspace_service") @@ -410,6 +454,17 @@ def workspace_from_transformation( Reads the transformation, creates a config-tied workspace, and loads all input tables. Returns credentials ready for SQL debugging. """ + if should_hint(ctx): + emit_hint( + ctx, + "workspace.from-transformation", + project=project, + component_id=component_id, + config_id=config_id, + row_id=row_id, + backend=backend, + ) + return formatter = get_formatter(ctx) service = get_service(ctx, "workspace_service") diff --git a/src/keboola_agent_cli/hints/__init__.py b/src/keboola_agent_cli/hints/__init__.py new file mode 100644 index 0000000..0d6b1cf --- /dev/null +++ b/src/keboola_agent_cli/hints/__init__.py @@ -0,0 +1,78 @@ +"""Hint system — generates equivalent Python code for CLI commands. + +Usage: + from keboola_agent_cli.hints import HintRegistry, render_hint + from keboola_agent_cli.hints.models import HintMode + + output = render_hint("config.list", HintMode.CLIENT, params, stack_url, ...) +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any, ClassVar + +from .models import CommandHint, HintMode + + +class HintRegistry: + """Registry of hint definitions for CLI commands. + + Hints are registered at import time by definition modules. + """ + + _hints: ClassVar[dict[str, CommandHint]] = {} + + @classmethod + def register(cls, hint: CommandHint) -> None: + """Register a hint definition.""" + cls._hints[hint.cli_command] = hint + + @classmethod + def get(cls, command: str) -> CommandHint | None: + """Look up a hint by command key (e.g. 'config.list').""" + return cls._hints.get(command) + + @classmethod + def all_commands(cls) -> list[str]: + """Return all registered command keys, sorted.""" + return sorted(cls._hints.keys()) + + +def render_hint( + cli_command: str, + hint_mode: HintMode, + params: dict[str, Any], + stack_url: str | None, + config_dir: Path | None, + branch_id: int | None, +) -> str: + """Render Python code for a CLI command. + + Args: + cli_command: Dot-separated command key, e.g. 'config.list'. + hint_mode: CLIENT or SERVICE layer. + params: CLI parameters passed to the command. + stack_url: Resolved stack URL (or None for placeholder). + config_dir: Resolved config directory path (for service hints). + branch_id: Active branch ID (or None for production). + + Returns: + String of runnable Python code. + + Raises: + ValueError: If no hint is registered for the given command. + """ + # Ensure definitions are loaded + from . import definitions as _definitions # noqa: F401 + + hint = HintRegistry.get(cli_command) + if hint is None: + msg = f"No hint available for command '{cli_command}'." + raise ValueError(msg) + + from .renderer import ClientRenderer, ServiceRenderer + + if hint_mode == HintMode.SERVICE: + return ServiceRenderer.render(hint, params, stack_url, config_dir, branch_id) + return ClientRenderer.render(hint, params, stack_url, branch_id, config_dir=config_dir) diff --git a/src/keboola_agent_cli/hints/definitions/__init__.py b/src/keboola_agent_cli/hints/definitions/__init__.py new file mode 100644 index 0000000..189587d --- /dev/null +++ b/src/keboola_agent_cli/hints/definitions/__init__.py @@ -0,0 +1,15 @@ +"""Hint definitions — import all modules to trigger registration.""" + +from . import ( + branch, # noqa: F401 + component, # noqa: F401 + config, # noqa: F401 + encrypt, # noqa: F401 + job, # noqa: F401 + lineage, # noqa: F401 + org, # noqa: F401 + sharing, # noqa: F401 + storage, # noqa: F401 + tool, # noqa: F401 + workspace, # noqa: F401 +) diff --git a/src/keboola_agent_cli/hints/definitions/branch.py b/src/keboola_agent_cli/hints/definitions/branch.py new file mode 100644 index 0000000..846e38a --- /dev/null +++ b/src/keboola_agent_cli/hints/definitions/branch.py @@ -0,0 +1,87 @@ +"""Hint definitions for branch commands (list, create, delete).""" + +from .. import HintRegistry +from ..models import ClientCall, CommandHint, HintStep, ServiceCall + +# ── branch list ──────────────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="branch.list", + description="List development branches", + steps=[ + HintStep( + comment="List dev branches", + client=ClientCall( + method="list_dev_branches", + args={}, + result_var="branches", + result_hint="list[dict]", + ), + service=ServiceCall( + service_class="BranchService", + service_module="branch_service", + method="list_branches", + args={"aliases": "{project}"}, + ), + ), + ], + ) +) + +# ── branch create ────────────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="branch.create", + description="Create a new development branch", + steps=[ + HintStep( + comment="Create dev branch", + client=ClientCall( + method="create_dev_branch", + args={"name": "{name}", "description": "{description}"}, + result_var="branch", + result_hint="dict", + ), + service=ServiceCall( + service_class="BranchService", + service_module="branch_service", + method="create_branch", + args={ + "alias": "{project}", + "name": "{name}", + "description": "{description}", + }, + ), + ), + ], + notes=["Service layer also auto-activates the branch in CLI config."], + ) +) + +# ── branch delete ────────────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="branch.delete", + description="Delete a development branch", + steps=[ + HintStep( + comment="Delete dev branch", + client=ClientCall( + method="delete_dev_branch", + args={"branch_id": "{branch}"}, + result_var="result", + ), + service=ServiceCall( + service_class="BranchService", + service_module="branch_service", + method="delete_branch", + args={"alias": "{project}", "branch_id": "{branch}"}, + ), + ), + ], + notes=["Service layer auto-resets active branch if the deleted branch was active."], + ) +) diff --git a/src/keboola_agent_cli/hints/definitions/component.py b/src/keboola_agent_cli/hints/definitions/component.py new file mode 100644 index 0000000..1a65844 --- /dev/null +++ b/src/keboola_agent_cli/hints/definitions/component.py @@ -0,0 +1,68 @@ +"""Hint definitions for component commands (list, detail).""" + +from .. import HintRegistry +from ..models import ClientCall, CommandHint, HintStep, ServiceCall + +# ── component list ───────────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="component.list", + description="List available components", + steps=[ + HintStep( + comment="List components", + client=ClientCall( + method="list_components", + args={"component_type": "{type}"}, + result_var="components", + result_hint="list[dict]", + ), + service=ServiceCall( + service_class="ComponentService", + service_module="component_service", + method="list_components", + args={ + "aliases": "{project}", + "component_type": "{type}", + "query": "{query}", + }, + ), + ), + ], + notes=["With --query, service uses the AI search API for smart matching."], + ) +) + +# ── component detail ─────────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="component.detail", + description="Show component detail with schema summary", + steps=[ + HintStep( + comment="Get component detail via AI Service API", + client=ClientCall( + method="get_component_detail", + args={"component_id": "{component_id}"}, + result_var="detail", + result_hint="dict", + ), + service=ServiceCall( + service_class="ComponentService", + service_module="component_service", + method="get_component_detail", + args={ + "alias": "{project}", + "component_id": "{component_id}", + }, + ), + ), + ], + notes=[ + "Uses the AI Service API (ai.{stack}) for enriched schema information.", + "Client layer: use AiServiceClient, not KeboolaClient.", + ], + ) +) diff --git a/src/keboola_agent_cli/hints/definitions/config.py b/src/keboola_agent_cli/hints/definitions/config.py new file mode 100644 index 0000000..ed1ebe4 --- /dev/null +++ b/src/keboola_agent_cli/hints/definitions/config.py @@ -0,0 +1,120 @@ +"""Hint definitions for config commands (list, detail, search).""" + +from .. import HintRegistry +from ..models import ClientCall, CommandHint, HintStep, ServiceCall + +# ── config list ──────────────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="config.list", + description="List configurations from connected projects", + steps=[ + HintStep( + comment="List all components with their configurations", + client=ClientCall( + method="list_components", + args={ + "component_type": "{component_type}", + "branch_id": "{branch}", + }, + result_var="components", + result_hint="list[dict]", + ), + service=ServiceCall( + service_class="ConfigService", + service_module="config_service", + method="list_configs", + args={ + "aliases": "{project}", + "component_type": "{component_type}", + "component_id": "{component_id}", + "branch_id": "{branch}", + }, + ), + ), + ], + notes=[ + "Each component in the response has a 'configurations' list.", + "Service layer returns {'configs': [...], 'errors': [...]} with flattened results.", + ], + ) +) + +# ── config detail ────────────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="config.detail", + description="Show detailed information about a specific configuration", + steps=[ + HintStep( + comment="Get configuration detail", + client=ClientCall( + method="get_config_detail", + args={ + "component_id": "{component_id}", + "config_id": "{config_id}", + "branch_id": "{branch}", + }, + result_var="detail", + result_hint="dict", + ), + service=ServiceCall( + service_class="ConfigService", + service_module="config_service", + method="get_config_detail", + args={ + "alias": "{project}", + "component_id": "{component_id}", + "config_id": "{config_id}", + "branch_id": "{branch}", + }, + ), + ), + ], + ) +) + +# ── config search ────────────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="config.search", + description="Search through configuration bodies across projects", + steps=[ + HintStep( + comment="Search configurations for a pattern", + client=ClientCall( + method="list_components", + args={ + "component_type": "{component_type}", + "branch_id": "{branch}", + }, + result_var="components", + result_hint="list[dict]", + ), + service=ServiceCall( + service_class="ConfigService", + service_module="config_service", + method="search_configs", + args={ + "query": "{query}", + "aliases": "{project}", + "component_type": "{component_type}", + "component_id": "{component_id}", + "ignore_case": "{ignore_case}", + "use_regex": "{regex}", + "branch_id": "{branch}", + }, + ), + ), + ], + notes=[ + "Client layer returns raw components — you need to search through " + "configuration JSON bodies yourself.", + "Service layer does the full-text search and returns " + "{'matches': [...], 'errors': [...], 'stats': {...}}.", + ], + ) +) diff --git a/src/keboola_agent_cli/hints/definitions/encrypt.py b/src/keboola_agent_cli/hints/definitions/encrypt.py new file mode 100644 index 0000000..e735485 --- /dev/null +++ b/src/keboola_agent_cli/hints/definitions/encrypt.py @@ -0,0 +1,39 @@ +"""Hint definitions for encrypt commands.""" + +from .. import HintRegistry +from ..models import ClientCall, CommandHint, HintStep, ServiceCall + +# ── encrypt values ───────────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="encrypt.values", + description="Encrypt configuration values", + steps=[ + HintStep( + comment="Encrypt values via Encryption API", + client=ClientCall( + method="encrypt_values", + args={"component_id": "{component_id}", "data": "{input}"}, + result_var="encrypted", + result_hint="dict", + ), + service=ServiceCall( + service_class="EncryptService", + service_module="encrypt_service", + method="encrypt", + args={ + "alias": "{project}", + "component_id": "{component_id}", + "input_data": "{input}", + }, + ), + ), + ], + notes=[ + "Uses the Encryption API (encryption.keboola.com).", + "Values must be prefixed with '#' to be encrypted.", + "Already-encrypted values (KBC:: prefix) are passed through.", + ], + ) +) diff --git a/src/keboola_agent_cli/hints/definitions/job.py b/src/keboola_agent_cli/hints/definitions/job.py new file mode 100644 index 0000000..da5ebc9 --- /dev/null +++ b/src/keboola_agent_cli/hints/definitions/job.py @@ -0,0 +1,122 @@ +"""Hint definitions for job commands (list, detail, run).""" + +from .. import HintRegistry +from ..models import ClientCall, CommandHint, HintStep, ServiceCall + +# ── job list ─────────────────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="job.list", + description="List recent jobs from the Queue API", + steps=[ + HintStep( + comment="List jobs", + client=ClientCall( + method="list_jobs", + args={ + "component_id": "{component_id}", + "config_id": "{config_id}", + "status": "{status}", + "limit": "{limit}", + }, + result_var="jobs", + result_hint="list[dict]", + ), + service=ServiceCall( + service_class="JobService", + service_module="job_service", + method="list_jobs", + args={ + "aliases": "{project}", + "component_id": "{component_id}", + "config_id": "{config_id}", + "status": "{status}", + "limit": "{limit}", + }, + ), + ), + ], + notes=["Uses the Queue API (queue.keboola.com), not Storage API."], + ) +) + +# ── job detail ───────────────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="job.detail", + description="Show detailed job information", + steps=[ + HintStep( + comment="Get job detail from Queue API", + client=ClientCall( + method="get_job_detail", + args={"job_id": "{job_id}"}, + result_var="job", + result_hint="dict", + ), + service=ServiceCall( + service_class="JobService", + service_module="job_service", + method="get_job_detail", + args={"alias": "{project}", "job_id": "{job_id}"}, + ), + ), + ], + notes=["Uses the Queue API (queue.keboola.com), not Storage API."], + ) +) + +# ── job run ──────────────────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="job.run", + description="Run a component configuration as a job", + steps=[ + HintStep( + comment="Create and submit job to Queue API", + client=ClientCall( + method="create_job", + args={ + "component_id": "{component_id}", + "config_id": "{config_id}", + "config_row_ids": "{row_id}", + }, + result_var="job", + result_hint="dict", + ), + service=ServiceCall( + service_class="JobService", + service_module="job_service", + method="run_job", + args={ + "alias": "{project}", + "component_id": "{component_id}", + "config_id": "{config_id}", + "config_row_ids": "{row_id}", + "wait": "{wait}", + "timeout": "{timeout}", + }, + ), + ), + HintStep( + comment="Poll until job completes (when --wait is used)", + client=ClientCall( + method="get_job_detail", + args={"job_id": 'str(job["id"])'}, + result_var="job", + ), + kind="poll_loop", + poll_interval=5.0, + poll_condition='not job.get("isFinished")', + ), + ], + notes=[ + "Uses the Queue API (queue.keboola.com), not Storage API.", + "Without --wait, returns immediately after job creation.", + "Service layer handles both create + optional poll in one call.", + ], + ) +) diff --git a/src/keboola_agent_cli/hints/definitions/lineage.py b/src/keboola_agent_cli/hints/definitions/lineage.py new file mode 100644 index 0000000..40f7ab7 --- /dev/null +++ b/src/keboola_agent_cli/hints/definitions/lineage.py @@ -0,0 +1,35 @@ +"""Hint definitions for lineage commands.""" + +from .. import HintRegistry +from ..models import ClientCall, CommandHint, HintStep, ServiceCall + +# ── lineage show ─────────────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="lineage.show", + description="Show cross-project data lineage via bucket sharing", + steps=[ + HintStep( + comment="List buckets with linked-bucket metadata", + client=ClientCall( + method="list_buckets", + args={"include": '"linkedBuckets"'}, + result_var="buckets", + result_hint="list[dict]", + ), + service=ServiceCall( + service_class="LineageService", + service_module="lineage_service", + method="get_lineage", + args={"aliases": "{project}"}, + ), + ), + ], + notes=[ + "Client layer returns raw buckets — analyze sharing metadata yourself.", + "Service layer queries all projects in parallel and builds " + "data flow edges with deduplication.", + ], + ) +) diff --git a/src/keboola_agent_cli/hints/definitions/org.py b/src/keboola_agent_cli/hints/definitions/org.py new file mode 100644 index 0000000..c704fec --- /dev/null +++ b/src/keboola_agent_cli/hints/definitions/org.py @@ -0,0 +1,39 @@ +"""Hint definitions for org commands.""" + +from .. import HintRegistry +from ..models import ClientCall, CommandHint, HintStep, ServiceCall + +# ── org setup ────────────────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="org.setup", + description="Onboard organization projects into kbagent", + steps=[ + HintStep( + comment="List organization projects via Manage API", + client=ClientCall( + method="list_organization_projects", + args={"org_id": "{org_id}"}, + client_type="manage", + result_var="projects", + result_hint="list[dict]", + ), + service=ServiceCall( + service_class="OrgService", + service_module="org_service", + method="setup_organization", + args={ + "stack_url": "{url}", + "org_id": "{org_id}", + "dry_run": "{dry_run}", + }, + ), + ), + ], + notes=[ + "Uses the Manage API with KBC_MANAGE_API_TOKEN (not Storage token).", + "Service layer creates per-project tokens and registers them in CLI config.", + ], + ) +) diff --git a/src/keboola_agent_cli/hints/definitions/sharing.py b/src/keboola_agent_cli/hints/definitions/sharing.py new file mode 100644 index 0000000..406028f --- /dev/null +++ b/src/keboola_agent_cli/hints/definitions/sharing.py @@ -0,0 +1,153 @@ +"""Hint definitions for sharing commands (list, share, unshare, link, unlink).""" + +from .. import HintRegistry +from ..models import ClientCall, CommandHint, HintStep, ServiceCall + +# ── sharing list ─────────────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="sharing.list", + description="List shared and linked buckets", + steps=[ + HintStep( + comment="List shared buckets", + client=ClientCall( + method="list_shared_buckets", + args={}, + result_var="shared", + result_hint="list[dict]", + ), + service=ServiceCall( + service_class="SharingService", + service_module="sharing_service", + method="list_shared", + args={"aliases": "{project}"}, + ), + ), + ], + ) +) + +# ── sharing share ────────────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="sharing.share", + description="Share a bucket with other projects", + steps=[ + HintStep( + comment="Share bucket", + client=ClientCall( + method="share_bucket", + args={ + "bucket_id": "{bucket_id}", + "sharing_type": "{type}", + "target_project_ids": "{target_project_ids}", + "target_users": "{target_users}", + }, + result_var="result", + ), + service=ServiceCall( + service_class="SharingService", + service_module="sharing_service", + method="share", + args={ + "alias": "{project}", + "bucket_id": "{bucket_id}", + "sharing_type": "{type}", + "target_project_ids": "{target_project_ids}", + "target_users": "{target_users}", + }, + ), + ), + ], + notes=["May require a master token for organization-level sharing."], + ) +) + +# ── sharing unshare ──────────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="sharing.unshare", + description="Stop sharing a bucket", + steps=[ + HintStep( + comment="Unshare bucket", + client=ClientCall( + method="unshare_bucket", + args={"bucket_id": "{bucket_id}"}, + result_var="result", + ), + service=ServiceCall( + service_class="SharingService", + service_module="sharing_service", + method="unshare", + args={"alias": "{project}", "bucket_id": "{bucket_id}"}, + ), + ), + ], + notes=["Requires a master token."], + ) +) + +# ── sharing link ─────────────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="sharing.link", + description="Link a shared bucket into this project", + steps=[ + HintStep( + comment="Link shared bucket", + client=ClientCall( + method="link_bucket", + args={ + "source_project_id": "{source_project_id}", + "source_bucket_id": "{bucket_id}", + "name": "{name}", + }, + result_var="result", + ), + service=ServiceCall( + service_class="SharingService", + service_module="sharing_service", + method="link", + args={ + "alias": "{project}", + "source_project_id": "{source_project_id}", + "source_bucket_id": "{bucket_id}", + "name": "{name}", + }, + ), + ), + ], + ) +) + +# ── sharing unlink ───────────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="sharing.unlink", + description="Remove a linked bucket from this project", + steps=[ + HintStep( + comment="Unlink bucket", + client=ClientCall( + method="delete_bucket", + args={"bucket_id": "{bucket_id}"}, + result_var="result", + ), + service=ServiceCall( + service_class="SharingService", + service_module="sharing_service", + method="unlink", + args={"alias": "{project}", "bucket_id": "{bucket_id}"}, + ), + ), + ], + notes=["Service layer validates the bucket is linked before deleting."], + ) +) diff --git a/src/keboola_agent_cli/hints/definitions/storage.py b/src/keboola_agent_cli/hints/definitions/storage.py new file mode 100644 index 0000000..6381295 --- /dev/null +++ b/src/keboola_agent_cli/hints/definitions/storage.py @@ -0,0 +1,628 @@ +"""Hint definitions for storage commands (buckets, tables, files).""" + +from .. import HintRegistry +from ..models import ClientCall, CommandHint, HintStep, ServiceCall + +# ── storage buckets ──────────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="storage.buckets", + description="List storage buckets", + steps=[ + HintStep( + comment="List all buckets", + client=ClientCall( + method="list_buckets", + args={"include": '"linkedBuckets"', "branch_id": "{branch}"}, + result_var="buckets", + result_hint="list[dict]", + ), + service=ServiceCall( + service_class="StorageService", + service_module="storage_service", + method="list_buckets", + args={"aliases": "{project}", "branch_id": "{branch}"}, + ), + ), + ], + ) +) + +# ── storage bucket-detail ────────���───────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="storage.bucket-detail", + description="Show bucket detail with tables", + steps=[ + HintStep( + comment="Get bucket detail", + client=ClientCall( + method="get_bucket_detail", + args={"bucket_id": "{bucket_id}", "branch_id": "{branch}"}, + result_var="bucket", + result_hint="dict", + ), + service=ServiceCall( + service_class="StorageService", + service_module="storage_service", + method="get_bucket_detail", + args={ + "alias": "{project}", + "bucket_id": "{bucket_id}", + "branch_id": "{branch}", + }, + ), + ), + ], + ) +) + +# ── storage create-bucket ────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="storage.create-bucket", + description="Create a new storage bucket", + steps=[ + HintStep( + comment="Create bucket", + client=ClientCall( + method="create_bucket", + args={ + "stage": "{stage}", + "name": "{name}", + "description": "{description}", + "backend": "{backend}", + "branch_id": "{branch}", + }, + result_var="bucket", + result_hint="dict", + ), + service=ServiceCall( + service_class="StorageService", + service_module="storage_service", + method="create_bucket", + args={ + "alias": "{project}", + "stage": "{stage}", + "name": "{name}", + "description": "{description}", + "backend": "{backend}", + "branch_id": "{branch}", + }, + ), + ), + ], + ) +) + +# ─��� storage delete-bucket ────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="storage.delete-bucket", + description="Delete one or more storage buckets", + steps=[ + HintStep( + comment="Delete bucket(s)", + client=ClientCall( + method="delete_bucket", + args={ + "bucket_id": "{bucket_id}", + "force": "{force}", + "branch_id": "{branch}", + }, + result_var="result", + ), + service=ServiceCall( + service_class="StorageService", + service_module="storage_service", + method="delete_buckets", + args={ + "alias": "{project}", + "bucket_ids": "{bucket_id}", + "force": "{force}", + "dry_run": "{dry_run}", + "branch_id": "{branch}", + }, + ), + ), + ], + notes=["Client layer deletes one bucket at a time. Loop for batch."], + ) +) + +# ── storage tables ─────────────────────────────────���─────────────── + +HintRegistry.register( + CommandHint( + cli_command="storage.tables", + description="List tables in a project", + steps=[ + HintStep( + comment="List tables", + client=ClientCall( + method="list_tables", + args={ + "bucket_id": "{bucket_id}", + "include": '"columns"', + "branch_id": "{branch}", + }, + result_var="tables", + result_hint="list[dict]", + ), + service=ServiceCall( + service_class="StorageService", + service_module="storage_service", + method="list_tables", + args={ + "alias": "{project}", + "bucket_id": "{bucket_id}", + "branch_id": "{branch}", + }, + ), + ), + ], + ) +) + +# ── storage table-detail ─────��─────────────────────────────────���─── + +HintRegistry.register( + CommandHint( + cli_command="storage.table-detail", + description="Show detailed table information", + steps=[ + HintStep( + comment="Get table detail", + client=ClientCall( + method="get_table_detail", + args={"table_id": "{table_id}", "branch_id": "{branch}"}, + result_var="table", + result_hint="dict", + ), + service=ServiceCall( + service_class="StorageService", + service_module="storage_service", + method="get_table_detail", + args={ + "alias": "{project}", + "table_id": "{table_id}", + "branch_id": "{branch}", + }, + ), + ), + ], + ) +) + +# ── storage create-table ─────────────────────────────────────���───── + +HintRegistry.register( + CommandHint( + cli_command="storage.create-table", + description="Create a new table with typed columns", + steps=[ + HintStep( + comment="Create table", + client=ClientCall( + method="create_table", + args={ + "bucket_id": "{bucket_id}", + "name": "{name}", + "columns": "{column}", + "primary_key": "{primary_key}", + "branch_id": "{branch}", + }, + result_var="table", + result_hint="dict", + ), + service=ServiceCall( + service_class="StorageService", + service_module="storage_service", + method="create_table", + args={ + "alias": "{project}", + "bucket_id": "{bucket_id}", + "name": "{name}", + "columns": "{column}", + "primary_key": "{primary_key}", + "branch_id": "{branch}", + }, + ), + ), + ], + notes=["Columns format: 'name:TYPE' (e.g. 'id:INTEGER', 'name:STRING')."], + ) +) + +# ── storage upload-table ─────────────────────────────────────────��─ + +HintRegistry.register( + CommandHint( + cli_command="storage.upload-table", + description="Upload a CSV file into a table", + steps=[ + HintStep( + comment="Upload CSV file to table (handles file upload + async import)", + client=ClientCall( + method="upload_table", + args={ + "table_id": "{table_id}", + "file_path": "{file}", + "incremental": "{incremental}", + "branch_id": "{branch}", + }, + result_var="result", + ), + service=ServiceCall( + service_class="StorageService", + service_module="storage_service", + method="upload_table", + args={ + "alias": "{project}", + "table_id": "{table_id}", + "file_path": "{file}", + "incremental": "{incremental}", + "branch_id": "{branch}", + }, + ), + ), + ], + notes=[ + "Internally: prepare upload -> upload to cloud -> async import job.", + "With --auto-create, bucket and table are created if missing.", + ], + ) +) + +# ��─ storage download-table ──���──────────────────────────────────���─── + +HintRegistry.register( + CommandHint( + cli_command="storage.download-table", + description="Download a table to a local CSV file", + steps=[ + HintStep( + comment="Export and download table data", + client=ClientCall( + method="export_table_async", + args={ + "table_id": "{table_id}", + "columns": "{columns}", + "limit": "{limit}", + "branch_id": "{branch}", + }, + result_var="result", + ), + service=ServiceCall( + service_class="StorageService", + service_module="storage_service", + method="download_table", + args={ + "alias": "{project}", + "table_id": "{table_id}", + "output_path": "{output}", + "columns": "{columns}", + "limit": "{limit}", + "branch_id": "{branch}", + }, + ), + ), + ], + notes=[ + "Client layer: export_table_async -> get_file_info -> download_file.", + "Service layer handles the full flow including CSV header prepending.", + ], + ) +) + +# ─�� storage delete-table ─────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="storage.delete-table", + description="Delete one or more tables", + steps=[ + HintStep( + comment="Delete table(s)", + client=ClientCall( + method="delete_table", + args={"table_id": "{table_id}", "branch_id": "{branch}"}, + result_var="result", + ), + service=ServiceCall( + service_class="StorageService", + service_module="storage_service", + method="delete_tables", + args={ + "alias": "{project}", + "table_ids": "{table_id}", + "dry_run": "{dry_run}", + "branch_id": "{branch}", + }, + ), + ), + ], + notes=["Client layer deletes one table at a time. Loop for batch."], + ) +) + +# ── storage files ──────��──────────────────────────────────���──────── + +HintRegistry.register( + CommandHint( + cli_command="storage.files", + description="List files in Storage", + steps=[ + HintStep( + comment="List files", + client=ClientCall( + method="list_files", + args={ + "limit": "{limit}", + "offset": "{offset}", + "tags": "{tag}", + "query": "{query}", + "branch_id": "{branch}", + }, + result_var="files", + result_hint="list[dict]", + ), + service=ServiceCall( + service_class="StorageService", + service_module="storage_service", + method="list_files", + args={ + "alias": "{project}", + "limit": "{limit}", + "offset": "{offset}", + "tags": "{tag}", + "query": "{query}", + "branch_id": "{branch}", + }, + ), + ), + ], + ) +) + +# ── storage file-detail ──────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="storage.file-detail", + description="Show file detail", + steps=[ + HintStep( + comment="Get file info", + client=ClientCall( + method="get_file_info", + args={"file_id": "{file_id}"}, + result_var="file_info", + result_hint="dict", + ), + service=ServiceCall( + service_class="StorageService", + service_module="storage_service", + method="get_file_info", + args={"alias": "{project}", "file_id": "{file_id}"}, + ), + ), + ], + ) +) + +# ── storage file-upload ──────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="storage.file-upload", + description="Upload a file to Storage", + steps=[ + HintStep( + comment="Upload file to Storage", + client=ClientCall( + method="upload_file", + args={ + "file_path": "{file}", + "tags": "{tag}", + "is_permanent": "{permanent}", + "branch_id": "{branch}", + }, + result_var="result", + ), + service=ServiceCall( + service_class="StorageService", + service_module="storage_service", + method="upload_file", + args={ + "alias": "{project}", + "file_path": "{file}", + "name": "{name}", + "tags": "{tag}", + "is_permanent": "{permanent}", + "branch_id": "{branch}", + }, + ), + ), + ], + notes=["Internally: prepare upload -> upload to cloud storage (S3/GCS/Azure)."], + ) +) + +# ── storage file-download ────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="storage.file-download", + description="Download a file from Storage", + steps=[ + HintStep( + comment="Download file from Storage", + client=ClientCall( + method="get_file_info", + args={"file_id": "{file_id}"}, + result_var="file_info", + ), + service=ServiceCall( + service_class="StorageService", + service_module="storage_service", + method="download_file", + args={ + "alias": "{project}", + "file_id": "{file_id}", + "tags": "{tag}", + "output_path": "{output}", + }, + ), + ), + ], + notes=[ + "Client layer: get_file_info -> download from cloud URL.", + "Service layer handles tag-based lookup and sliced file assembly.", + ], + ) +) + +# ── storage file-delete ─────────────────────────────────────────��── + +HintRegistry.register( + CommandHint( + cli_command="storage.file-delete", + description="Delete one or more files", + steps=[ + HintStep( + comment="Delete file(s)", + client=ClientCall( + method="delete_file", + args={"file_id": "{file_id}"}, + result_var="result", + ), + service=ServiceCall( + service_class="StorageService", + service_module="storage_service", + method="delete_files", + args={ + "alias": "{project}", + "file_ids": "{file_id}", + "dry_run": "{dry_run}", + }, + ), + ), + ], + ) +) + +# ── storage file-tag ────────���────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="storage.file-tag", + description="Add or remove tags on a file", + steps=[ + HintStep( + comment="Manage file tags", + client=ClientCall( + method="tag_file", + args={"file_id": "{file_id}", "tag": "{add}"}, + result_var="result", + ), + service=ServiceCall( + service_class="StorageService", + service_module="storage_service", + method="tag_file", + args={ + "alias": "{project}", + "file_id": "{file_id}", + "add_tags": "{add}", + "remove_tags": "{remove}", + }, + ), + ), + ], + notes=["Client layer: call tag_file() per tag. Service handles add + remove."], + ) +) + +# ── storage load-file ─────────────────────────────────────���──────── + +HintRegistry.register( + CommandHint( + cli_command="storage.load-file", + description="Load an existing Storage file into a table", + steps=[ + HintStep( + comment="Import file into table (async)", + client=ClientCall( + method="import_table_async", + args={ + "table_id": "{table_id}", + "file_id": "{file_id}", + "incremental": "{incremental}", + "delimiter": "{delimiter}", + "enclosure": "{enclosure}", + "branch_id": "{branch}", + }, + result_var="result", + ), + service=ServiceCall( + service_class="StorageService", + service_module="storage_service", + method="load_file_to_table", + args={ + "alias": "{project}", + "file_id": "{file_id}", + "table_id": "{table_id}", + "incremental": "{incremental}", + "delimiter": "{delimiter}", + "enclosure": "{enclosure}", + "branch_id": "{branch}", + }, + ), + ), + ], + ) +) + +# ─�� storage unload-table ───────────────────────────────────────��─── + +HintRegistry.register( + CommandHint( + cli_command="storage.unload-table", + description="Export a table to a Storage file", + steps=[ + HintStep( + comment="Export table to Storage file (async)", + client=ClientCall( + method="export_table_async", + args={ + "table_id": "{table_id}", + "columns": "{columns}", + "limit": "{limit}", + "branch_id": "{branch}", + }, + result_var="result", + ), + service=ServiceCall( + service_class="StorageService", + service_module="storage_service", + method="unload_table_to_file", + args={ + "alias": "{project}", + "table_id": "{table_id}", + "columns": "{columns}", + "limit": "{limit}", + "tags": "{tag}", + "download": "{download}", + "output_path": "{output}", + "branch_id": "{branch}", + }, + ), + ), + ], + notes=["Service layer handles optional tagging and local download."], + ) +) diff --git a/src/keboola_agent_cli/hints/definitions/tool.py b/src/keboola_agent_cli/hints/definitions/tool.py new file mode 100644 index 0000000..b920456 --- /dev/null +++ b/src/keboola_agent_cli/hints/definitions/tool.py @@ -0,0 +1,79 @@ +"""Hint definitions for MCP tool commands (list, call).""" + +from .. import HintRegistry +from ..models import ClientCall, CommandHint, HintStep, ServiceCall + +# ── tool list ────────────────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="tool.list", + description="List available MCP tools from keboola-mcp-server", + steps=[ + HintStep( + comment="List MCP tools", + client=ClientCall( + method="list_tools", + args={"aliases": "{project}", "branch_id": "{branch}"}, + client_type="mcp", + result_var="tools", + result_hint="dict", + ), + service=ServiceCall( + service_class="McpService", + service_module="mcp_service", + method="list_tools", + args={"aliases": "{project}", "branch_id": "{branch}"}, + ), + ), + ], + notes=[ + "MCP tools are served by keboola-mcp-server (installed separately).", + "Client layer: use McpService directly (there is no raw HTTP equivalent).", + "Tools are the same across projects — only one project is queried.", + ], + ) +) + +# ── tool call ────────────────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="tool.call", + description="Call an MCP tool on keboola-mcp-server", + steps=[ + HintStep( + comment="Validate and call MCP tool", + client=ClientCall( + method="validate_and_call_tool", + args={ + "tool_name": "{tool_name}", + "tool_input": "{input}", + "alias": "{project}", + "branch_id": "{branch}", + }, + client_type="mcp", + result_var="result", + result_hint="dict", + ), + service=ServiceCall( + service_class="McpService", + service_module="mcp_service", + method="validate_and_call_tool", + args={ + "tool_name": "{tool_name}", + "tool_input": "{input}", + "alias": "{project}", + "branch_id": "{branch}", + }, + ), + ), + ], + notes=[ + "MCP tools are served by keboola-mcp-server (installed separately).", + "Client layer: use McpService directly (there is no raw HTTP equivalent).", + "Read tools run across ALL projects in parallel.", + "Write tools require --project to specify the target.", + ], + ) +) diff --git a/src/keboola_agent_cli/hints/definitions/workspace.py b/src/keboola_agent_cli/hints/definitions/workspace.py new file mode 100644 index 0000000..436ecb6 --- /dev/null +++ b/src/keboola_agent_cli/hints/definitions/workspace.py @@ -0,0 +1,254 @@ +"""Hint definitions for workspace commands.""" + +from .. import HintRegistry +from ..models import ClientCall, CommandHint, HintStep, ServiceCall + +# ── workspace create ─────────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="workspace.create", + description="Create a new SQL workspace", + steps=[ + HintStep( + comment="Create workspace (headless mode)", + client=ClientCall( + method="create_config_workspace", + args={"backend": "{backend}"}, + result_var="workspace", + result_hint="dict", + ), + service=ServiceCall( + service_class="WorkspaceService", + service_module="workspace_service", + method="create_workspace", + args={ + "alias": "{project}", + "name": "{name}", + "backend": "{backend}", + "read_only": "{read_only}", + }, + ), + ), + ], + notes=[ + "Service layer handles sandbox config creation + workspace provisioning.", + "With --ui flag, creates via job run (slower, ~15s) for UI visibility.", + ], + ) +) + +# ── workspace list ───────────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="workspace.list", + description="List workspaces across projects", + steps=[ + HintStep( + comment="List workspaces", + client=ClientCall( + method="list_workspaces", + args={"branch_id": "{branch}"}, + result_var="workspaces", + result_hint="list[dict]", + ), + service=ServiceCall( + service_class="WorkspaceService", + service_module="workspace_service", + method="list_workspaces", + args={"aliases": "{project}"}, + ), + ), + ], + ) +) + +# ── workspace detail ─────────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="workspace.detail", + description="Show workspace detail", + steps=[ + HintStep( + comment="Get workspace detail", + client=ClientCall( + method="get_workspace", + args={"workspace_id": "{workspace_id}"}, + result_var="workspace", + result_hint="dict", + ), + service=ServiceCall( + service_class="WorkspaceService", + service_module="workspace_service", + method="get_workspace", + args={"alias": "{project}", "workspace_id": "{workspace_id}"}, + ), + ), + ], + ) +) + +# ── workspace delete ─────────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="workspace.delete", + description="Delete a workspace and its config", + steps=[ + HintStep( + comment="Delete workspace", + client=ClientCall( + method="delete_workspace", + args={"workspace_id": "{workspace_id}"}, + result_var="result", + ), + service=ServiceCall( + service_class="WorkspaceService", + service_module="workspace_service", + method="delete_workspace", + args={"alias": "{project}", "workspace_id": "{workspace_id}"}, + ), + ), + ], + notes=["Service layer also cleans up the associated sandbox configuration."], + ) +) + +# ── workspace password ───────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="workspace.password", + description="Reset workspace password", + steps=[ + HintStep( + comment="Reset workspace password", + client=ClientCall( + method="reset_workspace_password", + args={"workspace_id": "{workspace_id}"}, + result_var="credentials", + result_hint="dict", + ), + service=ServiceCall( + service_class="WorkspaceService", + service_module="workspace_service", + method="reset_password", + args={"alias": "{project}", "workspace_id": "{workspace_id}"}, + ), + ), + ], + ) +) + +# ── workspace load ───────────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="workspace.load", + description="Load tables into a workspace", + steps=[ + HintStep( + comment="Load tables into workspace", + client=ClientCall( + method="load_workspace_tables", + args={"workspace_id": "{workspace_id}", "tables": "{tables}"}, + result_var="result", + ), + service=ServiceCall( + service_class="WorkspaceService", + service_module="workspace_service", + method="load_tables", + args={ + "alias": "{project}", + "workspace_id": "{workspace_id}", + "tables": "{tables}", + "preserve": "{preserve}", + }, + ), + ), + ], + notes=["Tables format: 'bucket.table' or 'bucket.table/dest_name'."], + ) +) + +# ── workspace query ──────────────────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="workspace.query", + description="Execute SQL query in a workspace", + steps=[ + HintStep( + comment="Submit SQL query", + client=ClientCall( + method="submit_query", + args={"workspace_id": "{workspace_id}", "sql": "{sql}"}, + result_var="query_job", + ), + service=ServiceCall( + service_class="WorkspaceService", + service_module="workspace_service", + method="execute_query", + args={ + "alias": "{project}", + "workspace_id": "{workspace_id}", + "sql": "{sql}", + "transactional": "{transactional}", + }, + ), + ), + HintStep( + comment="Poll until query completes", + client=ClientCall( + method="wait_for_query_job", + args={"query_job_id": 'query_job["id"]'}, + result_var="query_job", + ), + kind="poll_loop", + poll_interval=1.0, + poll_condition='query_job.get("status") not in ("finished", "error")', + ), + ], + notes=[ + "Uses the Query Service API (query.keboola.com).", + "Service layer handles poll + result export in one call.", + ], + ) +) + +# ── workspace from-transformation ────────────────────────────────── + +HintRegistry.register( + CommandHint( + cli_command="workspace.from-transformation", + description="Create a workspace pre-loaded with transformation inputs", + steps=[ + HintStep( + comment="Create workspace from transformation config", + client=ClientCall( + method="get_config_detail", + args={"component_id": "{component_id}", "config_id": "{config_id}"}, + result_var="config", + ), + service=ServiceCall( + service_class="WorkspaceService", + service_module="workspace_service", + method="create_from_transformation", + args={ + "alias": "{project}", + "component_id": "{component_id}", + "config_id": "{config_id}", + "row_id": "{row_id}", + "backend": "{backend}", + }, + ), + ), + ], + notes=[ + "Service layer reads transformation input mapping, creates workspace, " + "and loads input tables automatically.", + ], + ) +) diff --git a/src/keboola_agent_cli/hints/models.py b/src/keboola_agent_cli/hints/models.py new file mode 100644 index 0000000..53c2a0a --- /dev/null +++ b/src/keboola_agent_cli/hints/models.py @@ -0,0 +1,109 @@ +"""Data models for the --hint code generation system. + +Pure dataclasses describing what code to generate for each CLI command. +No external imports — these are consumed by the renderer. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import StrEnum + + +class HintMode(StrEnum): + """Which code layer to generate.""" + + CLIENT = "client" + SERVICE = "service" + + +@dataclass +class ClientCall: + """Describes a KeboolaClient (or ManageClient) method call. + + Used by the renderer to generate direct API usage code. + """ + + method: str + """Client method name, e.g. 'list_components'.""" + + args: dict[str, str] = field(default_factory=dict) + """Mapping of param name -> code expression (inserted literally). + Example: {"branch_id": "123", "component_type": '"extractor"'} + """ + + client_type: str = "storage" + """Which client class: 'storage' (KeboolaClient) or 'manage' (ManageClient).""" + + result_var: str = "result" + """Variable name to assign the result to.""" + + result_hint: str = "" + """Type hint for the result, e.g. 'list[dict]'.""" + + +@dataclass +class ServiceCall: + """Describes a service-layer method call. + + Used by the renderer to generate code that leverages CLI config. + """ + + service_class: str + """Class name, e.g. 'ConfigService'.""" + + service_module: str + """Module name under services/, e.g. 'config_service'.""" + + method: str + """Method name, e.g. 'list_configs'.""" + + args: dict[str, str] = field(default_factory=dict) + """Mapping of param name -> code expression.""" + + +@dataclass +class HintStep: + """One logical step in a command's hint. + + Simple commands have one step. Multi-step commands (e.g. job run --wait) + have multiple steps to show the full flow. + """ + + comment: str + """Human-readable description of what this step does.""" + + client: ClientCall + """How to do this with KeboolaClient.""" + + service: ServiceCall | None = None + """How to do this with the service layer. None if not applicable.""" + + kind: str = "single" + """Step kind: 'single' for one-shot, 'poll_loop' for polling patterns.""" + + poll_interval: float = 5.0 + """Seconds between polls (only for poll_loop kind).""" + + poll_condition: str = "" + """Python expression for poll loop condition, e.g. 'not job.get(\"isFinished\")'.""" + + +@dataclass +class CommandHint: + """Complete hint definition for one CLI command. + + Registered in HintRegistry and looked up by cli_command key. + """ + + cli_command: str + """Dot-separated command path, e.g. 'config.list'.""" + + description: str + """What the command does (used in generated docstring).""" + + steps: list[HintStep] + """Ordered list of steps to render.""" + + notes: list[str] = field(default_factory=list) + """Extra tips or caveats added as comments in generated code.""" diff --git a/src/keboola_agent_cli/hints/renderer.py b/src/keboola_agent_cli/hints/renderer.py new file mode 100644 index 0000000..1e25d91 --- /dev/null +++ b/src/keboola_agent_cli/hints/renderer.py @@ -0,0 +1,361 @@ +"""Code renderers for the --hint system. + +Produces runnable Python code from CommandHint definitions. +Two renderers: ClientRenderer (direct API) and ServiceRenderer (CLI config). +""" + +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from .models import CommandHint + +# Placeholder URL when project alias cannot be resolved +_DEFAULT_STACK_URL = "https://connection.keboola.com" + + +def _escape_for_python_string(value: str) -> str: + """Escape a value for safe embedding inside a double-quoted Python string literal. + + Prevents code injection via crafted parameter values (CWE-94). + """ + return value.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n").replace("\r", "\\r") + + +def _sanitize_for_comment(value: str) -> str: + """Remove characters that could break Python comments or docstrings.""" + return value.replace("\n", " ").replace("\r", " ").replace('"""', "...") + + +def _build_original_command(hint: CommandHint, params: dict[str, Any]) -> str: + """Reconstruct the original CLI command string for the docstring. + + Values are sanitized to prevent docstring/comment injection. + """ + parts = ["kbagent"] + # Convert "config.list" -> "config list" + parts.extend(hint.cli_command.split(".")) + + for key, value in params.items(): + if value is None: + continue + flag = f"--{key.replace('_', '-')}" + if isinstance(value, bool): + if value: + parts.append(flag) + elif isinstance(value, list): + for item in value: + parts.extend([flag, _sanitize_for_comment(str(item))]) + else: + parts.extend([flag, _sanitize_for_comment(str(value))]) + + return " ".join(parts) + + +def _substitute_params(args: dict[str, str], params: dict[str, Any]) -> dict[str, str]: + """Substitute {param} placeholders in hint args with actual CLI values. + + Placeholders like {component_type} are replaced with the actual value + from params. If a param is None, the arg is omitted. + """ + result: dict[str, str] = {} + for key, template in args.items(): + if not isinstance(template, str) or "{" not in template: + result[key] = template + continue + + # Find all {placeholder} references + resolved = template + skip = False + for param_name, param_value in params.items(): + placeholder = f"{{{param_name}}}" + if placeholder in resolved: + if param_value is None: + skip = True + break + if isinstance(param_value, str): + safe = _escape_for_python_string(param_value) + resolved = resolved.replace(placeholder, f'"{safe}"') + elif isinstance(param_value, list): + list_repr = ( + "[" + + ", ".join(f'"{_escape_for_python_string(str(v))}"' for v in param_value) + + "]" + ) + resolved = resolved.replace(placeholder, list_repr) + else: + resolved = resolved.replace(placeholder, str(param_value)) + + if not skip: + result[key] = resolved + + return result + + +def _format_call_args(args: dict[str, str]) -> str: + """Format keyword arguments for a Python method call.""" + if not args: + return "" + parts = [f"{k}={v}" for k, v in args.items()] + # Short calls on one line, long calls multi-line + joined = ", ".join(parts) + if len(joined) <= 60: + return joined + return "\n " + ",\n ".join(parts) + ",\n" + + +class ClientRenderer: + """Renders Python code using KeboolaClient (direct API calls).""" + + @staticmethod + def render( + hint: CommandHint, + params: dict[str, Any], + stack_url: str | None, + branch_id: int | None, + config_dir: Path | None = None, + ) -> str: + """Generate Python code using the client layer.""" + url = stack_url or _DEFAULT_STACK_URL + original_cmd = _build_original_command(hint, params) + lines: list[str] = [] + + # Shebang + docstring (escape to prevent triple-quote injection) + safe_cmd = original_cmd.replace('"""', '"\\"" ') + lines.append("#!/usr/bin/env python3") + lines.append(f'"""Equivalent of: {safe_cmd}"""') + lines.append("") + + # Imports + client_types: set[str] = set() + needs_time = any(step.kind == "poll_loop" for step in hint.steps) + for step in hint.steps: + client_types.add(step.client.client_type) + + needs_os = "storage" in client_types or "manage" in client_types + if needs_os: + lines.append("import os") + if needs_time: + lines.append("import time") + if "mcp" in client_types: + lines.append("from pathlib import Path") + lines.append("") + + if "storage" in client_types: + lines.append("from keboola_agent_cli.client import KeboolaClient") + if "manage" in client_types: + lines.append("from keboola_agent_cli.manage_client import ManageClient") + if "mcp" in client_types: + lines.append("from keboola_agent_cli.config_store import ConfigStore") + lines.append("from keboola_agent_cli.services.mcp_service import McpService") + + lines.append("") + + # Client construction + if "storage" in client_types: + url_comment = "" + if stack_url: + # Find the project alias in params for the comment + project = params.get("project") + if project: + proj_label = project[0] if isinstance(project, list) else project + safe_label = _escape_for_python_string(str(proj_label)) + url_comment = f" # from project '{safe_label}'" + lines.append("client = KeboolaClient(") + lines.append(f' base_url="{url}",{url_comment}') + lines.append(' token=os.environ["KBC_STORAGE_TOKEN"],') + lines.append(")") + + if "manage" in client_types: + lines.append("manage_client = ManageClient(") + lines.append(f' base_url="{url}",') + lines.append(' token=os.environ["KBC_MANAGE_API_TOKEN"],') + lines.append(")") + + if "mcp" in client_types: + config_dir_str = str(config_dir) if config_dir else "/path/to/.kbagent" + lines.append("# MCP tools require ConfigStore (they go through keboola-mcp-server)") + lines.append( + f'mcp_service = McpService(config_store=ConfigStore(config_dir=Path("{config_dir_str}")))' + ) + + # Determine if we need try/finally for cleanup + close_vars = [] + if "storage" in client_types: + close_vars.append("client") + if "manage" in client_types: + close_vars.append("manage_client") + + indent = " " if close_vars else "" + lines.append("") + if close_vars: + lines.append("try:") + + # Steps + for i, step in enumerate(hint.steps): + resolved_args = _substitute_params(step.client.args, params) + + # Inject branch_id if present and method accepts it + if branch_id is not None and "branch_id" not in resolved_args: + if any("{branch" in v for v in step.client.args.values()): + pass # Already handled by substitution + elif "branch_id" in step.client.args: + pass # Already in template + else: + # Add branch_id for methods that typically accept it + resolved_args["branch_id"] = str(branch_id) + + client_var_map = { + "storage": "client", + "manage": "manage_client", + "mcp": "mcp_service", + } + client_var = client_var_map.get(step.client.client_type, "client") + call_args = _format_call_args(resolved_args) + + lines.append(f"{indent}# Step {i + 1}: {step.comment}") + + if step.kind == "poll_loop": + # Generate polling loop + lines.append( + f"{indent}{step.client.result_var} = {client_var}.{step.client.method}({call_args})" + ) + lines.append(f"{indent}while {step.poll_condition}:") + lines.append(f"{indent} time.sleep({step.poll_interval})") + lines.append( + f"{indent} {step.client.result_var} = {client_var}.{step.client.method}({call_args})" + ) + else: + lines.append( + f"{indent}{step.client.result_var} = {client_var}.{step.client.method}({call_args})" + ) + + if i < len(hint.steps) - 1: + lines.append("") + + # Print result + last_var = hint.steps[-1].client.result_var + lines.append("") + lines.append(f"{indent}print({last_var})") + + # Finally (only for clients that need closing) + if close_vars: + lines.append("finally:") + for var in close_vars: + lines.append(f" {var}.close()") + + # Notes + if hint.notes: + lines.append("") + for note in hint.notes: + lines.append(f"# NOTE: {note}") + + return "\n".join(lines) + + +class ServiceRenderer: + """Renders Python code using the service layer (CLI config).""" + + @staticmethod + def render( + hint: CommandHint, + params: dict[str, Any], + stack_url: str | None, + config_dir: Path | None, + branch_id: int | None, + ) -> str: + """Generate Python code using the service layer.""" + original_cmd = _build_original_command(hint, params) + lines: list[str] = [] + + # Check if any step has a service call + has_service = any(step.service is not None for step in hint.steps) + if not has_service: + # Fall back to client renderer with a note + lines.append(f"# No service-layer equivalent for: {original_cmd}") + lines.append("# Use --hint client instead (this command uses direct API calls).") + lines.append("") + return "\n".join(lines) + ClientRenderer.render( + hint, params, stack_url, branch_id, config_dir=config_dir + ) + + # Shebang + docstring (escape to prevent triple-quote injection) + safe_cmd = original_cmd.replace('"""', '"\\"" ') + lines.append("#!/usr/bin/env python3") + lines.append(f'"""Equivalent of: {safe_cmd}"""') + lines.append("") + + # Collect unique service imports + service_imports: dict[str, str] = {} # module -> class + for step in hint.steps: + if step.service: + service_imports[step.service.service_module] = step.service.service_class + + # Imports + lines.append("from pathlib import Path") + lines.append("") + lines.append("from keboola_agent_cli.config_store import ConfigStore") + for module, cls in sorted(service_imports.items()): + lines.append(f"from keboola_agent_cli.services.{module} import {cls}") + + lines.append("") + + # Config store setup + dir_str = str(config_dir) if config_dir else "/path/to/.kbagent" + lines.append("# Path to the directory containing config.json") + lines.append(f'# (same as: kbagent --config-dir "{dir_str}" ...)') + lines.append(f'store = ConfigStore(config_dir=Path("{dir_str}"))') + + # Service construction + created_services: set[str] = set() + for step in hint.steps: + if step.service and step.service.service_class not in created_services: + var_name = _service_var_name(step.service.service_class) + lines.append(f"{var_name} = {step.service.service_class}(config_store=store)") + created_services.add(step.service.service_class) + + lines.append("") + + # Steps + for i, step in enumerate(hint.steps): + if step.service is None: + lines.append(f"# Step {i + 1}: {step.comment}") + lines.append("# (No service-layer equivalent — use client layer for this step)") + continue + + resolved_args = _substitute_params(step.service.args, params) + var_name = _service_var_name(step.service.service_class) + call_args = _format_call_args(resolved_args) + + lines.append(f"# Step {i + 1}: {step.comment}") + result_var = step.client.result_var + lines.append(f"{result_var} = {var_name}.{step.service.method}({call_args})") + + if i < len(hint.steps) - 1: + lines.append("") + + # Print result + last_service_step = next((s for s in reversed(hint.steps) if s.service is not None), None) + if last_service_step: + lines.append("") + lines.append(f"print({last_service_step.client.result_var})") + + # Notes + if hint.notes: + lines.append("") + for note in hint.notes: + lines.append(f"# NOTE: {note}") + + return "\n".join(lines) + + +def _service_var_name(class_name: str) -> str: + """Convert 'ConfigService' -> 'config_service'.""" + # Insert underscore before each uppercase letter (except first), then lowercase + result = [] + for i, char in enumerate(class_name): + if char.isupper() and i > 0: + result.append("_") + result.append(char.lower()) + return "".join(result) diff --git a/tests/test_hints.py b/tests/test_hints.py new file mode 100644 index 0000000..d1111da --- /dev/null +++ b/tests/test_hints.py @@ -0,0 +1,609 @@ +"""Tests for the --hint code generation system.""" + +from pathlib import Path +from unittest.mock import patch + +import pytest +from typer.testing import CliRunner + +from keboola_agent_cli.cli import app +from keboola_agent_cli.hints import HintRegistry, render_hint +from keboola_agent_cli.hints.models import ( + ClientCall, + CommandHint, + HintMode, + HintStep, + ServiceCall, +) +from keboola_agent_cli.hints.renderer import ClientRenderer, ServiceRenderer + +runner = CliRunner() + +STACK_URL = "https://connection.eu-central-1.keboola.com" +CONFIG_DIR = Path("/tmp/test-config") + + +# ── Renderer unit tests ──────────────────────────────────────────── + + +class TestClientRenderer: + """Tests for ClientRenderer code generation.""" + + def test_simple_command_produces_valid_python(self) -> None: + """Generated code must be syntactically valid Python.""" + hint = CommandHint( + cli_command="config.list", + description="List configurations", + steps=[ + HintStep( + comment="List components", + client=ClientCall( + method="list_components", + args={"component_type": "{component_type}"}, + result_var="components", + ), + ), + ], + ) + code = ClientRenderer.render( + hint, + params={"component_type": "extractor", "project": "myproj"}, + stack_url=STACK_URL, + branch_id=None, + ) + # Must compile without errors + compile(code, "", "exec") + + def test_includes_stack_url(self) -> None: + """Generated code includes the resolved stack URL.""" + hint = _simple_hint() + code = ClientRenderer.render(hint, {}, stack_url=STACK_URL, branch_id=None) + assert STACK_URL in code + + def test_uses_placeholder_when_no_stack_url(self) -> None: + """Falls back to placeholder URL when project not resolved.""" + hint = _simple_hint() + code = ClientRenderer.render(hint, {}, stack_url=None, branch_id=None) + assert "https://connection.keboola.com" in code + + def test_never_contains_real_token(self) -> None: + """Token must always be an env var reference, never a real value.""" + hint = _simple_hint() + code = ClientRenderer.render(hint, {}, stack_url=STACK_URL, branch_id=None) + assert "KBC_STORAGE_TOKEN" in code + assert 'os.environ["KBC_STORAGE_TOKEN"]' in code + # No hardcoded token patterns + assert "901-" not in code + + def test_branch_id_injected(self) -> None: + """Branch ID is passed to client methods when provided.""" + hint = CommandHint( + cli_command="config.list", + description="List configs", + steps=[ + HintStep( + comment="List components", + client=ClientCall( + method="list_components", + args={"branch_id": "{branch}"}, + result_var="components", + ), + ), + ], + ) + code = ClientRenderer.render( + hint, params={"branch": 789}, stack_url=STACK_URL, branch_id=789 + ) + assert "branch_id=789" in code + + def test_none_params_omitted(self) -> None: + """Parameters with None values are not included in method calls.""" + hint = CommandHint( + cli_command="config.list", + description="List configs", + steps=[ + HintStep( + comment="List components", + client=ClientCall( + method="list_components", + args={ + "component_type": "{component_type}", + "branch_id": "{branch}", + }, + result_var="components", + ), + ), + ], + ) + code = ClientRenderer.render( + hint, + params={"component_type": None, "branch": None}, + stack_url=STACK_URL, + branch_id=None, + ) + assert "component_type" not in code + assert "branch_id" not in code + + def test_original_command_in_docstring(self) -> None: + """Docstring includes the reconstructed original CLI command.""" + hint = _simple_hint() + code = ClientRenderer.render( + hint, + params={"project": "myproj"}, + stack_url=STACK_URL, + branch_id=None, + ) + assert "kbagent config test --project myproj" in code + + def test_manage_client_import(self) -> None: + """Manage API commands use ManageClient with correct token env var.""" + hint = CommandHint( + cli_command="org.setup", + description="Setup org", + steps=[ + HintStep( + comment="List projects", + client=ClientCall( + method="list_organization_projects", + args={"org_id": "123"}, + client_type="manage", + result_var="projects", + ), + ), + ], + ) + code = ClientRenderer.render(hint, {}, stack_url=STACK_URL, branch_id=None) + assert "ManageClient" in code + assert "KBC_MANAGE_API_TOKEN" in code + assert "KeboolaClient" not in code + + def test_poll_loop_rendering(self) -> None: + """Poll loop steps generate a while loop with sleep.""" + hint = CommandHint( + cli_command="job.run", + description="Run a job and wait", + steps=[ + HintStep( + comment="Create job", + client=ClientCall( + method="create_job", + args={"component_id": '"keboola.ex-http"', "config_id": '"123"'}, + result_var="job", + ), + ), + HintStep( + comment="Poll until job completes", + client=ClientCall( + method="get_job_detail", + args={"job_id": 'str(job["id"])'}, + result_var="job", + ), + kind="poll_loop", + poll_interval=5.0, + poll_condition='not job.get("isFinished")', + ), + ], + ) + code = ClientRenderer.render(hint, {}, stack_url=STACK_URL, branch_id=None) + assert "while not job.get" in code + assert "time.sleep(5.0)" in code + assert "import time" in code + compile(code, "", "exec") + + def test_notes_appended(self) -> None: + """Notes are appended as comments at the end.""" + hint = CommandHint( + cli_command="test.cmd", + description="Test", + steps=[_simple_step()], + notes=["This is a test note."], + ) + code = ClientRenderer.render(hint, {}, stack_url=STACK_URL, branch_id=None) + assert "# NOTE: This is a test note." in code + + +class TestServiceRenderer: + """Tests for ServiceRenderer code generation.""" + + def test_simple_command_produces_valid_python(self) -> None: + """Generated service-layer code must be syntactically valid Python.""" + hint = CommandHint( + cli_command="config.list", + description="List configurations", + steps=[ + HintStep( + comment="List configs", + client=ClientCall( + method="list_components", + result_var="result", + ), + service=ServiceCall( + service_class="ConfigService", + service_module="config_service", + method="list_configs", + args={"aliases": "{project}"}, + ), + ), + ], + ) + code = ServiceRenderer.render( + hint, + params={"project": ["myproj"]}, + stack_url=STACK_URL, + config_dir=CONFIG_DIR, + branch_id=None, + ) + compile(code, "", "exec") + + def test_includes_config_dir(self) -> None: + """Generated code includes the explicit config_dir path.""" + hint = _service_hint() + code = ServiceRenderer.render( + hint, + params={"project": ["myproj"]}, + stack_url=STACK_URL, + config_dir=CONFIG_DIR, + branch_id=None, + ) + assert str(CONFIG_DIR) in code + assert "ConfigStore" in code + assert "config_dir=Path" in code + + def test_falls_back_to_client_when_no_service(self) -> None: + """Commands without service equivalent show client code with a note.""" + hint = CommandHint( + cli_command="branch.create", + description="Create dev branch", + steps=[ + HintStep( + comment="Create branch", + client=ClientCall( + method="create_dev_branch", + args={"name": '"my-branch"'}, + result_var="branch", + ), + service=None, + ), + ], + ) + code = ServiceRenderer.render( + hint, {}, stack_url=STACK_URL, config_dir=CONFIG_DIR, branch_id=None + ) + assert "No service-layer equivalent" in code + assert "KeboolaClient" in code + + def test_never_contains_real_token(self) -> None: + """Service layer code should not contain any token references.""" + hint = _service_hint() + code = ServiceRenderer.render( + hint, + params={"project": ["myproj"]}, + stack_url=STACK_URL, + config_dir=CONFIG_DIR, + branch_id=None, + ) + assert "901-" not in code + + +# ── HintRegistry tests ───────────────────────────────────────────── + + +class TestHintRegistry: + """Tests for the hint registry.""" + + def test_config_hints_registered(self) -> None: + """Config commands should be registered after importing definitions.""" + from keboola_agent_cli.hints import definitions as _ # noqa: F401 + + assert HintRegistry.get("config.list") is not None + assert HintRegistry.get("config.detail") is not None + assert HintRegistry.get("config.search") is not None + + def test_get_nonexistent_returns_none(self) -> None: + """Querying an unregistered command returns None.""" + assert HintRegistry.get("nonexistent.command") is None + + def test_all_commands_returns_sorted(self) -> None: + """all_commands() returns a sorted list.""" + commands = HintRegistry.all_commands() + assert commands == sorted(commands) + assert len(commands) >= 3 # At least config.list/detail/search + + +# ── render_hint integration tests ────────────────────────────────── + + +class TestRenderHint: + """Tests for the public render_hint() function.""" + + def test_client_mode(self) -> None: + """render_hint with CLIENT mode produces KeboolaClient code.""" + code = render_hint( + "config.list", + HintMode.CLIENT, + params={"project": ["myproj"]}, + stack_url=STACK_URL, + config_dir=CONFIG_DIR, + branch_id=None, + ) + assert "KeboolaClient" in code + compile(code, "", "exec") + + def test_service_mode(self) -> None: + """render_hint with SERVICE mode produces ConfigService code.""" + code = render_hint( + "config.list", + HintMode.SERVICE, + params={"project": ["myproj"]}, + stack_url=STACK_URL, + config_dir=CONFIG_DIR, + branch_id=None, + ) + assert "ConfigService" in code + compile(code, "", "exec") + + def test_unknown_command_raises(self) -> None: + """render_hint raises ValueError for unknown commands.""" + with pytest.raises(ValueError, match="No hint available"): + render_hint( + "nonexistent.cmd", + HintMode.CLIENT, + params={}, + stack_url=None, + config_dir=None, + branch_id=None, + ) + + +# ── CLI integration tests ────────────────────────────────────────── + + +class TestHintCLI: + """End-to-end tests via CliRunner.""" + + def test_hint_client_config_list(self, tmp_path: Path) -> None: + """kbagent --hint client config list produces Python code on stdout.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + result = runner.invoke( + app, + ["--config-dir", str(config_dir), "--hint", "client", "config", "list"], + ) + assert result.exit_code == 0 + assert "KeboolaClient" in result.stdout + assert "list_components" in result.stdout + compile(result.stdout, "", "exec") + + def test_hint_service_config_list(self, tmp_path: Path) -> None: + """kbagent --hint service config list produces service layer code.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + result = runner.invoke( + app, + ["--config-dir", str(config_dir), "--hint", "service", "config", "list"], + ) + assert result.exit_code == 0 + assert "ConfigService" in result.stdout + assert "ConfigStore" in result.stdout + + def test_hint_invalid_value(self, tmp_path: Path) -> None: + """kbagent --hint invalid exits with code 2.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + result = runner.invoke( + app, + ["--config-dir", str(config_dir), "--hint", "invalid", "config", "list"], + ) + assert result.exit_code == 2 + + def test_hint_config_detail_with_params(self, tmp_path: Path) -> None: + """Parameters are correctly substituted in hint output.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + result = runner.invoke( + app, + [ + "--config-dir", + str(config_dir), + "--hint", + "client", + "config", + "detail", + "--project", + "myproj", + "--component-id", + "keboola.ex-db-snowflake", + "--config-id", + "42", + ], + ) + assert result.exit_code == 0 + assert "keboola.ex-db-snowflake" in result.stdout + assert '"42"' in result.stdout + assert "get_config_detail" in result.stdout + + def test_hint_does_not_call_api(self, tmp_path: Path) -> None: + """Hint mode must not make any HTTP calls.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + + with patch("keboola_agent_cli.client.KeboolaClient") as MockClient: + result = runner.invoke( + app, + [ + "--config-dir", + str(config_dir), + "--hint", + "client", + "config", + "list", + ], + ) + # No client should have been instantiated + MockClient.assert_not_called() + + assert result.exit_code == 0 + + def test_hint_skips_auto_update(self, tmp_path: Path) -> None: + """Hint mode should not trigger auto-update checks.""" + config_dir = tmp_path / "config" + config_dir.mkdir() + + with patch("keboola_agent_cli.auto_update.maybe_auto_update") as mock_update: + result = runner.invoke( + app, + [ + "--config-dir", + str(config_dir), + "--hint", + "client", + "config", + "list", + ], + ) + mock_update.assert_not_called() + + assert result.exit_code == 0 + + +# ── Security tests ───────────────────────────────────────────────── + + +class TestHintSecurity: + """Security-focused tests for the hint system (CWE-94 prevention).""" + + def test_string_param_with_quotes_is_escaped(self) -> None: + """Quotes in parameter values must be escaped to prevent code injection.""" + hint = CommandHint( + cli_command="test.cmd", + description="Test", + steps=[ + HintStep( + comment="Do something", + client=ClientCall( + method="some_method", + args={"param": "{evil}"}, + result_var="result", + ), + ), + ], + ) + # Attempt injection via double-quote breakout + code = ClientRenderer.render( + hint, + params={"evil": 'value", x=__import__("os").system("id") #'}, + stack_url=STACK_URL, + branch_id=None, + ) + # The quotes must be escaped — generated code must be safe to compile + compile(code, "", "exec") + # Injected code must NOT appear as executable + assert "__import__" not in code or '\\"' in code + + def test_list_param_with_quotes_is_escaped(self) -> None: + """Quotes in list parameter values must be escaped.""" + hint = CommandHint( + cli_command="test.cmd", + description="Test", + steps=[ + HintStep( + comment="Do something", + client=ClientCall( + method="some_method", + args={"items": "{evil_list}"}, + result_var="result", + ), + ), + ], + ) + code = ClientRenderer.render( + hint, + params={"evil_list": ["normal", 'evil"]; import os #']}, + stack_url=STACK_URL, + branch_id=None, + ) + compile(code, "", "exec") + assert "import os" not in code or '\\"' in code + + def test_docstring_injection_prevented(self) -> None: + """Triple quotes in params must not break the docstring.""" + hint = CommandHint( + cli_command="test.cmd", + description="Test", + steps=[ + HintStep( + comment="Do something", + client=ClientCall(method="some_method", result_var="result"), + ), + ], + ) + code = ClientRenderer.render( + hint, + params={"project": '"""\nimport os\nos.system("id")\n"""'}, + stack_url=STACK_URL, + branch_id=None, + ) + compile(code, "", "exec") + + def test_newlines_in_params_escaped(self) -> None: + """Newline characters in params must be escaped.""" + hint = CommandHint( + cli_command="test.cmd", + description="Test", + steps=[ + HintStep( + comment="Do something", + client=ClientCall( + method="some_method", + args={"name": "{name}"}, + result_var="result", + ), + ), + ], + ) + code = ClientRenderer.render( + hint, + params={"name": "line1\nline2\rline3"}, + stack_url=STACK_URL, + branch_id=None, + ) + compile(code, "", "exec") + # Raw newlines must not appear in the string literal + assert "line1\nline2" not in code + + +# ── Test helpers ─────────────────────────────────────────────────── + + +def _simple_step() -> HintStep: + return HintStep( + comment="Do something", + client=ClientCall(method="some_method", result_var="result"), + ) + + +def _simple_hint() -> CommandHint: + return CommandHint( + cli_command="config.test", + description="Test command", + steps=[_simple_step()], + ) + + +def _service_hint() -> CommandHint: + return CommandHint( + cli_command="config.svc", + description="Test service command", + steps=[ + HintStep( + comment="Do something", + client=ClientCall(method="some_method", result_var="result"), + service=ServiceCall( + service_class="ConfigService", + service_module="config_service", + method="list_configs", + args={"aliases": "{project}"}, + ), + ), + ], + )