kbagent notification: fleet-wide audit of Flow Notification subscriptions (Notification Service API)

[MichalProchazkaP3]

Problem

Auditing a Keboola org's Flows for "who gets notified when something breaks" is only half-possible today.

kbagent flow list / config detail --component-id keboola.flow fully expose:

• flow descriptions
• in-flow Notification tasks (a task node of type: "notification" with recipients: [{channel, address}], added inside a phase)

They do not expose the Flow Builder's own Notifications tab — the per-flow Success / Error / Processing-delay / Warning cards with their own recipient lists, reachable via the bell icon in the UI. That's a genuinely different mechanism: it's backed by a separate Notification service, not stored inside the flow's configuration JSON.

I ran a fleet-wide audit for a customer (20 projects, 276 keboola.flow + keboola.orchestrator configs) checking every email address referenced anywhere in flow descriptions, in-flow notification tasks, and email-sending component configs (kds-team.app-email-smtp-sender, kds-team.app-mailgun-v2) — and found real issues (placeholder addresses, typo'd recipients, missing owners). But the Notifications-tab recipients — arguably the most important ones, since they're what actually pages someone when a production flow fails — were completely unreachable from kbagent, the MCP server, or any CLI. The only fallback was "open each flow in the UI by hand."

What exists today (confirmed via public discovery, no auth needed)

• GET https://connection.<region>.keboola.com/v2/storage lists platform services, including:

  {"id": "notification", "url": "https://notification.eu-central-1.keboola.com"}

• Its swagger is public: https://notification.eu-central-1.keboola.com/docs/swagger.yaml. Relevant paths:
  • GET /project-subscriptions — list all subscriptions for the project resolved from the auth token. Optional ?event= filter.
  • POST /project-subscriptions / DELETE /project-subscriptions/{id} — create/remove.
  • Auth: security: [StorageApiTokenAuth, StorageApiBearerToken, ...] — a plain Storage API token, the same kind every KeboolaClient in kbagent already holds per project. No elevated scope (e.g. canManageTokens) needed for the read path — I originally assumed it would and burned a while confirming otherwise.
  • Schema: event (jobFailed, jobSucceeded, jobSucceededWithWarning, jobProcessingLong, plus phaseJob* variants), filters (field/value/operator, e.g. matching configurationId/component), recipient (channel: email|webhook + address), optional expiresAt.

So the missing piece is purely a client + CLI surface in kbagent — the API itself is a normal sibling service exactly like Queue/Encryption/Sync-Actions, which kbagent already talks to.

Proposed solution

Follow the existing queue / encryption / sync-actions pattern in client/_core.py + the existing schedule command as the structural template (it already solves the identical shape of problem — "audit a per-project sibling-service concept across every registered project, joined against keboola.flow/keboola.orchestrator config names").

1. client/_core.py: add

   @property
   def _notification_base_url(self) -> str:
       return self._derive_service_url(self._stack_url, "notification")
   
   def _notification_request(self, method: str, path: str, **kwargs: Any) -> httpx.Response:
       client = self._get_or_create_sub_client("_notification_client", self._notification_base_url)
       return self._do_request(method, path, client=client, base_url=self._notification_base_url, **kwargs)

   plus the matching self._notification_client: httpx.Client | None = None init and close() wiring — copy-paste of the _queue_* / _encrypt_* trio.

2. client/notifications.py (new mixin, modeled on client/queue.py):

   class _NotificationMixin(_CoreClient):
       def list_project_subscriptions(self, event: str | None = None) -> list[dict[str, Any]]:
           params = {"event": event} if event else {}
           return self._notification_request("GET", "/project-subscriptions", params=params).json()

   (Write methods — create_project_subscription / delete_project_subscription — are trivial to add too, but out of scope for this ask; read-only is enough to close the audit gap.)

3. services/notification_service.py (new, modeled on services/schedule_service.py::ScheduleService):

   • list_subscriptions(aliases: list[str] | None, event: str | None, component_id: str | None, config_id: str | None) -> dict — fans out self._run_parallel over self.resolve_projects(aliases), calls list_project_subscriptions(event) per project.
   • Joins each subscription's filters (configurationId / component) against list_components_with_configs() from the same project — exactly the _partition_components trick ScheduleService uses to resolve parent_name — so the output row can show which flow a subscription belongs to, not just a bare configurationId.
   • Same error-accumulation contract as ScheduleService: per-project failures land in errors, never abort the whole fan-out.

4. commands/notification.py (new, modeled on commands/schedule.py):

   kbagent notification list [--project ALIAS...] [--event jobFailed|jobSucceeded|jobSucceededWithWarning|jobProcessingLong] [--component-id keboola.flow] [--config-id ID]
   

   Output columns: project_alias, subscription_id, event, flow_name (resolved), component_id/config_id, channel, address, expires_at. Register the command group next to schedule_app in the root Typer app and in check_cli_permission the same way (read-only, safe under --deny-writes).

5. Docs: mention notification list from flow list's docstring as the companion command for "what does the Notifications tab say", the way flow list already flags the legacy_orchestrator_count caveat for orchestrator flows. Add a changelog entry.

Open questions / things to verify while implementing

• Whether subscriptions are branch-aware (dev branch flows). The swagger doesn't show a branch filter — worth checking against a real dev-branch flow before assuming production-only scope, similar to the existing branch caveat on schedule find --not-run-since (Queue API isn't branch-aware either).
• Whether filters always contains a configurationId for flow-level subscriptions, or whether some events are project-wide with no filters at all (e.g. a catch-all "notify me on any job failure" subscription) — the join logic needs to degrade gracefully to "project-wide, no specific flow" rather than erroring.

Non-goals for this issue

• Write path (create/delete subscriptions) — natural follow-up, not needed to close the audit gap.
• Anything about the in-flow type: "notification" task (already fully supported today via flow detail/config detail) — this issue is only about the Notifications-tab / Notification-service side.

Why it matters

This was the one blind spot left after an otherwise complete fleet-wide "are our flow notification emails valid" audit across 20 projects — every other notification surface (owner emails in descriptions, in-flow notification tasks, email-sending component recipients) was fully auditable via existing kbagent commands in a few minutes; this one required opening every flow by hand in the UI. Given kbagent's stated goal of wrapping "everything ... into workflow-oriented commands where dev branches propagate automatically, multi-project operations run in parallel" — this is one of the last multi-project operations that still can't be done from the CLI.

[padak]

Verified this proposal against both the codebase and the live swagger (https://notification.eu-central-1.keboola.com/docs/swagger.yaml, fetched 2026-08-19). The shape of the plan holds up — the Notification service is a plain sibling of Queue/Encryption/Billing and _derive_service_url(stack_url, "notification") works unchanged. But three details in the spec don't match the API, and both open questions are answerable from the swagger. Posting before implementing so nobody builds on the wrong keys.

Confirmed as stated

• {"id": "notification", "url": "https://notification.eu-central-1.keboola.com"} is in the /v2/storage service index.
• Auth on the read path is StorageApiTokenAuth / StorageApiBearerToken — a plain project Storage token, no elevated scope. Your correction to your own initial assumption was right.
• The _queue_* copy-paste in client/_core.py and the ScheduleService template are the right structural precedents.

Three corrections

1. Event names are kebab-case, and EventName is not an enum.

The issue lists jobFailed / jobSucceeded / jobSucceededWithWarning / jobProcessingLong. The wire format is:

job-failed
job-succeeded
job-succeeded-with-warning
job-processing-long
phase-job-failed
phase-job-succeeded
phase-job-succeeded-with-warning
phase-job-processing-long

EventName is declared as a free-form type: string, minLength: 1 with example: job-failed — there is no enum in the schema. So --event should be a plain string option, not a hard Enum that would reject any event the service adds later. Validating against the known set as a warning is fine; rejecting is not.

2. Filter field names are dotted paths, not flat keys.

The issue assumes configurationId and component. Actual filter fields:

filters:
  - field: job.component.id        # not "component"
    value: 'my.component'
  - field: job.configuration.id    # not "configurationId"
    value: '12345'

plus branch.id, phase.id (phase-job-* events) and durationOvertimePercentage (threshold filter). operator is an enum of exactly == and >=, defaulting to == when omitted.

The naming makes sense once you look at Event_JobFailed: the event payload is a nested document ({project: {...}, branch: {id}, job: {id, component: {id, name}, configuration: {id, name}, tasks: [...]}}) and field is a path into it. Worth encoding that as a comment in the service, because it also tells you which other fields are matchable.

This is the correction that matters most for the join logic in step 3 — with the flat keys the resolved flow_name column would silently be empty on every row.

3. Webhook recipients carry url, not address.

recipient is a oneOf discriminated on channel:

• RecipientChannel_Email → required channel, address (format: email)
• RecipientChannel_Webhook → required channel, url (format: uri)

So the proposed address output column needs to be address or url, otherwise every webhook subscription renders blank.

Both open questions, answered from the swagger

Branch-awareness: yes. branch.id is a supported filter field and there is a dedicated Subscription_JobFailed_Branched example. Still worth a live check that the Flow Builder UI actually writes that filter when a subscription is created inside a dev branch — the schema supporting it doesn't prove the producer emits it.

Always a configurationId? No. Only event and recipient are required on NewProjectSubscription; filters is optional. A project-wide catch-all ("notify me on any job failure") is a legitimate, schema-valid subscription. The join must degrade to "project-wide, no specific flow" exactly as you anticipated, not error.

One thing the issue leaves on the table

GET /project-subscriptions/{subscriptionId} exists with the same Storage-token auth, so a notification detail subcommand is essentially free alongside list. Still read-only, still inside the stated scope.

POST /notifications (send a direct notification) is a different beast — it requires a Manage API application token with the notifications:send-notification scope, not a Storage token. Out of scope here, and it would need the same default-deny handling the rest of the manage-token surface gets.

Implementation note

Adding a new command group also triggers the full plugin/doc sync checklist (CLAUDE.md convention #17): context.py, the ## All CLI Commands section, keboola-expert.md, SKILL.md, commands-reference.md, gotchas.md, an E2E test, plus changelog. FYI on keboola-expert.md: it is at 60656 / 62000 bytes, so the entry there has to be a one-line trigger, not prose.

Picking this up now on claude/issue-600-3d974b, read-only notification list + notification detail, using the corrected event names and filter paths.