Adding a Provider

Payments, Storage, Notifications and the Scheduler all work the same way: kirak.json lists provider instances, each instance has a type, and a class implements that type. This page is the one place that explains how to add a new type. The module pages (Payments, Storage, Notifications, Scheduler) give a worked example for each.

How it fits together

kirak.json                     registry                      instance
"providers": {                 type name -> class            name -> object
  "main": {"type": "aws"} ---> built-ins, then           ---> created on first use,
}                              register_provider(),          cached, only listed
                               then entry points             names are usable

A type is an implementation (aws, smtp, redis). A class declares its type in TYPE_NAME.
An instance is a named, configured use of a type. The name is the key under providers. Two instances can share a type (two S3 buckets). Callers pick an instance with "provider": "<instance name>", never a type.
default_provider names the instance used when a call passes no provider.
Only instances listed in kirak.json can be used, including from a request payload. Anything else is PROVIDER_NOT_CONFIGURED (400).
Provider code is imported only when an instance is first used, so a missing optional SDK never breaks apps that don’t use it.

Pick a path

I want to…	Do this
Use a provider in one app	Write the class and call `register_provider` in `on_kirak_ready`
Share a provider as a package	Add an entry point; users need only `pip install` and a `type` in `kirak.json`
Add it to Kirak itself	Add the class and one registry line, plus tests and docs

All three start with the same class.

Per-module reference

Module	Base class	Register with	Entry point group	Secrets prefix
Payments	`kirak.payments.providers.base.PaymentProvider`	`kirak.payments.register_provider(type, cls)`	`kirak.payments_providers`	`KIRAK_PAYMENT_<INSTANCE>_`
Storage	`kirak.storage.providers.base.StorageProvider`	`kirak.storage.register_provider(type, cls)`	`kirak.storage_providers`	`KIRAK_STORAGE_<INSTANCE>_`
Notifications, email	`kirak.notifications.providers.base.EmailProvider`	`kirak.notifications.register_provider("email", type, cls)`	`kirak.notifications_email_providers`	`KIRAK_NOTIFICATION_EMAIL_<INSTANCE>_`
Notifications, SMS	`...base.SMSProvider`	`kirak.notifications.register_provider("sms", type, cls)`	`kirak.notifications_sms_providers`	`KIRAK_NOTIFICATION_SMS_<INSTANCE>_`
Notifications, push	`...base.PushProvider`	`kirak.notifications.register_provider("push", type, cls)`	`kirak.notifications_push_providers`	`KIRAK_NOTIFICATION_PUSH_<INSTANCE>_`
Notifications, chat	`...base.IMProvider`	`kirak.notifications.register_provider("im", type, cls)`	`kirak.notifications_im_providers`	`KIRAK_NOTIFICATION_IM_<INSTANCE>_`
Notifications, webhook	`...base.WebhookProvider`	`kirak.notifications.register_provider("webhook", type, cls)`	`kirak.notifications_webhook_providers`	`KIRAK_NOTIFICATION_WEBHOOK_<INSTANCE>_`
Scheduler	`kirak.scheduler.backends.base.QueueBackend`	`kirak.scheduler.register_provider(type, cls)`	`kirak.scheduler_providers`	`KIRAK_SCHEDULER_<INSTANCE>_`

The methods each base class requires are listed on its module page. The AI module is not part of this mechanism: each agent names its own provider:model.

The contract

Every provider class:

class AcmeProvider(SomeBaseClass):
    TYPE_NAME = "acme"                    # the "type" used in kirak.json
    SECRET_FIELDS = ("api_key",)          # config fields read from env, never from kirak.json

    def __init__(self, config, module=None):
        super().__init__(config, module)
        self._require("api_key")          # fail early when required config is missing
        ...

Item	Rule
`TYPE_NAME`	Lowercase letters, digits and `_`, starting with a letter. It is what users write as `"type"`.
`SECRET_FIELDS`	Field `F` of instance `main` is read from `<prefix>MAIN_F`, upper-cased (see the table above). A secret written in `kirak.json` is rejected.
Constructor	`(config, module=None)`. `config` holds the instance’s own `kirak.json` entry, its secrets, and `name` (the instance name). `type` and `name` are reserved keys.
`self.instance_name`	The instance name. Use it wherever you record which provider handled something.
`self._require(*fields)`	Raises `ConfigurationError` naming the missing fields.
Failures	Raise `KirakException` with a stable `code`; never return a `{"success": False}` dict.
SDK imports	Import your SDK at the top of your module. If it is missing, Kirak reports `PROVIDER_IMPORT_ERROR` with an install hint.
Shared module settings	Read them from the owning module (for example `self.scheduler.config.poll_interval`), not from your instance config.

A provider only ever sees its own instance config, never another instance’s secrets.

Register it

In one app

async def on_kirak_ready(kirak):
    kirak.storage.register_provider("acme", AcmeProvider)

on_kirak_ready runs after Kirak starts and before provider config is validated, so the type is known when kirak.json is checked. register_provider can also be used as a decorator by leaving out the class. It raises ValueError if the type name is already taken (including by a built-in) and TypeError if the class does not subclass the module’s base class.

Then list an instance:

"storage": {
  "default_provider": "main",
  "providers": { "main": { "type": "acme", "bucket": "assets" } }
}

As a pip package (entry point)

# pyproject.toml of your package
[project.entry-points."kirak.storage_providers"]
acme = "kirak_acme.provider:AcmeProvider"

After pip install kirak-acme, a "type": "acme" in kirak.json finds it with no registration code. The rules:

Lookup order is built-ins, then register_provider, then entry points. A built-in or registered name wins over an entry point.
Only the entry point whose name matches a configured type is imported.
Installing a package activates nothing: the type must still be listed in kirak.json.
Two installed packages claiming the same name is an error (PROVIDER_INVALID). A broken import is PROVIDER_IMPORT_ERROR and names the package.

Into Kirak core

Add the class under kirak/<module>/services/ (or kirak/scheduler/backends/, kirak/vector/providers/).
Declare it with a ProviderSpec in kirak/catalog/specs/<module>.py, in the right provider set: type, one-line summary, website (the vendor’s site, used by tools such as Kirak Studio for a logo; left out only for a provider with no vendor, like local or smtp), class_path ("dotted.path:ClassName"), pip_extra and install_check (a package the extra installs, or both None), every setting the class reads from kirak.json (name, type, description, default, required), every secret (name, description, required) and the names of the capability mixins it implements. The module’s registry builds its built-in table from these specs; there is no second list to edit.
In the class, take the secret field names from the spec: SECRET_FIELDS = _specs.<NAME>.secret_fields with from kirak.catalog.specs import <module> as _specs.
If the SDK is optional, add the extra to pyproject.toml.
Add tests. tests/test_catalog/test_provider_websites.py fails without a website, and tests/test_catalog/test_specs_match_code.py fails until the spec matches the class: TYPE_NAME, SECRET_FIELDS, capability mixins, and every config key the class reads (config.get("x"), config["x"], _int_setting("x"), _require("x")).
Implement its credential check (see below) and describe it in the spec’s verifies; test_specs_match_code.py fails without both.
Document the type in the module page. kirak providers <module> <type> shows its settings, secrets and a kirak.json example straight from the spec.

A spec file imports nothing but kirak.catalog.types, so kirak providers can describe a provider whose SDK is not installed.

A third-party provider (its own package, registered by entry point) needs no spec: kirak providers lists it from its entry point and reads SECRET_FIELDS, capability mixins and an optional WEBSITE (the vendor’s URL) from its class.

Credential check

Every provider can tell whether the config it was built with works. The caller passes everything: the instance’s kirak.json settings and its secrets under their field names. Nothing is read from .env, kirak.json or the environment.

provider = AcmeProvider({"bucket": "assets", "api_key": "...", "name": "main"}, None)
result = await provider.check()               # live=True, write=False, timeout=10.0
result.status    # "ok" | "warning" | "unverified" | "failed"
result.code      # e.g. "credentials_invalid"; None when ok
result.message   # one line, never a secret value
result.detail    # non-secret facts, e.g. {"bucket": "assets"}

live=False runs only the checks that need no network (fast enough to run while a key is typed).
write=True lets a storage provider write and delete one small object; other providers ignore it.
A constructor that refuses the config raises ConfigurationError before check() can run; report it as failed / credentials_missing.
A live check is read-only: nothing is billed, sent, created or changed (write=True is the only exception). An answer the check cannot classify is unverified, never ok.
The codes are listed in Problem codes (those marked provider check()).

check() comes from the base class; a provider implements the parts behind it:

from kirak.core.provider_check import CheckResult, http_failure, http_get


class AcmeProvider(SomeBaseClass):
    async def _check_live(self, *, write, timeout) -> CheckResult:
        # The cheapest read-only call that proves the credentials.
        response = await http_get("https://api.acme.com/v1/account", timeout=timeout,
                                  headers={"Authorization": f"Bearer {self.config['api_key']}"})
        if not response.is_success:
            return http_failure(response, "Acme")   # 401 / 403 / 404 / other
        return CheckResult.ok("Acme accepted the API key")

    def _check_offline(self):
        # Optional: formats, test-mode keys. None when nothing is wrong.
        return self._mode_result(self.config["api_key"].startswith("test_"), "api_key")

Helpers in kirak.core.provider_check: http_get and http_request (one request, no retries; http_request for a POST such as fetching an OAuth token), http_failure (401, 403, 404 and the rest; pass resource=False when the URL names no configured resource, so a 404 is inconclusive), aws_call (one boto3 call off the event loop, AWS errors classified) gcs_call (the same for one google-cloud-storage call) and azure_call (one awaited azure-storage-blob call, Azure errors classified). self._mode_result(test, "secret_key") reports a test-mode (a warning) or live-mode key or environment from _check_offline.

Network failures and timeouts may simply propagate from _check_live: check() reports them as provider_unreachable / provider_timeout, and masks every secret value and URL password in the result. A provider without _check_live reports unverified / check_not_supported. A built-in provider also describes its check in its ProviderSpec (verifies), which kirak providers and the provider tables show.

Test it

Kirak ships a static conformance check. It does not instantiate your class or need its SDK:

from kirak.core.providers import assert_provider_conforms
from kirak.storage.providers.base import StorageProvider

from my_package import AcmeProvider


def test_conforms():
    assert_provider_conforms(AcmeProvider, StorageProvider)

It reports every problem at once: wrong base class, unimplemented abstract methods, a missing or malformed TYPE_NAME, bad SECRET_FIELDS, a constructor that does not take (config, module), and a malformed EVENT_MAP. Add your own tests for behavior, and cover at least these cases:

the happy path, using the SDK’s own test mode or a mock;
a failing call raises KirakException with your code (not a raw SDK exception);
missing required config raises ConfigurationError;
two instances of your type do not share state or secrets;
(payments) a webhook delivered twice runs the completed hook once: mark the transaction with await self._claim_completion(transaction_id, data), and when it returns False add "already_processed": True to the result handle_webhook returns. Payments.handle_webhook then skips the event’s hook. The claim is one conditional update on transactions, so it is safe against two deliveries arriving together.

Errors callers can see

Code	Status	Meaning
`PROVIDER_NOT_CONFIGURED`	400	`provider` is not listed in `kirak.json` (or the module has no providers)
`PROVIDER_UNKNOWN_TYPE`	500	A configured `type` is not registered anywhere
`PROVIDER_IMPORT_ERROR`	500	The provider’s package or SDK could not be imported; the message includes an install hint
`PROVIDER_INVALID`	500	A registered class does not subclass the base class, or two entry points claim one type
`CONFIGURATION_ERROR`	500	Startup validation failed: no providers, missing `default_provider`, a missing `type`, a secret written in `kirak.json`, or (Scheduler) more than one `database` provider

Startup validation runs when routers are mounted, so these are reported when the app starts and not on the first request.

Checklist

Subclasses the module’s base class and implements every abstract method
TYPE_NAME and SECRET_FIELDS declared; assert_provider_conforms passes
Built-in only: ProviderSpec in kirak/catalog/specs/<module>.py; tests/test_catalog passes
Constructor takes (config, module=None) and uses _require for required config
_check_live makes the cheapest read-only call that proves the credentials (built-in: verifies in the spec)
Failures raise KirakException with a stable code
Records self.instance_name, not a hardcoded type name, wherever it identifies itself
Tested with two instances of the type
Registered (app, entry point, or core registry line) and documented