diff --git a/.claude-plugin/marketplace.json b/.claude-plugin/marketplace.json index 24221e05..74cc3314 100644 --- a/.claude-plugin/marketplace.json +++ b/.claude-plugin/marketplace.json @@ -10,7 +10,7 @@ "plugins": [ { "name": "kbagent", - "version": "0.43.9", + "version": "0.44.0", "source": "./plugins/kbagent", "description": "AI-friendly interface to Keboola Connection projects — explore configs, jobs, lineage, call MCP tools, manage dev branches, and debug SQL in workspaces", "category": "development" diff --git a/CLAUDE.md b/CLAUDE.md index b83def7a..bc41cd99 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -465,6 +465,26 @@ kbagent http delete PATH [--timeout SECONDS] # subprocesses by the scheduler). Use this from inside a scheduled agent # task instead of forking another `kbagent` CLI process tree. +kbagent agent list +kbagent agent show TASK_ID +kbagent agent create --name NAME [--description D] [--cron CRON] [--manual] [--enabled/--disabled] (--type ai_agent --cli CLI --prompt P [--extra-arg ARG ...] [--timeout SECONDS] | --type cli_command --argv ARG [--argv ARG ...] [--timeout SECONDS] | --type mcp_tool --tool TOOL [--mcp-project ALIAS] [--mcp-branch ID] [--input JSON|@file|-] [--timeout SECONDS] | --from-file PATH|@path|-) [--trigger-task-id ID --trigger-on success|error|always] +kbagent agent update TASK_ID [--name N] [--description D] [--cron C] [--enabled/--disabled] [--manual/--auto] [--clear-trigger] [--trigger-task-id ID --trigger-on success|error|always] +kbagent agent delete TASK_ID [--yes] +kbagent agent run TASK_ID [--stream] [--runtime-prompt TEXT | --runtime-input JSON|@file|-] +kbagent agent runs TASK_ID [--limit N] +kbagent agent run-detail TASK_ID RUN_ID +kbagent agent run-events TASK_ID RUN_ID +kbagent agent test (--type ai_agent --cli CLI --prompt P | --type cli_command --argv ARG ... | --type mcp_tool --tool T ... | --from-file PATH) [--name N] [--stream] [--timeout SECONDS] +kbagent agent cron-preview --cron "0 6 * * 1" [--count N] +kbagent agent prompt-improve --goal "..." [--draft "..."] [--cli claude|codex|gemini] [--project ALIAS] [--extra-arg X ...] [--stream/--no-stream] +# `agent` reads/writes /agents.json directly (offline-first, no +# serve required for CRUD + ad-hoc run). The cron loop that fires scheduled +# tasks still requires `kbagent serve` running. Three action flavours +# (ai_agent / cli_command / mcp_tool) mirror the /agents REST surface +# byte-for-byte. Every subcommand taking TASK_ID / RUN_ID accepts it +# positionally OR via flag (--id / --task-id; --run-id for run-detail / +# run-events) -- the flag form matches the rest of the CLI (--job-id, ...). + kbagent kai ping [--project NAME] kbagent kai preflight [--project NAME] kbagent kai ask --message "question" [--project NAME] diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index 59b85cf9..bb0aee3f 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -575,8 +575,8 @@ users safe from accidentally landing on a beta: 1. Bump `pyproject.toml` to the PEP 440 pre-release version (e.g. `0.43.0b1`). 2. Add a changelog entry under that key in `src/keboola_agent_cli/changelog.py`. -3. `make version-sync` propagates the version to `plugin.json` / - `marketplace.json`. +3. `make version-sync` propagates the version to `plugin.json`, + `marketplace.json`, and the `uv.lock` self-version pin. 4. Tag and push: `git tag v0.43.0b1 && git push origin v0.43.0b1`. 5. Create the GitHub release **with the `--prerelease` flag**: ```bash @@ -591,6 +591,25 @@ users safe from accidentally landing on a beta: (`0.43.0`), retag, and create the release **without** `--prerelease` so auto-update picks it up. +**Rebasing a beta onto a moved `main`.** Tags are immutable and pinned to a +commit; rebasing the feature branch (to clear merge conflicts or pull in +newer `main` fixes) leaves the existing `vX.Y.Zb1` tag pointing at the +now-orphaned pre-rebase commit. Do **not** force-move a published tag -- +cut the next pre-release number instead: + +1. Rebase the branch and force-push it (`git push --force-with-lease`). +2. Bump `pyproject.toml` to the next beta (`0.44.0b1` -> `0.44.0b2`), add a + short changelog entry noting "rebased onto current main, no behaviour + change", and `make version-sync`. +3. Commit + push, then tag the rebased HEAD: + `git tag v0.44.0b2 && git push origin v0.44.0b2`. +4. `gh release create v0.44.0b2 --prerelease ...`. Leave the old `b1` + tag/release intact as history -- it documents the earlier base. + +Every published tag stays immutable (a tester who pinned `b1` still gets +exactly what `b1` always was), while `kbagent update --beta` resolves to the +highest PEP 440 version -- the freshly rebased `b2`. + **Users opt in two ways:** - One-shot: `kbagent update --beta` (resolver is told `--prerelease=allow` diff --git a/Makefile b/Makefile index 716b52e4..12c095db 100644 --- a/Makefile +++ b/Makefile @@ -69,13 +69,13 @@ skill-check: ## Check SKILL.md is up-to-date (fails if stale) version-sync: ## Sync version from pyproject.toml to plugin.json uv run python scripts/sync_version.py -version-check: ## Check plugin.json version matches pyproject.toml (fails if mismatched) +version-check: ## Check version-bearing files match pyproject.toml (fails if mismatched) @uv run python scripts/sync_version.py > /dev/null 2>&1 - @if git diff --quiet plugins/kbagent/.claude-plugin/plugin.json; then \ - echo "plugin.json version is in sync"; \ + @if git diff --quiet plugins/kbagent/.claude-plugin/plugin.json .claude-plugin/marketplace.json uv.lock; then \ + echo "version is in sync (plugin.json, marketplace.json, uv.lock)"; \ else \ - echo "ERROR: plugin.json version mismatch. Run 'make version-sync' and commit."; \ - git diff plugins/kbagent/.claude-plugin/plugin.json; \ + echo "ERROR: version mismatch. Run 'make version-sync' and commit."; \ + git diff plugins/kbagent/.claude-plugin/plugin.json .claude-plugin/marketplace.json uv.lock; \ exit 1; \ fi diff --git a/plugins/kbagent/.claude-plugin/plugin.json b/plugins/kbagent/.claude-plugin/plugin.json index 84f10907..3e5fe1bf 100644 --- a/plugins/kbagent/.claude-plugin/plugin.json +++ b/plugins/kbagent/.claude-plugin/plugin.json @@ -1,6 +1,6 @@ { "name": "kbagent", - "version": "0.43.9", + "version": "0.44.0", "description": "AI-friendly interface to Keboola Connection projects — explore configs, jobs, lineage, call MCP tools, manage dev branches, and debug SQL in workspaces", "author": { "name": "Keboola", diff --git a/plugins/kbagent/agents/keboola-expert.md b/plugins/kbagent/agents/keboola-expert.md index 01c84720..a4737125 100644 --- a/plugins/kbagent/agents/keboola-expert.md +++ b/plugins/kbagent/agents/keboola-expert.md @@ -111,6 +111,7 @@ a critical failure. HTTP `?include_sandbox_annotation=true` = 0.43.1+ #312, `kbagent update --beta` = 0.43.3+, `data-app logs` = 0.43.8+, + `kbagent agent ` (CLI parity /agents REST) = 0.44.0+, `storage retype` is a future composite), you MUST refuse the task and return a handoff message to the parent: `"Cannot proceed safely on kbagent . Missing: . @@ -177,7 +178,7 @@ a critical failure. | Rename a project alias | `kbagent project edit --project OLD --new-alias NEW [--dry-run]` (0.31.0+) -- cascades through `config.json` (`projects` key + `default_project`) and the nested-sync directory `//`. Combined with `--url`/`--token` in one call, those mutations target the new alias post-rename. `--dry-run` previews collision detection, planned disk-rename method, and the lineage-cache warning without mutating state. **Lineage cache (if any) is NOT auto-updated**: rebuild via `kbagent lineage build` after the rename | `kbagent project remove` + `kbagent project add` (re-enters the token; loses any nested sync workspace) | hand-editing `~/.config/keboola-agent-cli/config.json` (no validation, easy to miss `default_project` cascade) | | Call the running `kbagent serve` from a scheduled-agent subprocess | `kbagent http get/post/patch/delete ` (0.40.0+) -- uses `KBAGENT_SERVE_URL` + `KBAGENT_SERVE_TOKEN` env vars auto-injected by the scheduler. `kbagent http get /openapi.json` to discover endpoints. Treats the live serve as source-of-truth (no stale local config) | forking `kbagent ` (also fine -- `KBAGENT_CONFIG_DIR` is propagated so the spawned CLI sees the SAME config the serve uses; no more "I'm in the wrong directory" surprises) | `curl $KBAGENT_SERVE_URL/...` by hand (works, but `kbagent http` adds auth header automatically, structured error mapping, and JSON-mode formatting) | | Launch the web UI for an end-user (browser dashboard, no Node BFF) | `kbagent serve --ui [--port PORT] [--ui-dist PATH]` (0.40.0+) -- single-process FastAPI mounts the bundled React SPA at `/`, sets an HttpOnly `kbagent_session` cookie on `GET /` so the browser is auto-authenticated. EventSource SSE works via the same cookie -- no token in URL, JS heap, or access log. Requires the bundled wheel (Node 20+ on the install host) OR `make web-build` from a checkout. CORS origins customisable via `--cors-origin` | `kbagent serve` (plain API) + Vite dev server + Node BFF -- the legacy three-process flow with hot reload, see `web/README.md` "Dev mode" section | inventing a `--token-in-url` flag; running uvicorn directly against `web.frontend.dist` -- the path-rewrite middleware + cookie bootstrap only fire from `kbagent serve --ui` | -| Schedule / manage Agent Tasks (cron, manual, chained) inside `kbagent serve` | `kbagent http /agents...` (0.40.0+) -- list `GET /agents`, create `POST /agents`, update `PATCH /agents/{id}`, run-on-demand `POST /agents/{id}/run`, run history `GET /agents/{id}/runs`, replay events `GET /agents/{id}/runs/{run_id}/events`. Three action flavours: `mcp_tool` / `cli_command` / `ai_agent`. **See [agent-tasks-workflow](../skills/kbagent/references/agent-tasks-workflow.md) for full payload schemas and chained-trigger setup** | Web UI sidebar "Agent Tasks" -- preferred for human authoring; UI calls the same REST endpoints | hand-editing `~/.config/keboola-agent-cli/agents.json` (no schema validation, no scheduler reload, easy to break the cron loop) | +| Schedule / manage Agent Tasks | `kbagent agent ` (0.44.0+) -- CRUD `list/show/create/update/delete`, exec `run [--stream]`, history `runs/run-detail/run-events`, util `test/cron-preview/prompt-improve`. Local-only; cron needs `kbagent serve`. See [agent-tasks-cli-workflow](../skills/kbagent/references/agent-tasks-cli-workflow.md) | `kbagent http /agents...` (0.40.0+) in scheduled subprocesses; Web UI for human authoring | hand-editing `agents.json` | | List models / metrics / entities in a semantic-layer model | `kbagent --json semantic-layer show --project P [--model M] [--type metric\|dataset\|relationship\|constraint\|glossary]` (0.41.0+); `kbagent --json semantic-layer model list --project P` to enumerate models when --model is ambiguous | `kbagent --json tool call get_semantic_layer_*` if the MCP exposes a read tool (none in the kbagent MCP at v0.41.0) | hand-rolled `urllib`/`httpx` loops against `metastore.*.keboola.com` (the `sl-builder` skill's old approach -- bypasses retry/backoff and the kbagent error envelope) | | Validate a semantic-layer model (phantom fields, constraint orphans, AGG-on-STRING) | `kbagent --json semantic-layer validate --project P [--model M] [--deep]` (0.41.0+) -- basic = local structural checks (duplicates, dangling refs, sum-on-pct, constraint orphans, severity-suffix); `--deep` adds parallel Snowflake column-existence probes via the in-process StorageService | hand-coded list+filter Python that re-implements the structural checks (loses the `--deep` Snowflake probe) | running validation by spinning up a workspace and SELECT * FROM every dataset (slow, requires workspace creation, no constraint-orphan detection) | | Snapshot a semantic-layer model to disk (before destructive edits) | `kbagent semantic-layer export --project P [--model M] [--output PATH]` (0.41.0+) -- self-describing JSON, default `./sl_export_{model_name}_{YYYYMMDD_HHMMSS}.json` | `kbagent --json semantic-layer show --project P` and pipe to a file (NOT a clean snapshot -- missing model metadata, no schemaVersion, no round-trip guarantee) | -- | diff --git a/plugins/kbagent/skills/kbagent/SKILL.md b/plugins/kbagent/skills/kbagent/SKILL.md index 1f031518..7acc918a 100644 --- a/plugins/kbagent/skills/kbagent/SKILL.md +++ b/plugins/kbagent/skills/kbagent/SKILL.md @@ -299,6 +299,18 @@ When working inside a git repository or project directory, run `kbagent init` (o | POST to an endpoint on the running kbagent serve | `kbagent http post ` | | PATCH an endpoint on the running kbagent serve | `kbagent http patch ` | | DELETE an endpoint on the running kbagent serve | `kbagent http delete ` | +| List all registered agent tasks | `kbagent agent list` | +| Show one task's full configuration | `kbagent agent show [TASK-ID]` | +| Register a new scheduled task | `kbagent agent create --name NAME` | +| Patch one or more fields on a task. | `kbagent agent update [TASK-ID]` | +| Remove a task. | `kbagent agent delete [TASK-ID]` | +| Trigger a task immediately (does not wait for the next cron firing) | `kbagent agent run [TASK-ID]` | +| Show the run history of a task (most recent first) | `kbagent agent runs [TASK-ID]` | +| Show a single AgentRun record (status, summary, output, error) | `kbagent agent run-detail [TASK-ID] [RUN-ID]` | +| Replay the persisted event timeline of an ai_agent run (line-by-line) | `kbagent agent run-events [TASK-ID] [RUN-ID]` | +| Execute an action ad-hoc (no persistence, no scheduling) | `kbagent agent test` | +| Show the next N firings of a cron expression | `kbagent agent cron-preview --cron CRON` | +| Polish a plain-English goal into an unattended-agent-ready prompt | `kbagent agent prompt-improve --goal GOAL` | ### Sync pull notable flags @@ -339,7 +351,8 @@ For detailed response parsing rules and common pitfalls, see [gotchas](reference | Creating new configurations | [scaffold-workflow](references/scaffold-workflow.md) | | MCP tools (multi-project read/write) | [mcp-workflow](references/mcp-workflow.md) | | Workspace SQL debugging | [workspace-workflow](references/workspace-workflow.md) | -| **Agent Tasks** (schedule AI agents inside `kbagent serve` -- cron / manual / chained; mcp_tool / cli_command / ai_agent action flavours) | [agent-tasks-workflow](references/agent-tasks-workflow.md) | +| **Agent Tasks via CLI** (`kbagent agent` CRUD + run + cron-preview + prompt-improve; cron / manual / chained; mcp_tool / cli_command / ai_agent action flavours) | [agent-tasks-cli-workflow](references/agent-tasks-cli-workflow.md) | +| **Agent Tasks via REST** (`kbagent http /agents...` from inside scheduled subprocesses; SSE streaming) | [agent-tasks-rest-workflow](references/agent-tasks-rest-workflow.md) | | **Data apps** (create / deploy / start / stop / password / delete; the §9 redeploy contract) | [data-app-workflow](references/data-app-workflow.md) | | Storage Files (upload, download, tags, load/unload) | [storage-files-workflow](references/storage-files-workflow.md) | | **Storage column types** (native types, NOT NULL, DEFAULT, branch materialize) | [storage-types-workflow](references/storage-types-workflow.md) | diff --git a/plugins/kbagent/skills/kbagent/references/agent-tasks-cli-workflow.md b/plugins/kbagent/skills/kbagent/references/agent-tasks-cli-workflow.md new file mode 100644 index 00000000..224b2118 --- /dev/null +++ b/plugins/kbagent/skills/kbagent/references/agent-tasks-cli-workflow.md @@ -0,0 +1,220 @@ +# Agent Tasks Workflow (CLI) + +Cron-scheduled, manual, or chained tasks defined locally on disk via the +`kbagent agent` CLI. Same on-disk format the in-process scheduler inside +`kbagent serve` reads, so a CLI-created task fires on its cron as soon as +the server is running. Available since v0.42.0. + +For the **REST API** form (what `kbagent serve` exposes for AI-agent +subprocesses to call back into) see +[agent-tasks-rest-workflow](agent-tasks-rest-workflow.md). Both surfaces +share the AgentTask / AgentAction schema -- pick whichever fits the +caller. + +## When to use which surface + +- **`kbagent agent ...` (this file)** -- authoring from a terminal, CI + pipelines, scripts, or one-shot manual runs. Works offline; the + scheduler still needs `kbagent serve` running for the cron loop, but + CRUD + ad-hoc runs do not. +- **`kbagent http /agents...`** -- from inside a scheduled + `ai_agent` subprocess that wants to call back into its own serve + (env vars `KBAGENT_SERVE_URL` + `KBAGENT_SERVE_TOKEN` are auto-injected). +- **Web UI** (`kbagent serve --ui`, sidebar "Agent Tasks") -- preferred + for human authoring; calls the same REST endpoints. + +## Prerequisites + +```bash +# Install (no extras required for the agent CLI; croniter is core since 0.42.0) +uv tool install 'git+https://github.com/padak/keboola_agent_cli' + +# Optional: start serve so the cron loop actually fires the tasks +# (the CLI commands work without it, but tasks won't auto-run on schedule). +kbagent serve --ui +``` + +## Three action flavours + +Every task carries an `action` envelope with `type` + `params`: + +```jsonc +// type: "mcp_tool" -- call any keboola-mcp-server tool +{ "type": "mcp_tool", + "params": { "tool": "get_jobs", "project": "padak", + "input": {"status": "error", "limit": 50} } } + +// type: "cli_command" -- spawn `kbagent ` subprocess +{ "type": "cli_command", + "params": { "argv": ["job", "list", "--project", "padak", "--status", "error"], + "timeout": 300 } } + +// type: "ai_agent" -- spawn an AI CLI (claude/codex/gemini) with a prompt +{ "type": "ai_agent", + "params": { "cli": "claude", + "prompt": "Summarise last night's failed jobs.", + "extra_args": ["--print"], + "timeout": 600 } } +``` + +`cli` accepts `claude`, `codex`, or `gemini`. The chosen CLI must be on +the server's `PATH` when the task fires (cron or `agent run`). + +## Common workflows + +### Create + inspect + run + +The convenience flags cover the typical single-action case. For complex +payloads, use `--from-file PATH|-` with the full JSON envelope. + +```bash +# (1) Scheduled cli_command -- daily 06:00 health check +kbagent agent create \ + --name "Daily Health Check" \ + --description "Verify project tokens + list error jobs" \ + --cron "0 6 * * *" \ + --type cli_command \ + --argv doctor + +# (2) Manual ai_agent -- ad-hoc storage cleanup advisor +kbagent agent create \ + --name "Storage Cleanup Advisor" \ + --manual \ + --type ai_agent \ + --cli claude \ + --prompt "You are an unattended Storage Cleanup Advisor for project padak. Scan tables not referenced for > 90 days, estimate monthly Snowflake savings ($23/TB), and write a Markdown report grouped by safe-to-delete / candidate-to-archive / verify-first." \ + --timeout 900 + +# (3) From a JSON file (preferred for prompts > a few lines) +kbagent agent create \ + --name "Weekly Triage" \ + --cron "0 8 * * 1" \ + --from-file @task.action.json + +# List + show what was registered +kbagent agent list +kbagent agent show +``` + +> **ID forms (since v0.44.0):** every subcommand that takes a task/run ID +> accepts it positionally (`agent show `) or via a named flag +> (`--id` / `--task-id`, plus `--run-id` for `run-detail` / `run-events`) -- +> matching the rest of the CLI (`--job-id`, `--config-id`, ...). Examples +> below use the positional form for brevity. + +### Run-on-demand (blocking + streaming) + +```bash +# Blocking: print the AgentRun record once the action finishes. +kbagent agent run + +# Live event stream (one line per event in human mode, NDJSON in --json). +kbagent agent run --stream + +# Manual tasks accept ad-hoc runtime input merged with the persisted prompt. +kbagent agent run --runtime-prompt "Focus only on prod-snowflake-etl." + +# Or full JSON merge (mcp_tool / cli_command also supported). +kbagent agent run --runtime-input '{"prompt": "Today only."}' +``` + +### Update + disable + delete + +```bash +# Toggle enabled/disabled (cron loop respects --enabled, --disabled). +kbagent agent update --disabled + +# Flip a cron task to manual (preserves cron for later re-enable). +kbagent agent update --manual + +# Change the cron expression (next_run_at is recomputed automatically). +kbagent agent update --cron "0 5 * * 1-5" + +# Remove a chained downstream trigger. +kbagent agent update --clear-trigger + +# Permanent deletion (run history on disk is preserved). +kbagent agent delete --yes +``` + +### Run history + +```bash +# Most-recent first, default limit 50. +kbagent agent runs + +# A single run record (status, summary, output, error). +kbagent agent run-detail + +# Per-event timeline -- only ai_agent runs from v0.10+ carry one. +kbagent agent run-events +``` + +### Test an action before saving + +```bash +# Same dispatcher as a real run, but nothing is written to agents.json. +kbagent agent test --type cli_command --argv project --argv list + +# With live streaming for ai_agent previews. +kbagent agent test --type ai_agent --cli claude --prompt "Hello" --stream +``` + +### Cron preview + prompt helper + +```bash +# Validate a cron expression and see the next 5 firings (UTC). +kbagent agent cron-preview --cron "0 6 * * 1" --count 5 + +# Polish a plain-English goal into a polished single-shot prompt. +kbagent agent prompt-improve \ + --goal "Audit last week's failed jobs and post a Slack-friendly summary" \ + --cli claude --project padak --stream +``` + +## Chained triggers + +`--trigger-task-id` chains a downstream task. The downstream's +subprocess inherits `KBAGENT_UPSTREAM_TASK_ID` + `KBAGENT_UPSTREAM_RUN_ID` +env vars so an `ai_agent` downstream can read the upstream run via +`kbagent http get /agents/$UP/runs/$RUN` (only available when the +downstream runs inside `kbagent serve`). + +```bash +# Repair runs only when the triage upstream succeeds. +kbagent agent create \ + --name "Failed Jobs Triage" \ + --cron "0 8 * * *" \ + --type ai_agent --cli claude --prompt "..." \ + --trigger-task-id \ + --trigger-on success +``` + +`--trigger-on` accepts `success` (default), `error`, or `always`. + +## Output modes + +- **Human** -- Rich tables, panels, syntax-highlighted JSON for action + payloads. +- **`--json`** -- canonical envelope (`{"status": "ok", "data": ...}`) + for scripted use. Errors land as `{"status": "error", "error": + {"code": ..., "message": ...}}` with stable codes (`NOT_FOUND`, + `CONFIG_ERROR`, `VALIDATION_ERROR`, `MISSING_PARAMETER`). +- **`--stream`** -- per-event output, one line per event. NDJSON in + `--json` mode (each event is a self-contained JSON object). + +## Lifecycle gotchas + +- **The cron loop only runs inside `kbagent serve`.** The CLI commands + read/write the same `agents.json`, but cron-firing requires the live + scheduler. Tasks created via CLI sit dormant until serve starts. +- **`agent run` does not consult cron.** It dispatches immediately and + fans out the chained downstream (if any) regardless of cron schedule. +- **One run per task at a time.** While a task is running, a second + `agent run` errors out (the runner refuses concurrent dispatch). +- **`--runtime-prompt` is ai_agent only.** For cli_command use + `--runtime-input '{"argv": ["--extra", "flag"]}'`; for mcp_tool use + `--runtime-input '{"key": "value"}'`. Merge semantics match the REST + endpoint exactly (`merge_runtime_input` is the shared helper). +- **Cron expressions are UTC.** No per-task timezone override (yet); + use `cron-preview` to translate. diff --git a/plugins/kbagent/skills/kbagent/references/agent-tasks-workflow.md b/plugins/kbagent/skills/kbagent/references/agent-tasks-rest-workflow.md similarity index 100% rename from plugins/kbagent/skills/kbagent/references/agent-tasks-workflow.md rename to plugins/kbagent/skills/kbagent/references/agent-tasks-rest-workflow.md diff --git a/plugins/kbagent/skills/kbagent/references/commands-reference.md b/plugins/kbagent/skills/kbagent/references/commands-reference.md index 91adf783..95b99ae4 100644 --- a/plugins/kbagent/skills/kbagent/references/commands-reference.md +++ b/plugins/kbagent/skills/kbagent/references/commands-reference.md @@ -237,6 +237,25 @@ Manage Keboola metastore models -- datasets, metrics, relationships, constraints Reads `KBAGENT_SERVE_URL` + `KBAGENT_SERVE_TOKEN` env vars. The scheduler auto-injects these (plus `KBAGENT_CONFIG_DIR`) into every AI-agent / `cli_command` subprocess. Outside a serve subprocess context the command refuses to run with exit code 2. **Inside a scheduled-agent task, prefer `kbagent http get /openapi.json` then a typed call over forking another `kbagent` CLI -- the HTTP path always sees the operator's live config (not the global `~/.config/keboola-agent-cli/` one).** +## Agent Tasks (since v0.44.0) + +CLI parity for the `/agents` REST surface. Reads/writes `/agents.json` -- the same on-disk format the in-process scheduler inside `kbagent serve` consumes. CRUD + ad-hoc `run` work offline; the cron loop that fires scheduled tasks still requires `kbagent serve` running. See [agent-tasks-cli-workflow.md](agent-tasks-cli-workflow.md) for full walkthroughs; [agent-tasks-rest-workflow.md](agent-tasks-rest-workflow.md) covers the REST/SSE form for AI-agent subprocesses. + +**ID forms (since v0.44.0):** every subcommand that takes `TASK_ID` / `RUN_ID` accepts it positionally (`agent show TASK_ID`) **or** via a named flag (`--id` / `--task-id`, plus `--run-id` for `run-detail` / `run-events`). The flag aliases bring agent commands in line with the rest of the CLI, which identifies entities by flag everywhere else (`--job-id`, `--config-id`, `--app-id`, ...). Passing both forms with conflicting values is a usage error (exit 2). + +- `agent list` -- list all registered tasks (id / name / cron / type / state / last-run / next-run). +- `agent show TASK_ID` -- full task detail including the action payload. +- `agent create --name N [--description D] [--cron CRON] [--manual] [--enabled/--disabled] (--type ai_agent --cli claude|codex|gemini --prompt P [--extra-arg ...] [--timeout SECONDS] | --type cli_command --argv ARG [--argv ARG ...] [--timeout SECONDS] | --type mcp_tool --tool TOOL [--mcp-project ALIAS] [--mcp-branch ID] [--input JSON|@file|-] [--timeout SECONDS] | --from-file PATH|@path|-) [--trigger-task-id ID --trigger-on success|error|always]` -- persist a new task. Convenience flags cover the typical single-action case; `--from-file` accepts the full `{"type": ..., "params": ...}` JSON envelope. +- `agent update TASK_ID [--name N] [--description D] [--cron C] [--enabled/--disabled] [--manual/--auto] [--clear-trigger] [--trigger-task-id ID --trigger-on ...]` -- patch one or more fields. Omitted flags leave the field unchanged. `--manual` nulls `next_run_at`; `--auto` recomputes it from the cron expression. +- `agent delete TASK_ID [--yes]` -- permanent removal. Run history on disk is preserved. +- `agent run TASK_ID [--stream] [--runtime-prompt TEXT | --runtime-input JSON|@file|-]` -- trigger immediately. `--stream` prints one line per event in human mode, NDJSON in `--json` mode. `--runtime-prompt` appends ad-hoc text to an ai_agent's persisted prompt for this run only; `--runtime-input` merges arbitrary JSON into the action params (mcp_tool: shallow-merge into `params.input`; cli_command: appends to `params.argv`). +- `agent runs TASK_ID [--limit N]` -- run history (newest first; default limit 50). +- `agent run-detail TASK_ID RUN_ID` -- single AgentRun record (status / summary / output / error). +- `agent run-events TASK_ID RUN_ID` -- replay the persisted ai_agent event timeline (only present for ai_agent runs from v0.10+). +- `agent test [--type ... | --from-file PATH] [--stream] [--name N] [common action flags]` -- execute an action ad-hoc; nothing is persisted. Same dispatcher as the cron scheduler, useful for sanity-checking a prompt / tool / argv before saving. +- `agent cron-preview --cron "..." [--count N]` -- validate a cron expression and show the next N firings (UTC, capped at 20). +- `agent prompt-improve --goal "..." [--draft "..."] [--cli claude|codex|gemini] [--project ALIAS] [--extra-arg X ...] [--stream/--no-stream]` -- AI-polished single-shot prompt for an unattended agent task. The final `done` event's `data.prompt` carries the cleaned body ready to drop into `agent create --prompt ...`. + ## Utility - `init [--from-global]` -- create local `.kbagent/` workspace (per-directory isolation) - `doctor [--fix]` -- health checks; `--fix` auto-installs MCP server binary diff --git a/plugins/kbagent/skills/kbagent/references/gotchas.md b/plugins/kbagent/skills/kbagent/references/gotchas.md index d8fae544..2184b93b 100644 --- a/plugins/kbagent/skills/kbagent/references/gotchas.md +++ b/plugins/kbagent/skills/kbagent/references/gotchas.md @@ -1994,3 +1994,38 @@ requires `canManageTokens` privilege, which only **master tokens** carry. styling and escapes the interpolated `app_id` / `project_alias` values via `rich.markup.escape` per the `commands/config.py` precedent. + +## `kbagent agent` CRUD works offline; cron firing needs `kbagent serve` running (since v0.44.0) + +The `kbagent agent ` command tree reads/writes `/agents.json` +directly via `AgentService`, so `agent list / show / create / update / delete / +run / runs / cron-preview / test / prompt-improve` all work without +`kbagent serve` running. But the cron loop that fires scheduled tasks lives +**inside** the FastAPI lifespan -- tasks created via CLI sit dormant until a +serve instance picks up the file on its next minute-tick. + +- **Symptom**: `kbagent agent create --cron "*/5 * * * *" --type ...` returns + the persisted task with `next_run_at` populated, but the task never + actually runs. `agent runs ` stays empty hour after hour. +- **Cause**: no `kbagent serve` process is running, so the cron loop never + ticks. The task is correctly persisted; the scheduler just isn't alive. +- **Fix**: start `kbagent serve` (with or without `--ui`) on the machine that + owns the config directory. The scheduler picks up the new task on the + next tick and fires it on the cron's schedule. +- **Diagnostic**: `kbagent agent run ` works offline (it dispatches via + the in-process runner without consulting cron). If that runs the action + successfully but cron-driven runs never happen, the scheduler is offline. + +This is the SAME on-disk format the REST router writes via `POST /agents`, +so CLI-created tasks are interchangeable with UI-created and REST-created +tasks -- the only difference is who fires them. + +## `agent` action helpers are shared between REST + CLI (since v0.44.0) + +The `validate_trigger` (cycle/self-loop check) and `merge_runtime_input` +(per-action-type runtime input merge) helpers live in +`keboola_agent_cli.server.agents_store` so the REST router and the CLI +`AgentService` share the exact same boundary behaviour. If you add a +new action type, update **both** the runner dispatcher (`agent_runner.py`) +**and** `merge_runtime_input` to keep CRUD parity. Tests in +`tests/test_agent_service.py` + `tests/test_agent_cli.py` catch drift. diff --git a/pyproject.toml b/pyproject.toml index 38671d0c..b94e8782 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "keboola-agent-cli" -version = "0.43.9" +version = "0.44.0" description = "AI-friendly CLI for managing Keboola projects" readme = "README.md" requires-python = ">=3.12" @@ -20,6 +20,7 @@ dependencies = [ "packaging>=23", "prompt-toolkit>=3.0", "kai-client>=0.11.0", + "croniter>=2.0", ] [project.optional-dependencies] @@ -28,7 +29,6 @@ server = [ "uvicorn[standard]>=0.30", "sse-starlette>=2.1", "python-multipart>=0.0.9", - "croniter>=2.0", ] [project.scripts] diff --git a/scripts/post-edit-quality.sh b/scripts/post-edit-quality.sh index 0215a2d9..c74a5380 100755 --- a/scripts/post-edit-quality.sh +++ b/scripts/post-edit-quality.sh @@ -54,10 +54,16 @@ fi uv run --quiet ruff format --quiet "$FILE_PATH" || true -# ty: report but do NOT fail the hook -- type checker is in warning-only mode -# while we migrate. Switch this to FAILED=1 when the codebase is clean. -if ! uv run --quiet ty check "$FILE_PATH" 2>&1 | tail -5; then - : # warnings only +# ty: runs on every edited .py file but does NOT fail the hook -- the type +# checker is in warning-only mode while the codebase carries pre-existing +# diagnostics (~100 across src/; clearing that backlog is tracked as a separate +# task). Running it here still surfaces NEW type regressions in the edited file +# immediately. Flip the warning-only block below to `FAILED=1` once the backlog +# is clean, so type errors start blocking edits. +TY_OUTPUT="$(uv run --quiet ty check "$FILE_PATH" 2>&1 || true)" +if printf '%s\n' "$TY_OUTPUT" | grep -qE '^(error|warning)'; then + echo "post-edit: ty (warning-only) flagged $FILE_PATH -- review, edit not blocked:" >&2 + printf '%s\n' "$TY_OUTPUT" | grep -E '^(error|warning)' | head -10 >&2 fi exit $FAILED diff --git a/scripts/sync_version.py b/scripts/sync_version.py index 9bdcbd4e..2266d659 100644 --- a/scripts/sync_version.py +++ b/scripts/sync_version.py @@ -1,15 +1,19 @@ #!/usr/bin/env python3 -"""Sync version across the three version-bearing files. +"""Sync version across the version-bearing files. -Single source of truth: ``pyproject.toml``. The plugin manifest and the -marketplace catalogue are kept in lock-step so Claude Code sees the -same version string everywhere. +Single source of truth: ``pyproject.toml``. The plugin manifest, the +marketplace catalogue, and the lockfile's own package pin are kept in +lock-step so Claude Code -- and uv -- see the same version string +everywhere. Files kept in sync: - ``pyproject.toml`` (source) - ``plugins/kbagent/.claude-plugin/plugin.json`` (plugin manifest) - ``.claude-plugin/marketplace.json`` -> ``plugins[*].version`` (per-plugin entry the marketplace descriptor exposes to Claude Code) +- ``uv.lock`` -> the ``keboola-agent-cli`` package's own ``version`` pin + (uv records the workspace package version; a pyproject bump leaves it + stale until ``uv lock`` reruns, which ``version-check`` then flags) The marketplace's own top-level ``version`` key is NOT touched here: it is the version of the marketplace *descriptor shape*, bumped only when @@ -33,10 +37,14 @@ PYPROJECT = REPO_ROOT / "pyproject.toml" PLUGIN_JSON = REPO_ROOT / "plugins" / "kbagent" / ".claude-plugin" / "plugin.json" MARKETPLACE_JSON = REPO_ROOT / ".claude-plugin" / "marketplace.json" +UV_LOCK = REPO_ROOT / "uv.lock" # Plugin name inside marketplace.json whose version must track the CLI/plugin. KBAGENT_PLUGIN_NAME = "kbagent" +# Distribution name of this project as it appears in uv.lock's [[package]] table. +KBAGENT_DIST_NAME = "keboola-agent-cli" + def get_pyproject_version() -> str: """Extract version from pyproject.toml.""" @@ -112,6 +120,34 @@ def sync_marketplace_json(version: str) -> bool: return True +def sync_uv_lock(version: str) -> bool: + """Update this project's own ``version`` pin in uv.lock. Returns True if changed. + + uv.lock records the workspace package's version in its ``[[package]]`` + table; a ``pyproject.toml`` bump leaves it stale until ``uv lock`` + reruns. We patch just that one entry with a scoped regex (the + distribution name is unique in the lockfile) so the lock stays + consistent without shelling out to uv. A full ``uv lock`` is still the + way to refresh *dependencies*; this only keeps the self-version honest. + """ + if not UV_LOCK.is_file(): + return False + text = UV_LOCK.read_text(encoding="utf-8") + # Match the version line that immediately follows our package's name line. + pattern = re.compile(r'(name = "' + re.escape(KBAGENT_DIST_NAME) + r'"\nversion = ")[^"]+(")') + new_text, count = pattern.subn(rf"\g<1>{version}\g<2>", text) + if count == 0: + print( + f"WARN: uv.lock has no '{KBAGENT_DIST_NAME}' package entry; skipping lock sync.", + file=sys.stderr, + ) + return False + if new_text == text: + return False + UV_LOCK.write_text(new_text, encoding="utf-8") + return True + + def main() -> None: version = get_pyproject_version() @@ -127,6 +163,12 @@ def main() -> None: else: print(f"marketplace.json plugins[{KBAGENT_PLUGIN_NAME}].version already at {version}") + uv_lock_changed = sync_uv_lock(version) + if uv_lock_changed: + print(f"Updated uv.lock {KBAGENT_DIST_NAME} version to {version}") + else: + print(f"uv.lock {KBAGENT_DIST_NAME} version already at {version}") + if __name__ == "__main__": main() diff --git a/src/keboola_agent_cli/changelog.py b/src/keboola_agent_cli/changelog.py index e162c740..c67ba0e3 100644 --- a/src/keboola_agent_cli/changelog.py +++ b/src/keboola_agent_cli/changelog.py @@ -8,6 +8,9 @@ # Ordered newest-first. Each value is a list of brief one-line descriptions. CHANGELOG: dict[str, list[str]] = { + "0.44.0": [ + 'New: `kbagent agent ` -- full CLI parity for the `/agents` REST surface that `kbagent serve` has exposed since v0.40.0. Twelve subcommands: `list`, `show`, `create`, `update`, `delete`, `run [--stream] [--runtime-prompt | --runtime-input]`, `runs`, `run-detail`, `run-events`, `test [--stream]`, `cron-preview --cron "..." [--count N]`, `prompt-improve --goal "..." [--cli claude|codex|gemini] [--stream/--no-stream]`. Pure-local: `AgentService` reads/writes `/agents.json` directly, so CRUD + ad-hoc `run` work offline; the cron loop that fires scheduled tasks still requires `kbagent serve` running, but the on-disk format is identical -- a CLI-created task fires on its cron as soon as the server boots and reads the file. Three action flavours mirror the REST endpoint byte-for-byte (`ai_agent` / `cli_command` / `mcp_tool`); `--from-file PATH|@path|-` takes the full `{type, params}` JSON envelope, and convenience flags (`--cli + --prompt + --extra-arg`, `--argv` repeatable, `--tool + --input + --mcp-project + --mcp-branch`) cover the common single-action case. `--runtime-prompt` (ai_agent-only) appends ad-hoc text to the persisted prompt for one run; `--runtime-input` merges arbitrary JSON into action params. Streaming variants render events line-by-line in human mode and NDJSON under `--json`. Trigger chaining via `--trigger-task-id ID --trigger-on success|error|always` with the same cycle/self-loop validation the REST router applies. Every subcommand accepts its `TASK_ID` / `RUN_ID` positionally OR via `--id` / `--task-id` (and `--run-id` for `run-detail` / `run-events`), matching the rest of the CLI (`--job-id`, `--config-id`, ...). Boundary helpers `validate_trigger` + `merge_runtime_input` were extracted from `routers/agents.py` into `server/agents_store.py` so the REST router and `AgentService` share the exact same behaviour byte-for-byte; a new blocking `POST /agents/prompt/improve` mirrors the SSE variant for scripted callers. `croniter` moved from `[server]` extras to core dependencies (cron-preview validation runs outside serve); `server/__init__.py` was split into a PEP 562 lazy shim + new `server/app.py` so importing `from keboola_agent_cli.server.agents_store import AgentStore` no longer drags FastAPI/uvicorn into the CLI path -- the agent CLI works on plain installs without `[server]` extras. The permission registry adds 12 entries (`list / show / runs / run-detail / run-events / cron-preview` = read, `create / update / run / test / prompt-improve` = write, `delete` = destructive); hint definitions ship an intentionally-empty `agent.py` (agent CRUD is pure-local, with no Keboola HTTP API to mimic). Tooling hardening: `make version-sync` now also pins the `keboola-agent-cli` self-version in `uv.lock`, and `version-check` guards `plugin.json` + `marketplace.json` + `uv.lock` together. Tests: 23 unit (`test_agent_service.py`) + 29 CLI (`test_agent_cli.py`, every subcommand in human + `--json`, positional + `--id`/`--task-id`/`--run-id` alias + conflict/missing-id paths) + 3 E2E (`tests/test_e2e.py::TestE2EAgentTasks`). Closes the gap between the React UI sidebar "Agent Tasks" (live since v0.40.0) and CLI users who previously had to fall back to `kbagent http /agents...` from inside scheduled subprocesses.', + ], "0.43.9": [ "Fix: `kbagent data-app list` leaked workspace/sandbox deployments into the data-app listing. The Data Science `GET /apps` collection returns EVERY deployment in the project -- not just data apps but also interactive Snowflake/BigQuery workspaces (`componentId=keboola.sandboxes`, `type=snowflake`/`bigquery`, no name, a `*.snowflakecomputing.com` URL). `list_data_apps` merged the whole collection, so a project with sandboxes showed phantom unnamed `(snowflake)` rows that do NOT appear in the Apps UI (which filters to `keboola.data-apps`). **Fix:** `list_data_apps` now skips any deployment whose `componentId` is present and not `keboola.data-apps`; an item that omits `componentId` (older API shape) is kept rather than hidden, so we never drop a row we cannot classify. The list envelope gains a `component_id` field per app for transparency. Verified live against a project with 4 sandboxes + 1 data app: the listing went from 5 rows to the single real data app, matching the UI. Tests: `tests/test_data_app_plain_env_keys.py::TestListSandboxFilter`.", 'Fix: `kbagent data-app secrets-get` and `secrets-remove` refused any key without a leading `#`, even though `secrets-list` enumerated those keys. The `parameters.dataApp.secrets` block legitimately holds BOTH `#`-prefixed encrypted secrets (value = `KBC::ProjectSecure*::...` ciphertext) AND plain unencrypted env-var config values (e.g. `ADMIN_EMAILS`, `SMTP_HOST`). `list_data_app_secrets` had no key validation and listed all of them; `get_data_app_secret`/`remove_data_app_secrets` ran `_validate_secret_key`, which enforced `SECRET_KEY_PATTERN = ^#[A-Za-z][A-Za-z0-9_-]{0,63}$` (mandatory `#`), so a plain key like `ADMIN_EMAILS` failed with *"Invalid secret key ... Keys must start with \'#\'"* -- listable but neither readable nor removable. **Fix:** `_validate_secret_key` gained a `require_hash` parameter (default `True`); a new `SECRET_OR_PLAIN_KEY_PATTERN = ^#?[A-Za-z][A-Za-z0-9_-]{0,63}$` (optional `#`) is used by the read/remove paths (`require_hash=False`), while `secrets-set` keeps `require_hash=True` because it encrypts and `#` carries meaning. `secrets-get` now dispatches on whether the stored value is a `KBC::` ciphertext: for an ENCRYPTED secret it stays metadata-only (`encrypted: true`, `value: null`, `fingerprint`/`encryption_prefix` -- the Encryption API has no decrypt endpoint, so the decrypted plaintext is still NEVER exposed), and for a PLAIN value it returns the literal value (`encrypted: false`, empty `fingerprint`/`encryption_prefix`) since that value is already stored in clear and visible via `config detail`. Lookup remains exact-match (no `#KEY`<->`KEY` fuzzing), so behaviour for existing `#` keys is unchanged. JSON envelope gains `encrypted` (bool) and `value` (string | null); human mode prints `fingerprint=... prefix=...` for encrypted keys and `value (plaintext, unencrypted): ...` for plain keys (with a stderr note that the value is unencrypted). Shape note for downstream consumers: `fingerprint`/`encryption_prefix` are now ALWAYS present but are EMPTY strings for plain keys (they used to be a reliable non-empty proxy for "is encrypted") -- the new `encrypted` bool is the canonical discriminator; `value` is `null` for encrypted keys and a string for plain keys. `secrets-set` is intentionally NOT changed -- adding a plain env var is `config update`, not `secrets-set`. Sync surfaces touched: `services/data_app_service.py`, `commands/data_app.py`, `commands/context.py` AGENT_CONTEXT, `CLAUDE.md ## All CLI Commands`, `plugins/kbagent/skills/kbagent/references/commands-reference.md`, `data-app-workflow.md`, and `gotchas.md` (new `(since v0.43.9)` entry; the existing "secrets-get NEVER echoes decrypted plaintext" gotcha clarified to scope it to encrypted values). The `data-app.secrets-get` hint description (`hints/definitions/data_app.py`) and the `secrets-remove` empty-keys error message were also updated to drop the now-inaccurate "always metadata-only" / "#KEY required" wording. Tests: service-layer coverage for get-plain (`encrypted=false` + value), get-encrypted (`encrypted=true`, `value=None`, metadata preserved), remove-plain, and continued rejection of malformed keys; CLI coverage for the plain-value human + JSON output; an E2E step in `test_e2e.py::test_data_app_secrets_round_trip` injects a plain key via `config update`, reads it back, and removes it (and the stale `removed == 1`/`== 0` assertions there were corrected -- `removed` is a list of env-var names, not a count).', diff --git a/src/keboola_agent_cli/cli.py b/src/keboola_agent_cli/cli.py index f522452f..74a3d825 100644 --- a/src/keboola_agent_cli/cli.py +++ b/src/keboola_agent_cli/cli.py @@ -6,6 +6,7 @@ import typer +from .commands.agent import agent_app from .commands.branch import branch_app from .commands.changelog import changelog_command from .commands.component import component_app @@ -40,6 +41,7 @@ from .models import PermissionPolicy from .output import OutputFormatter from .permissions import PermissionEngine +from .services.agent_service import AgentService from .services.branch_service import BranchService from .services.component_service import ComponentService from .services.config_service import ConfigService @@ -122,6 +124,7 @@ app.add_typer(semantic_layer_app, name="semantic-layer", rich_help_panel=_DEV) app.add_typer(semantic_layer_app, name="sl", rich_help_panel=_DEV, hidden=True) app.add_typer(http_app, name="http", rich_help_panel=_DEV) +app.add_typer(agent_app, name="agent", rich_help_panel=_DEV) def apply_firewall_flags( @@ -340,6 +343,7 @@ def main( doctor_service = DoctorService(config_store=config_store, mcp_service=mcp_service) version_service = VersionService() http_forwarder_service = HttpForwarderService() + agent_service = AgentService(config_store=config_store, mcp_service=mcp_service) try: config = config_store.load() @@ -399,6 +403,7 @@ def main( ctx.obj["doctor_service"] = doctor_service ctx.obj["version_service"] = version_service ctx.obj["http_forwarder_service"] = http_forwarder_service + ctx.obj["agent_service"] = agent_service # Warn if empty local config shadows global with projects (#104) if source == "local" and not json_output and ctx.invoked_subcommand != "init": diff --git a/src/keboola_agent_cli/commands/agent.py b/src/keboola_agent_cli/commands/agent.py new file mode 100644 index 00000000..c8ad6867 --- /dev/null +++ b/src/keboola_agent_cli/commands/agent.py @@ -0,0 +1,968 @@ +"""Agent task commands -- CLI parity for the `/agents` REST surface. + +Mirrors what ``kbagent serve --ui`` exposes: CRUD over scheduled tasks, +ad-hoc runs (blocking + streaming), run history, cron preview, and an +AI-assisted prompt helper. Reads/writes the same ``agents.json`` the +server scheduler uses, so a CLI-created task fires on cron as soon as +``kbagent serve`` is running. + +Thin layer: each command parses arguments, calls AgentService, formats +output. No business logic lives here. Async service methods are bridged +through ``asyncio.run`` at the command boundary so Typer stays sync. +""" + +from __future__ import annotations + +import asyncio +import json +import sys +from collections.abc import AsyncIterator +from pathlib import Path +from typing import Any + +import typer + +from ..errors import ConfigError, ErrorCode +from ..output import OutputFormatter +from ..server.agents_store import AgentAction, Trigger +from ..services.agent_service import AgentService +from ._helpers import check_cli_permission, get_formatter, get_service + +agent_app = typer.Typer(help="Scheduled agent tasks (cron / manual / chained)") + + +@agent_app.callback(invoke_without_command=True) +def _agent_permission_check(ctx: typer.Context) -> None: + check_cli_permission(ctx, "agent") + + +# ── Shared parsing helpers ──────────────────────────────────────────── + + +def _read_payload(value: str, formatter: OutputFormatter) -> str: + """Resolve --input style strings: inline JSON, ``@file``, or ``-`` (stdin).""" + if value == "-": + return sys.stdin.read() + if value.startswith("@"): + path = Path(value[1:]) + if not path.is_file(): + formatter.error( + message=f"Input file not found: {path}", + error_code=ErrorCode.INVALID_ARGUMENT, + ) + raise typer.Exit(code=2) from None + return path.read_text(encoding="utf-8") + return value + + +def _parse_json(value: str, formatter: OutputFormatter, *, label: str) -> dict[str, Any]: + """Decode a JSON string into a dict, exiting cleanly on bad JSON.""" + try: + parsed = json.loads(value) + except json.JSONDecodeError as exc: + formatter.error( + message=f"Invalid JSON in {label}: {exc}", + error_code=ErrorCode.INVALID_FORMAT, + ) + raise typer.Exit(code=2) from None + if not isinstance(parsed, dict): + formatter.error( + message=f"{label} must be a JSON object", + error_code=ErrorCode.INVALID_FORMAT, + ) + raise typer.Exit(code=2) from None + return parsed + + +def _action_from_flags( + formatter: OutputFormatter, + *, + action_type: str | None, + from_file: str | None, + cli: str | None, + prompt: str | None, + extra_arg: list[str], + argv: list[str], + tool: str | None, + mcp_project: str | None, + mcp_branch: int | None, + input_payload: str | None, + timeout: int | None, +) -> AgentAction: + """Build an AgentAction from CLI flags or ``--from-file PATH|-``. + + ``--from-file`` wins outright -- the file is expected to contain the + full ``{type, params}`` envelope, mirroring the REST POST body. The + convenience flags are for the common "single ai_agent" / "single + cli_command" / "single mcp_tool" cases where typing a JSON file would + be overkill. + """ + if from_file is not None: + raw = _read_payload(from_file, formatter) + payload = _parse_json(raw, formatter, label="--from-file") + try: + return AgentAction.model_validate(payload) + except Exception as exc: + formatter.error( + message=f"Invalid action payload: {exc}", + error_code=ErrorCode.VALIDATION_ERROR, + ) + raise typer.Exit(code=2) from None + + if action_type is None: + formatter.error( + message="--type is required (one of: ai_agent, cli_command, mcp_tool) " + "or pass --from-file PATH|- with a full {type, params} JSON.", + error_code=ErrorCode.MISSING_PARAMETER, + ) + raise typer.Exit(code=2) from None + + if action_type == "ai_agent": + if not cli or not prompt: + formatter.error( + message="ai_agent action requires --cli {claude|codex|gemini} and --prompt TEXT.", + error_code=ErrorCode.MISSING_PARAMETER, + ) + raise typer.Exit(code=2) from None + params: dict[str, Any] = {"cli": cli, "prompt": prompt} + if extra_arg: + params["extra_args"] = list(extra_arg) + if timeout is not None: + params["timeout"] = timeout + return AgentAction(type="ai_agent", params=params) + + if action_type == "cli_command": + if not argv: + formatter.error( + message="cli_command action requires --argv ARG (repeatable) -- e.g. " + "--argv job --argv list --argv --project=padak.", + error_code=ErrorCode.MISSING_PARAMETER, + ) + raise typer.Exit(code=2) from None + params: dict[str, Any] = {"argv": list(argv)} + if timeout is not None: + params["timeout"] = timeout + return AgentAction(type="cli_command", params=params) + + if action_type == "mcp_tool": + if not tool: + formatter.error( + message="mcp_tool action requires --tool NAME.", + error_code=ErrorCode.MISSING_PARAMETER, + ) + raise typer.Exit(code=2) from None + params: dict[str, Any] = {"tool": tool} + if mcp_project: + params["project"] = mcp_project + if mcp_branch is not None: + params["branch_id"] = mcp_branch + if input_payload: + raw = _read_payload(input_payload, formatter) + params["input"] = _parse_json(raw, formatter, label="--input") + return AgentAction(type="mcp_tool", params=params) + + formatter.error( + message=f"Unknown --type {action_type!r}; expected ai_agent|cli_command|mcp_tool.", + error_code=ErrorCode.INVALID_ARGUMENT, + ) + raise typer.Exit(code=2) from None + + +def _trigger_from_flags( + trigger_task_id: str | None, + trigger_on: str, +) -> Trigger | None: + """Build a Trigger from --trigger-task-id / --trigger-on flags.""" + if not trigger_task_id: + return None + # ty: trigger_on is a plain str here; Trigger.on is a Literal[success|error|always]. + # The value is constrained to that set by the CLI/REST boundary before reaching here. + return Trigger(on=trigger_on, task_id=trigger_task_id) # ty: ignore[invalid-argument-type] + + +def _resolve_id( + formatter: OutputFormatter, + positional: str | None, + option: str | None, + *, + label: str, + flag: str, +) -> str: + """Resolve an ID passed either positionally or via a named flag. + + Every agent subcommand accepts its task/run ID both ways: positionally + (``agent show TASK_ID``) for terse interactive use, and via a named flag + (``--id`` / ``--task-id`` / ``--run-id``) for consistency with the rest of + the CLI, which identifies entities by flag everywhere else (``--job-id``, + ``--config-id``, ``--app-id``, ...). Exactly one must be supplied; passing + both with conflicting values is a usage error rather than a silent pick. + """ + if positional is not None and option is not None and positional != option: + formatter.error( + message=f"{label} given both positionally ({positional!r}) and via {flag} " + f"({option!r}) -- pass it only one way.", + error_code=ErrorCode.INVALID_ARGUMENT, + ) + raise typer.Exit(code=2) from None + resolved = positional if positional is not None else option + if not resolved: + formatter.error( + message=f"{label} is required: pass it positionally or via {flag}.", + error_code=ErrorCode.MISSING_PARAMETER, + ) + raise typer.Exit(code=2) from None + return resolved + + +# ── Output renderers ────────────────────────────────────────────────── + + +def _render_tasks_table(console: Any, data: dict[str, Any]) -> None: + """Plain-text table of tasks: id / name / cron / state.""" + tasks = data.get("tasks") or [] + if not tasks: + console.print("[dim]No agent tasks registered.[/dim]") + return + from rich.table import Table + + table = Table(title="Agent Tasks", show_lines=False) + table.add_column("ID", style="cyan") + table.add_column("Name") + table.add_column("Schedule") + table.add_column("Type") + table.add_column("State") + table.add_column("Last run", style="dim") + table.add_column("Next run", style="dim") + for task in tasks: + state_bits = [] + if not task.get("enabled", True): + state_bits.append("[yellow]disabled[/yellow]") + if task.get("manual"): + state_bits.append("[blue]manual[/blue]") + elif task.get("enabled", True): + state_bits.append("[green]enabled[/green]") + state = " ".join(state_bits) or "-" + action = task.get("action") or {} + table.add_row( + str(task.get("id", "")), + task.get("name", ""), + "" if task.get("manual") else task.get("cron", "") or "-", + action.get("type", "?"), + state, + task.get("last_run_at") or "-", + task.get("next_run_at") or "-", + ) + console.print(table) + + +def _render_task_detail(console: Any, task: dict[str, Any]) -> None: + """Pretty single-task panel with the action payload pretty-printed.""" + from rich.panel import Panel + from rich.syntax import Syntax + + header_lines = [ + f"[cyan]{task.get('id', '')}[/cyan] [bold]{task.get('name', '')}[/bold]", + ] + if task.get("description"): + header_lines.append(task["description"]) + state_bits = [] + if not task.get("enabled", True): + state_bits.append("[yellow]disabled[/yellow]") + if task.get("manual"): + state_bits.append("[blue]manual[/blue]") + else: + state_bits.append(f"cron=[green]{task.get('cron', '')}[/green]") + header_lines.append("State: " + " ".join(state_bits)) + if task.get("last_run_at"): + header_lines.append(f"Last run: [dim]{task['last_run_at']}[/dim]") + if task.get("next_run_at"): + header_lines.append(f"Next run: [dim]{task['next_run_at']}[/dim]") + if task.get("trigger"): + trig = task["trigger"] + header_lines.append( + f"Trigger: on=[magenta]{trig.get('on')}[/magenta] -> {trig.get('task_id')}" + ) + action_json = json.dumps(task.get("action") or {}, indent=2, ensure_ascii=False) + console.print(Panel("\n".join(header_lines), title="Task", border_style="blue")) + console.print(Syntax(action_json, "json", theme="ansi_dark", word_wrap=True)) + + +def _render_created_task(console: Any, task: dict[str, Any]) -> None: + """Confirmation line + full detail panel after `agent create`.""" + console.print(f"[bold green]Created[/bold green] task [cyan]{task['id']}[/cyan]") + _render_task_detail(console, task) + + +def _render_updated_task(console: Any, task: dict[str, Any]) -> None: + """Confirmation line + full detail panel after `agent update`.""" + console.print(f"[bold green]Updated[/bold green] task [cyan]{task['id']}[/cyan]") + _render_task_detail(console, task) + + +def _render_deleted_task(console: Any, data: dict[str, Any]) -> None: + """Confirmation line after `agent delete`.""" + console.print(f"[bold green]Deleted[/bold green] task [cyan]{data['id']}[/cyan]") + + +def _render_run_result(console: Any, run: dict[str, Any]) -> None: + """One-line run status + timing after `agent run`.""" + status = run.get("status") + status_styled = f"[green]{status}[/green]" if status == "ok" else f"[red]{status}[/red]" + console.print(f"[bold]Run[/bold] [cyan]{run['run_id']}[/cyan] status={status_styled}") + console.print(f"started: [dim]{run['started_at']}[/dim]") + console.print(f"ended: [dim]{run.get('ended_at') or '-'}[/dim]") + if run.get("error"): + console.print(f"[red]error:[/red] {run['error']}") + + +def _render_test_result(console: Any, run: dict[str, Any]) -> None: + """Ad-hoc `agent test` preview: status + optional output / error.""" + console.print(f"[bold]Preview[/bold] status={run.get('status')}") + if run.get("output"): + console.print(json.dumps(run["output"], indent=2, ensure_ascii=False)) + if run.get("error"): + console.print(f"[red]error:[/red] {run['error']}") + + +def _render_improved_prompt(console: Any, data: dict[str, Any]) -> None: + """`agent prompt-improve` result: header + the cleaned prompt body.""" + console.print(f"[bold]Cleaned prompt[/bold] (status={data.get('status', '?')}):") + console.print(data.get("prompt") or "[dim][/dim]") + + +def _render_runs_table(console: Any, data: dict[str, Any]) -> None: + runs = data.get("runs") or [] + if not runs: + console.print("[dim]No run history.[/dim]") + return + from rich.table import Table + + table = Table(title="Run history", show_lines=False) + table.add_column("Run ID", style="cyan") + table.add_column("Started", style="dim") + table.add_column("Ended", style="dim") + table.add_column("Status") + table.add_column("Summary") + for run in runs: + status = run.get("status", "?") + status_styled = ( + f"[green]{status}[/green]" + if status == "ok" + else f"[red]{status}[/red]" + if status == "error" + else f"[yellow]{status}[/yellow]" + ) + summary = "" + if run.get("error"): + summary = f"[red]{run['error'][:80]}[/red]" + elif run.get("summary"): + s = run["summary"] + bits = [] + if s.get("model"): + bits.append(str(s["model"])) + if s.get("tokens"): + bits.append(f"tokens={s['tokens']}") + if s.get("cost_usd") is not None: + bits.append(f"${s['cost_usd']:.4f}") + summary = " ".join(bits) + table.add_row( + run.get("run_id", ""), + run.get("started_at") or "", + run.get("ended_at") or "-", + status_styled, + summary, + ) + console.print(table) + + +def _render_stream_event(formatter: OutputFormatter, evt: dict[str, Any]) -> None: + """Single event renderer for ``--stream`` mode. + + JSON: one NDJSON line per event (the same shape SSE consumers see). + Human: short labeled line per event; the ``done`` event gets a + multi-line summary with exit code, elapsed time, response preview. + """ + if formatter.json_mode: + sys.stdout.write(json.dumps(evt, ensure_ascii=False) + "\n") + sys.stdout.flush() + return + event = evt.get("event", "?") + data = evt.get("data") or {} + console = formatter.console + if event == "init": + info_bits = ["[bold blue]Init[/bold blue]"] + for key in ("name", "task_id", "action_type", "cli"): + if data.get(key): + info_bits.append(f"{key}=[cyan]{data[key]}[/cyan]") + console.print(" ".join(info_bits)) + return + if event == "stdout": + # claude JSONL: try to extract the most useful field + if isinstance(data, dict) and "type" in data and "raw" not in data: + kind = data.get("type") + if kind == "assistant": + content = data.get("message", {}).get("content") or [] + texts = [ + c.get("text", "") + for c in content + if isinstance(c, dict) and c.get("type") == "text" + ] + if texts: + console.print(f"[bold]assistant:[/bold] {' '.join(texts)}") + else: + tool_uses = [ + c.get("name") + for c in content + if isinstance(c, dict) and c.get("type") == "tool_use" + ] + if tool_uses: + console.print( + f"[magenta]tool_use:[/magenta] {', '.join(filter(None, tool_uses))}" + ) + else: + console.print("[dim]event=assistant (no text)[/dim]") + elif kind == "result": + # final result line carries the summary text + pass + elif kind: + console.print(f"[dim]event={kind}[/dim]") + return + # codex/gemini raw lines or unparseable json + raw = data.get("raw") if isinstance(data, dict) else None + if raw: + console.print(raw) + return + if event == "stderr": + raw = data.get("raw") if isinstance(data, dict) else str(data) + console.print(f"[dim red]{raw}[/dim red]") + return + if event == "done": + status = data.get("status", "?") + status_styled = ( + "[green]ok[/green]" + if status == "ok" + else "[red]error[/red]" + if status == "error" + else f"[yellow]{status}[/yellow]" + ) + bits = [f"[bold]Done[/bold] status={status_styled}"] + if data.get("exit_code") is not None: + bits.append(f"exit_code={data['exit_code']}") + if data.get("elapsed_seconds") is not None: + bits.append(f"elapsed={data['elapsed_seconds']}s") + console.print(" ".join(bits)) + if data.get("error"): + console.print(f"[red]error:[/red] {data['error']}") + if data.get("response"): + preview = str(data["response"])[:500] + console.print(f"[dim]response preview:[/dim] {preview}") + return + # Unknown event type; dump as-is + console.print(f"[dim]{event}:[/dim] {json.dumps(data, ensure_ascii=False)[:200]}") + + +def _stream_to_stdout(formatter: OutputFormatter, agen: AsyncIterator[dict[str, Any]]) -> None: + """Drive an async event generator from sync code and render each event.""" + + async def _drive() -> None: + async for evt in agen: + _render_stream_event(formatter, evt) + + try: + asyncio.run(_drive()) + except KeyboardInterrupt: + # The async generator's finally block kills any spawned subprocess. + # Print a short marker so the user knows the partial output is theirs. + if not formatter.json_mode: + formatter.console.print("[yellow]Interrupted by user.[/yellow]") + raise typer.Exit(code=130) from None + + +# ── Commands ─────────────────────────────────────────────────────────── + + +@agent_app.command("list") +def agent_list(ctx: typer.Context) -> None: + """List all registered agent tasks.""" + formatter = get_formatter(ctx) + service: AgentService = get_service(ctx, "agent_service") + try: + tasks = service.list_tasks() + except ConfigError as exc: + formatter.error(message=exc.message, error_code=ErrorCode.CONFIG_ERROR) + raise typer.Exit(code=5) from None + payload = {"tasks": [t.model_dump(mode="json") for t in tasks]} + formatter.output(payload, _render_tasks_table) + + +@agent_app.command("show") +def agent_show( + ctx: typer.Context, + task_id: str | None = typer.Argument(None, help="Task ID (12-char hex). Or use --id."), + task_id_opt: str | None = typer.Option( + None, "--id", "--task-id", help="Task ID (alias for the positional argument)." + ), +) -> None: + """Show one task's full configuration.""" + formatter = get_formatter(ctx) + service: AgentService = get_service(ctx, "agent_service") + task_id = _resolve_id(formatter, task_id, task_id_opt, label="Task ID", flag="--id/--task-id") + try: + task = service.get_task(task_id) + except ConfigError as exc: + formatter.error(message=exc.message, error_code=ErrorCode.NOT_FOUND) + raise typer.Exit(code=1) from None + formatter.output(task.model_dump(mode="json"), _render_task_detail) + + +@agent_app.command("create") +def agent_create( + ctx: typer.Context, + name: str = typer.Option(..., "--name", help="Human-readable task name"), + description: str = typer.Option("", "--description", help="Free-form description"), + cron: str = typer.Option("0 * * * *", "--cron", help="Cron expression (UTC)"), + manual: bool = typer.Option( + False, + "--manual", + help="Skip cron firing -- only run when triggered manually or as downstream.", + ), + enabled: bool = typer.Option(True, "--enabled/--disabled", help="Initial enabled state."), + action_type: str | None = typer.Option( + None, + "--type", + help="Action type when not using --from-file: ai_agent|cli_command|mcp_tool", + ), + from_file: str | None = typer.Option( + None, + "--from-file", + help='Full action JSON ({"type": "...", "params": {...}}). PATH, @path, or - for stdin.', + ), + cli: str | None = typer.Option(None, "--cli", help="ai_agent: claude|codex|gemini"), + prompt: str | None = typer.Option(None, "--prompt", help="ai_agent: prompt body"), + extra_arg: list[str] = typer.Option( + [], + "--extra-arg", + help="ai_agent: extra CLI arg (repeatable). Forwarded to claude/codex/gemini.", + ), + argv: list[str] = typer.Option( + [], + "--argv", + help="cli_command: argv element (repeatable). 'kbagent' prefix is auto-added.", + ), + tool: str | None = typer.Option(None, "--tool", help="mcp_tool: tool name (e.g. get_jobs)"), + mcp_project: str | None = typer.Option( + None, "--mcp-project", help="mcp_tool: project alias to dispatch into." + ), + mcp_branch: int | None = typer.Option( + None, "--mcp-branch", help="mcp_tool: branch ID (optional)." + ), + input_payload: str | None = typer.Option( + None, + "--input", + help="mcp_tool: JSON input. Inline, @path, or -.", + ), + timeout: int | None = typer.Option(None, "--timeout", help="Action timeout in seconds."), + trigger_task_id: str | None = typer.Option( + None, "--trigger-task-id", help="Chain: ID of downstream task to fire after this one." + ), + trigger_on: str = typer.Option( + "success", + "--trigger-on", + help="Chain filter: success|error|always.", + ), +) -> None: + """Register a new scheduled task. + + Two ways to specify the action: + 1. ``--from-file PATH|-`` with the full {type, params} JSON envelope. + 2. ``--type TYPE`` plus the type-specific flags (ai_agent: --cli/--prompt, + cli_command: --argv ..., mcp_tool: --tool/--input/--mcp-project). + """ + formatter = get_formatter(ctx) + service: AgentService = get_service(ctx, "agent_service") + action = _action_from_flags( + formatter, + action_type=action_type, + from_file=from_file, + cli=cli, + prompt=prompt, + extra_arg=list(extra_arg), + argv=list(argv), + tool=tool, + mcp_project=mcp_project, + mcp_branch=mcp_branch, + input_payload=input_payload, + timeout=timeout, + ) + trigger = _trigger_from_flags(trigger_task_id, trigger_on) + try: + task = service.create_task( + name=name, + action=action, + description=description, + cron=cron, + manual=manual, + enabled=enabled, + trigger=trigger, + ) + except ConfigError as exc: + formatter.error(message=exc.message, error_code=ErrorCode.CONFIG_ERROR) + raise typer.Exit(code=5) from None + formatter.output(task.model_dump(mode="json"), _render_created_task) + + +@agent_app.command("update") +def agent_update( + ctx: typer.Context, + task_id: str | None = typer.Argument(None, help="Task ID to update. Or use --id."), + task_id_opt: str | None = typer.Option( + None, "--id", "--task-id", help="Task ID (alias for the positional argument)." + ), + name: str | None = typer.Option(None, "--name"), + description: str | None = typer.Option(None, "--description"), + cron: str | None = typer.Option(None, "--cron"), + enabled: bool | None = typer.Option( + None, "--enabled/--disabled", help="Toggle scheduler firing." + ), + manual: bool | None = typer.Option( + None, + "--manual/--auto", + help="--manual disables cron loop; --auto re-enables it (and recomputes next_run_at).", + ), + clear_trigger: bool = typer.Option( + False, "--clear-trigger", help="Remove any chained downstream trigger." + ), + trigger_task_id: str | None = typer.Option( + None, "--trigger-task-id", help="Set/replace the downstream chain target." + ), + trigger_on: str = typer.Option( + "success", "--trigger-on", help="Chain filter when --trigger-task-id is set." + ), +) -> None: + """Patch one or more fields on a task. Omitted flags leave the field unchanged.""" + formatter = get_formatter(ctx) + service: AgentService = get_service(ctx, "agent_service") + task_id = _resolve_id(formatter, task_id, task_id_opt, label="Task ID", flag="--id/--task-id") + trigger = _trigger_from_flags(trigger_task_id, trigger_on) + try: + task = service.update_task( + task_id, + name=name, + description=description, + cron=cron, + manual=manual, + enabled=enabled, + trigger=trigger, + clear_trigger=clear_trigger, + ) + except ConfigError as exc: + formatter.error(message=exc.message, error_code=ErrorCode.CONFIG_ERROR) + raise typer.Exit(code=5) from None + formatter.output(task.model_dump(mode="json"), _render_updated_task) + + +@agent_app.command("delete") +def agent_delete( + ctx: typer.Context, + task_id: str | None = typer.Argument(None, help="Task ID to delete. Or use --id."), + task_id_opt: str | None = typer.Option( + None, "--id", "--task-id", help="Task ID (alias for the positional argument)." + ), + yes: bool = typer.Option(False, "--yes", "-y", help="Skip confirmation prompt."), +) -> None: + """Remove a task. Run history on disk is preserved. + + Disabled tasks should be deleted with this command (not just disabled), + when they will never run again -- a long history of disabled tasks + clutters the UI / list output. + """ + formatter = get_formatter(ctx) + service: AgentService = get_service(ctx, "agent_service") + task_id = _resolve_id(formatter, task_id, task_id_opt, label="Task ID", flag="--id/--task-id") + if not yes and not formatter.json_mode: + confirm = typer.confirm(f"Delete task '{task_id}'?") + if not confirm: + formatter.console.print("[yellow]Aborted.[/yellow]") + raise typer.Exit(code=0) + try: + service.delete_task(task_id) + except ConfigError as exc: + formatter.error(message=exc.message, error_code=ErrorCode.NOT_FOUND) + raise typer.Exit(code=1) from None + formatter.output({"status": "deleted", "id": task_id}, _render_deleted_task) + + +@agent_app.command("run") +def agent_run( + ctx: typer.Context, + task_id: str | None = typer.Argument(None, help="Task ID to run. Or use --id."), + task_id_opt: str | None = typer.Option( + None, "--id", "--task-id", help="Task ID (alias for the positional argument)." + ), + stream: bool = typer.Option( + False, + "--stream", + help="Stream events live (each event on its own line / NDJSON in --json mode).", + ), + runtime_prompt: str | None = typer.Option( + None, + "--runtime-prompt", + help="ai_agent: ad-hoc text appended to the persisted prompt for this run only.", + ), + runtime_input: str | None = typer.Option( + None, + "--runtime-input", + help="JSON input merged into the action params for this run only. Inline, @path, or -.", + ), +) -> None: + """Trigger a task immediately (does not wait for the next cron firing). + + By default blocks until the run finishes and prints the AgentRun + record. Use ``--stream`` to render live events as they arrive (one + line per event in human mode, NDJSON in --json mode). + + ``--runtime-prompt`` is a shortcut for ai_agent tasks (most common + use case for manual tasks). For full control over the runtime merge, + pass ``--runtime-input '{"key": "value"}'``. + """ + formatter = get_formatter(ctx) + service: AgentService = get_service(ctx, "agent_service") + task_id = _resolve_id(formatter, task_id, task_id_opt, label="Task ID", flag="--id/--task-id") + runtime: dict[str, Any] | None = None + if runtime_input: + raw = _read_payload(runtime_input, formatter) + runtime = _parse_json(raw, formatter, label="--runtime-input") + if runtime_prompt: + runtime = dict(runtime or {}) + runtime["prompt"] = runtime_prompt + try: + if stream: + _stream_to_stdout(formatter, service.stream_run(task_id, runtime_input=runtime)) + return + run = asyncio.run(service.run_task(task_id, runtime_input=runtime)) + except ConfigError as exc: + formatter.error(message=exc.message, error_code=ErrorCode.NOT_FOUND) + raise typer.Exit(code=1) from None + formatter.output(run.model_dump(mode="json"), _render_run_result) + + +@agent_app.command("runs") +def agent_runs( + ctx: typer.Context, + task_id: str | None = typer.Argument(None, help="Task ID whose history to show. Or use --id."), + task_id_opt: str | None = typer.Option( + None, "--id", "--task-id", help="Task ID (alias for the positional argument)." + ), + limit: int = typer.Option(50, "--limit", help="Max rows to return."), +) -> None: + """Show the run history of a task (most recent first).""" + formatter = get_formatter(ctx) + service: AgentService = get_service(ctx, "agent_service") + task_id = _resolve_id(formatter, task_id, task_id_opt, label="Task ID", flag="--id/--task-id") + try: + runs = service.list_runs(task_id, limit=limit) + except ConfigError as exc: + formatter.error(message=exc.message, error_code=ErrorCode.NOT_FOUND) + raise typer.Exit(code=1) from None + payload = {"runs": [r.model_dump(mode="json") for r in runs]} + formatter.output(payload, _render_runs_table) + + +@agent_app.command("run-detail") +def agent_run_detail( + ctx: typer.Context, + task_id: str | None = typer.Argument(None, help="Task ID. Or use --id/--task-id."), + run_id: str | None = typer.Argument(None, help="Run ID (12-char hex). Or use --run-id."), + task_id_opt: str | None = typer.Option( + None, "--id", "--task-id", help="Task ID (alias for the positional argument)." + ), + run_id_opt: str | None = typer.Option( + None, "--run-id", help="Run ID (alias for the positional argument)." + ), +) -> None: + """Show a single AgentRun record (status, summary, output, error).""" + formatter = get_formatter(ctx) + service: AgentService = get_service(ctx, "agent_service") + task_id = _resolve_id(formatter, task_id, task_id_opt, label="Task ID", flag="--id/--task-id") + run_id = _resolve_id(formatter, run_id, run_id_opt, label="Run ID", flag="--run-id") + try: + run = service.get_run(task_id, run_id) + except ConfigError as exc: + formatter.error(message=exc.message, error_code=ErrorCode.NOT_FOUND) + raise typer.Exit(code=1) from None + + def _render(c: Any, d: dict[str, Any]) -> None: + from rich.syntax import Syntax + + c.print(f"[bold]Run[/bold] [cyan]{d['run_id']}[/cyan] status={d.get('status', '?')}") + c.print(f"task_id: {d.get('task_id')}") + c.print(f"started: [dim]{d.get('started_at')}[/dim]") + c.print(f"ended: [dim]{d.get('ended_at') or '-'}[/dim]") + if d.get("error"): + c.print(f"[red]error:[/red] {d['error']}") + if d.get("summary"): + c.print(Syntax(json.dumps(d["summary"], indent=2), "json", theme="ansi_dark")) + if d.get("output"): + c.print("[bold]output:[/bold]") + c.print(Syntax(json.dumps(d["output"], indent=2), "json", theme="ansi_dark")) + + formatter.output(run.model_dump(mode="json"), _render) + + +@agent_app.command("run-events") +def agent_run_events( + ctx: typer.Context, + task_id: str | None = typer.Argument(None, help="Task ID. Or use --id/--task-id."), + run_id: str | None = typer.Argument(None, help="Run ID. Or use --run-id."), + task_id_opt: str | None = typer.Option( + None, "--id", "--task-id", help="Task ID (alias for the positional argument)." + ), + run_id_opt: str | None = typer.Option( + None, "--run-id", help="Run ID (alias for the positional argument)." + ), +) -> None: + """Replay the persisted event timeline of an ai_agent run (line-by-line).""" + formatter = get_formatter(ctx) + service: AgentService = get_service(ctx, "agent_service") + task_id = _resolve_id(formatter, task_id, task_id_opt, label="Task ID", flag="--id/--task-id") + run_id = _resolve_id(formatter, run_id, run_id_opt, label="Run ID", flag="--run-id") + try: + events = service.get_run_events(task_id, run_id) + except ConfigError as exc: + formatter.error(message=exc.message, error_code=ErrorCode.NOT_FOUND) + raise typer.Exit(code=1) from None + if formatter.json_mode: + formatter.output({"events": events, "count": len(events)}) + return + for evt in events: + _render_stream_event(formatter, evt) + + +@agent_app.command("test") +def agent_test( + ctx: typer.Context, + name: str = typer.Option("[preview]", "--name", help="Name shown in event init payload."), + stream: bool = typer.Option( + False, "--stream", help="Stream events live instead of returning the final run record." + ), + action_type: str | None = typer.Option(None, "--type", help="ai_agent|cli_command|mcp_tool"), + from_file: str | None = typer.Option(None, "--from-file", help="Action JSON (or @path / -)."), + cli: str | None = typer.Option(None, "--cli"), + prompt: str | None = typer.Option(None, "--prompt"), + extra_arg: list[str] = typer.Option([], "--extra-arg"), + argv: list[str] = typer.Option([], "--argv"), + tool: str | None = typer.Option(None, "--tool"), + mcp_project: str | None = typer.Option(None, "--mcp-project"), + mcp_branch: int | None = typer.Option(None, "--mcp-branch"), + input_payload: str | None = typer.Option(None, "--input"), + timeout: int | None = typer.Option(None, "--timeout"), +) -> None: + """Execute an action ad-hoc (no persistence, no scheduling). + + Exact dispatch logic as the cron scheduler -- useful for sanity-checking + a prompt / tool / cli_command before saving a task. + """ + formatter = get_formatter(ctx) + service: AgentService = get_service(ctx, "agent_service") + action = _action_from_flags( + formatter, + action_type=action_type, + from_file=from_file, + cli=cli, + prompt=prompt, + extra_arg=list(extra_arg), + argv=list(argv), + tool=tool, + mcp_project=mcp_project, + mcp_branch=mcp_branch, + input_payload=input_payload, + timeout=timeout, + ) + if stream: + _stream_to_stdout(formatter, service.stream_test_action(action, name=name)) + return + run = asyncio.run(service.test_action(action, name=name)) + formatter.output(run.model_dump(mode="json"), _render_test_result) + + +@agent_app.command("cron-preview") +def agent_cron_preview( + ctx: typer.Context, + cron: str = typer.Option(..., "--cron", help="Cron expression to evaluate."), + count: int = typer.Option(5, "--count", help="How many firings to return (1-20)."), +) -> None: + """Show the next N firings of a cron expression. + + Useful when authoring a task: paste the cron, eyeball the next few + times, then save. Works offline (no network calls). + """ + formatter = get_formatter(ctx) + service: AgentService = get_service(ctx, "agent_service") + try: + firings = service.cron_preview(cron, count=count) + except ConfigError as exc: + formatter.error(message=exc.message, error_code=ErrorCode.VALIDATION_ERROR) + raise typer.Exit(code=2) from None + + def _render(c: Any, d: dict[str, Any]) -> None: + c.print(f"[bold]Cron[/bold] [cyan]{d['cron']}[/cyan]") + for ts in d["firings"]: + c.print(f" - [dim]{ts}[/dim]") + + formatter.output({"cron": cron, "firings": firings}, _render) + + +@agent_app.command("prompt-improve") +def agent_prompt_improve( + ctx: typer.Context, + goal: str = typer.Option(..., "--goal", help="Plain-English goal to polish."), + draft: str = typer.Option("", "--draft", help="Optional half-baked prompt to refine."), + cli: str = typer.Option("claude", "--cli", help="AI CLI to invoke: claude|codex|gemini."), + project: str | None = typer.Option(None, "--project", help="Pinned project alias hint."), + extra_arg: list[str] = typer.Option([], "--extra-arg", help="Extra args for the AI CLI."), + stream: bool = typer.Option( + True, "--stream/--no-stream", help="Stream events as they arrive (default: on)." + ), +) -> None: + """Polish a plain-English goal into an unattended-agent-ready prompt. + + Spawns the chosen AI CLI exactly the way an ai_agent run does, with a + meta-prompt that asks for a single polished prompt body. The final + ``done`` event carries the cleaned prompt under ``data.prompt``. + """ + formatter = get_formatter(ctx) + service: AgentService = get_service(ctx, "agent_service") + + async def _drive_no_stream() -> dict[str, Any] | None: + last_done: dict[str, Any] | None = None + async for evt in service.improve_prompt( + cli=cli, goal=goal, draft=draft, project=project, extra_args=list(extra_arg) + ): + if evt.get("event") == "done": + last_done = evt.get("data") + return last_done + + try: + if stream: + _stream_to_stdout( + formatter, + service.improve_prompt( + cli=cli, + goal=goal, + draft=draft, + project=project, + extra_args=list(extra_arg), + ), + ) + return + final = asyncio.run(_drive_no_stream()) + except ConfigError as exc: + formatter.error(message=exc.message, error_code=ErrorCode.VALIDATION_ERROR) + raise typer.Exit(code=2) from None + if final is None: + formatter.error( + message="Prompt helper produced no output.", + error_code=ErrorCode.UNKNOWN_ERROR, + ) + raise typer.Exit(code=1) from None + formatter.output(final, _render_improved_prompt) diff --git a/src/keboola_agent_cli/commands/context.py b/src/keboola_agent_cli/commands/context.py index cf113b4f..a7689545 100644 --- a/src/keboola_agent_cli/commands/context.py +++ b/src/keboola_agent_cli/commands/context.py @@ -902,6 +902,82 @@ the operator configured (not the global ~/.config one). Outside a serve subprocess context the command refuses to run. +### Agent Tasks (CLI parity with the `/agents` REST surface) + + Reads/writes /agents.json -- the same on-disk format the + cron loop inside `kbagent serve` consumes. CLI CRUD + ad-hoc `run` + work offline; cron firing still requires the live server. + + Every subcommand below that takes TASK_ID / RUN_ID accepts it either + positionally (`agent show TASK_ID`) or via flag (`--id` / `--task-id`, + plus `--run-id` for run-detail / run-events) -- the flag form matches + the rest of the CLI (`--job-id`, `--config-id`, ...). + + kbagent agent list + List all registered tasks (id, name, cron, type, state, last/next run). + + kbagent agent show TASK_ID + Full task detail including the action payload. + + kbagent agent create --name N [--description D] [--cron CRON] [--manual] + [--enabled/--disabled] + (--type ai_agent --cli claude|codex|gemini --prompt P + [--extra-arg ARG ...] [--timeout SECONDS] + |--type cli_command --argv ARG [--argv ARG ...] + [--timeout SECONDS] + |--type mcp_tool --tool TOOL [--mcp-project ALIAS] + [--mcp-branch ID] [--input JSON|@file|-] + [--timeout SECONDS] + |--from-file PATH|@path|-) + [--trigger-task-id ID --trigger-on success|error|always] + Persist a new scheduled task. --manual skips the cron loop. Use + --from-file for the full {{"type":..., "params":...}} JSON envelope + when prompts/args grow large. + + kbagent agent update TASK_ID [--name N] [--description D] [--cron C] + [--enabled/--disabled] [--manual/--auto] + [--clear-trigger] + [--trigger-task-id ID --trigger-on ...] + Patch one or more fields. Omitted flags leave the field unchanged. + --manual nulls next_run_at; --auto recomputes it from the cron expr. + + kbagent agent delete TASK_ID [--yes] + Permanent removal. Run history on disk is preserved. + + kbagent agent run TASK_ID [--stream] + [--runtime-prompt TEXT | --runtime-input JSON|@file|-] + Trigger immediately. --stream emits live events (one line per event; + NDJSON in --json mode). --runtime-prompt appends ad-hoc text to an + ai_agent's persisted prompt for this run only; --runtime-input merges + arbitrary JSON into the action params. + + kbagent agent runs TASK_ID [--limit N] + Run history (newest first). + + kbagent agent run-detail TASK_ID RUN_ID + Single AgentRun record (status, summary, output, error). + + kbagent agent run-events TASK_ID RUN_ID + Replay the persisted ai_agent event timeline. + + kbagent agent test [--type ... | --from-file PATH] [--stream] [--name N] + [common action flags from `create`] + Execute an action ad-hoc -- nothing is persisted. Useful for + sanity-checking a prompt / tool / argv before saving. + + kbagent agent cron-preview --cron "0 6 * * 1" [--count N] + Validate a cron expression and show the next N firings (UTC, max 20). + + kbagent agent prompt-improve --goal "..." [--draft "..."] + [--cli claude|codex|gemini] [--project ALIAS] + [--extra-arg X ...] [--stream/--no-stream] + AI-polished single-shot prompt for an unattended agent task. Spawns + the chosen AI CLI with a meta-prompt; the final `done` event's + `data.prompt` carries the cleaned body ready to paste into + `agent create --prompt ...`. + + See agent-tasks-cli-workflow.md skill reference for full walkthroughs. + ### MCP Tools (Multi-Project) kbagent tool list [--project NAME] [--branch ID] diff --git a/src/keboola_agent_cli/hints/definitions/__init__.py b/src/keboola_agent_cli/hints/definitions/__init__.py index 1cbda7d0..9729e689 100644 --- a/src/keboola_agent_cli/hints/definitions/__init__.py +++ b/src/keboola_agent_cli/hints/definitions/__init__.py @@ -1,6 +1,7 @@ """Hint definitions — import all modules to trigger registration.""" from . import ( + agent, # noqa: F401 branch, # noqa: F401 component, # noqa: F401 config, # noqa: F401 diff --git a/src/keboola_agent_cli/hints/definitions/agent.py b/src/keboola_agent_cli/hints/definitions/agent.py new file mode 100644 index 00000000..e41c83c1 --- /dev/null +++ b/src/keboola_agent_cli/hints/definitions/agent.py @@ -0,0 +1,16 @@ +"""Hint definitions for `kbagent agent ...` commands. + +Intentionally empty: the agent CRUD/run commands are pure-local (they +read/write ``/agents.json`` and spawn local subprocesses via +the runner). There is no Keboola HTTP API behind them, so the ``client`` +mode hints would have to invent a fake ``KeboolaClient.list_agent_tasks`` +that doesn't exist. + +The ``--hint`` framework falls back to a "No --hint available" message +when a command has no registered hint, which is the right answer here. +For programmatic use, refer to +:class:`keboola_agent_cli.services.agent_service.AgentService` directly +-- its methods mirror the CLI surface one-to-one. +""" + +from __future__ import annotations diff --git a/src/keboola_agent_cli/permissions.py b/src/keboola_agent_cli/permissions.py index 2ff95bf7..384d2b71 100644 --- a/src/keboola_agent_cli/permissions.py +++ b/src/keboola_agent_cli/permissions.py @@ -92,6 +92,22 @@ "workspace.query": "write", "workspace.from-transformation": "write", "workspace.gc": "destructive", + # Scheduled agent tasks (kbagent agent ...; touches only local + # agents.json + spawns subprocesses via the runner). + "agent.list": "read", + "agent.show": "read", + "agent.create": "write", + "agent.update": "write", + "agent.delete": "destructive", + # Run is classified write -- ai_agent/cli_command actions can mutate + # external state via subprocesses; deny-writes should block this. + "agent.run": "write", + "agent.runs": "read", + "agent.run-detail": "read", + "agent.run-events": "read", + "agent.test": "write", + "agent.cron-preview": "read", + "agent.prompt-improve": "write", # MCP tools "tool.list": "read", "tool.call": "write", diff --git a/src/keboola_agent_cli/server/__init__.py b/src/keboola_agent_cli/server/__init__.py index 874fe42b..9164ea31 100644 --- a/src/keboola_agent_cli/server/__init__.py +++ b/src/keboola_agent_cli/server/__init__.py @@ -1,743 +1,34 @@ -"""FastAPI HTTP server for kbagent. +"""Server package. -Wraps all services as REST endpoints. Designed to be consumed by the -web/backend Node.js BFF (which in turn serves the React UI in web/frontend). - -Key design choices: -- Bearer token auth (random secret generated at startup, printed to stdout) -- Localhost-only by default (bind 127.0.0.1) -- Per-request X-Manage-Token header for write ops requiring manage token -- Reuses existing services with their existing dict return shapes -- SSE for streaming: job log tail, branch reset progress, kai chat +Lazy attribute access via PEP 562 so importing helpers like +``from keboola_agent_cli.server.agents_store import AgentStore`` does NOT +drag in FastAPI/uvicorn. Those heavy deps are loaded only when +``create_app`` is actually fetched (typically by ``kbagent serve`` or +the FastAPI test suite). """ from __future__ import annotations -import asyncio -import contextlib -import logging -import secrets -from contextlib import asynccontextmanager - -from fastapi import FastAPI -from fastapi.middleware.cors import CORSMiddleware -from fastapi.openapi.utils import get_openapi -from fastapi.responses import JSONResponse -from starlette.exceptions import HTTPException as StarletteHTTPException - -from .. import __version__ -from ..config_store import ConfigStore, resolve_config_dir -from ..errors import ConfigError, ErrorCode, KeboolaApiError -from .agents_store import AgentStore -from .auth import PUBLIC_PATHS, AuthSettings, install_auth -from .dependencies import ServiceRegistry, install_registry -from .routers import ( - agents, - ai_chat, - branches, - components, - configs, - data_apps, - encrypt, - flows, - health, - jobs, - kai, - lineage, - mcp, - members, - org, - projects, - schedules, - search, - semantic_layer, - sharing, - storage, - workspaces, -) - -logger = logging.getLogger(__name__) - - -# OpenAPI tag metadata. Order matches the CLI groupings printed by -# ``kbagent --help`` so the Swagger UI sidebar reads top-down the same -# way users explore commands on the terminal: -# -# Project Management -> Configurations -> Data -> Execution -> -# Development -> AI & Tools -> System -# -# FastAPI uses this list both for ordering and for the per-section -# descriptions. The names below MUST match the ``tags=[...]`` value on -# each ``APIRouter`` in ``server/routers/`` -- a typo silently demotes -# a section to the end of the sidebar with no description. -OPENAPI_TAGS: list[dict[str, str]] = [ - # ---- Project Management ---- - { - "name": "projects", - "description": ( - "**Project Management.** " - "Register, list, edit, and remove Keboola project aliases. " - "Mirrors `kbagent project add|list|remove|edit|status|use|current|info`." - ), - }, - { - "name": "members", - "description": ( - "**Project Management.** " - "Invite users, list members and pending invitations, " - "change roles, and remove members. " - "Mirrors `kbagent project invite|member-*|invitation-*`." - ), - }, - { - "name": "org", - "description": ( - "**Project Management.** " - "Bulk-onboard an entire organization (Manage API). Requires " - "the `X-Manage-Token` header on every request -- the manage " - "token is never persisted in config. " - "Mirrors `kbagent org setup|refresh`." - ), - }, - # ---- Configurations ---- - { - "name": "configs", - "description": ( - "**Configurations.** " - "Browse, search, update, and manage component configurations " - "and rows (variables, metadata, folder, default bucket, " - "OAuth URL). " - "Mirrors `kbagent config *`." - ), - }, - { - "name": "components", - "description": ( - "**Configurations.** " - "Discover components (extractors, writers, applications, " - "transformations) and fetch their JSON schemas. " - "Mirrors `kbagent component list|detail`." - ), - }, - { - "name": "encrypt", - "description": ( - "**Configurations.** " - "Encrypt secret values for a specific project + component " - "using the Keboola encryption API. " - "Mirrors `kbagent encrypt values`." - ), - }, - # ---- Data ---- - { - "name": "storage", - "description": ( - "**Data.** " - "Buckets, tables, columns, files. Create, upload, download, " - "describe, swap, delete. " - "Mirrors `kbagent storage *`." - ), - }, - { - "name": "search", - "description": ( - "**Data.** " - "Cross-resource search over tables, buckets, configs, " - "flows, data-apps, and transformations. " - "Mirrors `kbagent search`." - ), - }, - { - "name": "sharing", - "description": ( - "**Data.** " - "Share buckets across projects and inspect the sharing " - "graph (edges). " - "Mirrors `kbagent sharing *`." - ), - }, - # ---- Execution ---- - { - "name": "jobs", - "description": ( - "**Execution.** " - "Run components, inspect job history, terminate running " - "jobs. " - "Mirrors `kbagent job list|detail|run|terminate`." - ), - }, - { - "name": "flows", - "description": ( - "**Execution.** " - "Orchestrator and Flow CRUD, scheduling, run history. " - "Mirrors `kbagent flow *`." - ), - }, - { - "name": "schedules", - "description": ( - "**Execution.** " - "Cron-style schedules attached to flows / configurations. " - "Mirrors `kbagent schedule list|detail|find`." - ), - }, - { - "name": "data-apps", - "description": ( - "**Execution.** " - "Streamlit / R / Python data apps -- create, deploy, " - "start/stop, manage secrets. " - "Mirrors `kbagent data-app *`." - ), - }, - { - "name": "workspaces", - "description": ( - "**Execution.** " - "Snowflake / BigQuery workspaces -- CRUD, load tables, " - "run SQL via Query Service, GC orphans. " - "Mirrors `kbagent workspace *`." - ), - }, - # ---- Development ---- - { - "name": "branches", - "description": ( - "**Development.** " - "Dev branch lifecycle (create / use / reset / delete / " - "merge) and branch metadata. " - "Mirrors `kbagent branch *`." - ), - }, - { - "name": "lineage", - "description": ( - "**Development.** " - "Build and query cross-project data lineage (table-level " - "and column-level). " - "Mirrors `kbagent lineage build|show|info`." - ), - }, - { - "name": "semantic-layer", - "description": ( - "**Development.** " - "Model, validate, import/export, diff, promote, and build " - "semantic layer artifacts (datasets, metrics, " - "relationships, constraints, glossary). " - "Mirrors `kbagent semantic-layer *`." - ), - }, - # ---- AI & Tools ---- - { - "name": "mcp", - "description": ( - "**AI & Tools.** " - "List and call MCP tools across one or all projects. " - "Mirrors `kbagent tool list|call`." - ), - }, - { - "name": "kai", - "description": ( - "**AI & Tools.** " - "Keboola AI (Kai) -- ping, preflight, single-shot ask, " - "chat with history. " - "Mirrors `kbagent kai *`." - ), - }, - { - "name": "ai-chat", - "description": ( - "**AI & Tools.** " - "Server-side streaming AI chat (SSE) used by the kbagent " - "web UI. No CLI equivalent." - ), - }, - { - "name": "agents", - "description": ( - "**AI & Tools.** " - "Scheduled / on-demand AI agent tasks. Server-only feature " - "(no CLI equivalent) -- the scheduler loop runs inside " - "`kbagent serve` and persists tasks + runs to the config " - "directory." - ), - }, - # ---- System ---- - { - "name": "health", - "description": ( - "**System.** " - "Liveness ping, auth-info bootstrap, version, changelog, " - "and doctor checks. `/health/ping` is the only public " - "endpoint -- everything else requires Bearer auth." - ), - }, -] - - -APP_DESCRIPTION = """\ -HTTP API surface for **kbagent**. Wraps every CLI command as a REST -endpoint so the kbagent web UI and any HTTP client (curl, the -`kbagent http` proxy, scheduled agents, Node BFFs, ...) can drive -Keboola the same way the terminal does. - -## Authentication - -Every endpoint except `GET /health/ping`, `GET /docs`, `GET /redoc`, -and `GET /openapi.json` requires a bearer token. The token is -generated when `kbagent serve` starts and printed to stdout -- click -**Authorize** at the top right of this page and paste it once. - -Endpoints under **org** additionally require an `X-Manage-Token` -header (the Keboola Manage API token). It is never persisted; pass -it per request. - -## Layout - -Sections below are grouped roughly the same way `kbagent --help` groups -its command tree: - -- **Project Management** -- projects, members, org -- **Configurations** -- configs, components, encrypt -- **Data** -- storage, search, sharing -- **Execution** -- jobs, flows, schedules, data-apps, workspaces -- **Development** -- branches, lineage, semantic-layer -- **AI & Tools** -- mcp, kai, ai-chat, agents -- **System** -- health - -Most endpoints accept a `project` alias either in the body or as a -query parameter; multi-project endpoints accept `project` repeatedly. -""" - - -def _build_custom_openapi(app: FastAPI): - """Return a closure that generates the OpenAPI schema with auth schemes. - - FastAPI's default ``get_openapi(...)`` does not know about the - ``BearerAuthMiddleware`` (it's an ASGI middleware, not a per-route - dependency), so the generated schema has no ``securitySchemes`` and - Swagger UI shows no **Authorize** button. We patch the schema after - generation: - - 1. Declare a ``BearerAuth`` HTTP scheme (the global default for every - endpoint that is not in ``PUBLIC_PATHS``). - 2. Declare a ``ManageToken`` API-key scheme (header ``X-Manage-Token``) - used by the ``org`` router and any future endpoint that requires the - Manage API token. Endpoints opt in by listing it in their - ``openapi_extra={"security": [{"BearerAuth": [], "ManageToken": []}]}``. - 3. Apply ``BearerAuth`` globally and clear ``security`` for public paths - so Swagger UI shows them as unsecured (matching the actual middleware - behavior). - - The result is cached on ``app.openapi_schema`` per FastAPI convention. - """ - - def custom_openapi() -> dict: - if app.openapi_schema: - return app.openapi_schema - schema = get_openapi( - title=app.title, - version=app.version, - description=app.description, - routes=app.routes, - tags=app.openapi_tags, - ) - components = schema.setdefault("components", {}) - components["securitySchemes"] = { - "BearerAuth": { - "type": "http", - "scheme": "bearer", - "description": ( - "Bearer token printed to stdout when `kbagent serve` starts. " - "Paste it once and Swagger UI will attach it to every request." - ), - }, - "ManageToken": { - "type": "apiKey", - "in": "header", - "name": "X-Manage-Token", - "description": ( - "Keboola Manage API token. Required only by `/org/*` " - "endpoints. Never persisted; passed per request." - ), - }, - } - schema["security"] = [{"BearerAuth": []}] - # Public paths in PUBLIC_PATHS are exempt from the bearer-auth - # middleware. Reflect that in the schema so Swagger UI does not - # mislabel them as locked. - for path in PUBLIC_PATHS: - path_item = schema.get("paths", {}).get(path) - if not path_item: - continue - for op in path_item.values(): - if isinstance(op, dict): - op["security"] = [] - app.openapi_schema = schema - return schema - - return custom_openapi - - -def _format_error( - message: str, error_code: ErrorCode | str, *, http_status: int = 400 -) -> JSONResponse: - """Render a kbagent-style error envelope at the given HTTP status. - - ``error_code`` accepts both :class:`ErrorCode` enum members (the canonical - surface; matches CLI error envelopes byte-for-byte) and raw strings (for - forward compatibility when a router wraps a third-party exception whose - code is not yet in the enum). The :class:`ErrorCode` mixes in ``str``, so - both shapes serialise as plain strings in the JSON body. - """ - return JSONResponse( - status_code=http_status, - content={ - "status": "error", - "error": { - "code": str(error_code), - "message": message, - }, - }, - ) - - -def create_app( - *, - config_dir: str | None = None, - auth_token: str | None = None, - cors_origins: list[str] | None = None, - serve_url: str | None = None, - ui_dist: str | None = None, -) -> FastAPI: - """Build and configure the FastAPI application. - - Args: - config_dir: Override config directory (matches kbagent --config-dir). - auth_token: Bearer token clients must send. If None, generates one. - cors_origins: Allowed CORS origins. Default: localhost dev ports. - serve_url: Self-URL (``http://host:port``) of this server. Stored on - the registry so agent subprocesses can be told where to call back. - If None, defaults are still injected into subprocess env using - ``127.0.0.1:8001`` -- the serve_command default. - ui_dist: Optional absolute path to a built React ``dist/`` directory - (the output of ``npm run build`` in ``web/frontend``). When set, - the FastAPI app additionally: - - 1) accepts ``/api/`` as an alias for the bare ```` - (the BFF used to re-strip this prefix; in single-process mode - we do it server-side via an ASGI path-rewrite middleware), - 2) mounts the dist directory at ``/`` so static assets and the - SPA fallback are served by uvicorn directly, - 3) intercepts ``GET /`` to inject ``window.__KBAGENT_TOKEN`` into - ``index.html`` so the SPA boots already authenticated -- no - BFF and no manual paste step. - - If the path does not exist, the UI mount is skipped silently and - a warning is logged so ``--ui`` typos don't break the API path. - - Returns: - Configured FastAPI app ready for uvicorn. - """ - resolved_token = auth_token or secrets.token_urlsafe(32) - - @asynccontextmanager - async def _lifespan(app_: FastAPI): - # Start the agent scheduler loop in the background once services - # exist (registry is installed below before include_router calls). - from .agent_runner import scheduler_loop - - scheduler_task = None - store = getattr(app_.state, "agent_store", None) - registry_ = getattr(app_.state, "registry", None) - if store is not None and registry_ is not None: - scheduler_task = asyncio.create_task(scheduler_loop(store, registry_)) - logger.info("Agent scheduler task spawned") - try: - yield - finally: - if scheduler_task is not None: - scheduler_task.cancel() - with contextlib.suppress(asyncio.CancelledError): - await scheduler_task - - app = FastAPI( - lifespan=_lifespan, # type: ignore[arg-type] - title="kbagent serve", - description=APP_DESCRIPTION, - version=__version__, - docs_url="/docs", - redoc_url="/redoc", - openapi_tags=OPENAPI_TAGS, - swagger_ui_parameters={ - # Keep the bearer token across page refreshes so the user only - # has to paste it once per browser session. The token is held in - # the Swagger UI in-memory store (and localStorage when persist - # is true) -- safe for a localhost-only dev tool, and a major - # ergonomics win for exploring the API. - "persistAuthorization": True, - # Default tag ordering follows ``openapi_tags`` above; this - # toggle just keeps Swagger from re-sorting operations within - # each tag alphabetically (we want them in router declaration - # order, which usually mirrors a logical workflow). - "operationsSorter": None, - "docExpansion": "none", - }, - ) - app.openapi = _build_custom_openapi(app) # type: ignore[method-assign] +from typing import TYPE_CHECKING, Any - app.add_middleware( - CORSMiddleware, - allow_origins=cors_origins - or [ - "http://localhost:5173", # Vite dev default - "http://localhost:8000", # Node BFF - "http://127.0.0.1:5173", - "http://127.0.0.1:8000", - ], - allow_credentials=True, - allow_methods=["*"], - allow_headers=["*"], - ) - - install_auth(app, AuthSettings(token=resolved_token)) - - resolved_dir, source = resolve_config_dir(cli_config_dir=config_dir) - config_store = ConfigStore(config_dir=resolved_dir, source=source) - registry = ServiceRegistry( - config_store=config_store, - serve_url=serve_url, - serve_token=resolved_token, - ) - install_registry(app, registry) - - app.state.agent_store = AgentStore(resolved_dir) - - from .run_broadcaster import install_broadcaster - - install_broadcaster(app) - - @app.exception_handler(ConfigError) - async def _config_error_handler(_request, exc: ConfigError): - return _format_error(str(exc), ErrorCode.CONFIG_ERROR, http_status=400) - - @app.exception_handler(KeboolaApiError) - async def _api_error_handler(_request, exc: KeboolaApiError): - code = getattr(exc, "error_code", ErrorCode.API_ERROR) - msg = getattr(exc, "message", str(exc)) or str(exc) - return _format_error(msg, code, http_status=502) - - @app.exception_handler(StarletteHTTPException) - async def _starlette_handler(_request, exc: StarletteHTTPException): - return _format_error( - exc.detail or "HTTP error", ErrorCode.HTTP_ERROR, http_status=exc.status_code - ) - - @app.exception_handler(Exception) - async def _generic_handler(_request, exc: Exception): - logger.exception("Unhandled error: %s", exc) - return _format_error(str(exc) or repr(exc), ErrorCode.INTERNAL_ERROR, http_status=500) - - app.include_router(health.router) - app.include_router(projects.router) - app.include_router(members.router) - app.include_router(configs.router) - app.include_router(components.router) - app.include_router(storage.router) - app.include_router(jobs.router) - app.include_router(branches.router) - app.include_router(workspaces.router) - app.include_router(flows.router) - app.include_router(schedules.router) - app.include_router(lineage.router) - app.include_router(sharing.router) - app.include_router(data_apps.router) - app.include_router(mcp.router) - app.include_router(kai.router) - app.include_router(ai_chat.router) - app.include_router(encrypt.router) - app.include_router(search.router) - app.include_router(semantic_layer.router) - app.include_router(org.router) - app.include_router(agents.router) - - app.state.auth_token = resolved_token - - if ui_dist: - _install_ui(app, ui_dist=ui_dist, token=resolved_token) - - return app - - -_SESSION_COOKIE_NAME = "kbagent_session" - - -def _install_ui(app: FastAPI, *, ui_dist: str, token: str) -> None: - """Mount the built React SPA at ``/`` and bridge ``/api/*`` to bare routes. - - Three pieces: - - 1) **Path-rewrite middleware** for ``/api/`` -> ````. Mounted - BEFORE auth so the auth header check runs against the rewritten path - (auth doesn't care about path, but PUBLIC_PATHS exact-matches do). - 2) **Cookie-setting** ``GET /`` and ``GET /index.html``: read the built - ``index.html``, return it with a ``Set-Cookie: kbagent_session=; - HttpOnly; SameSite=Strict; Path=/`` header. Public (no auth) so the - SPA can bootstrap. The browser then attaches the cookie to every - same-origin REST + SSE request automatically. The token is HttpOnly - (no JS access -- XSS-resistant), SameSite=Strict (no cross-origin - sends -- CSRF-resistant), and lives only for the browser session. - - This replaces the older "inject ``window.__KBAGENT_TOKEN`` into a - ``