Agentic Workflow Steps (Python code steps)#

What this covers. How a plugin declares its own Agentic Workflow steps: the authoring config.json schema field by field, the authored→packaged rename map, the automation-step/* build-time validation rules, the complete data_type enum and its Python/wire encodings, the inputs/outputs contract, what the workflow builder user sees, and the Python runtime contract (inputs, outputs, secrets, kizen) including the kizen.api client, secrets namespacing, retry patterns, database-connector patterns, and how a step fails cleanly.

See also: manifest reference · platform API · auth, secrets and services · record actions · errors and observability · release and publish

Model#

A plugin-provided step is a Python script plus a declaration of its inputs and outputs. Publishing the plugin registers it; from then on it appears in the Agentic Workflow builder's action picker alongside native steps, and a workflow author maps each declared input to a record field, a workflow variable, or a static value, and each declared output to a field or variable.

Steps run server-side, in a sandboxed Python container — not in the browser worker that runs blocks, actions, and the other JavaScript surfaces. There is no this, no DOM, no toasts, no modals. The only I/O a step has is its declared inputs and outputs, its granted secrets, outbound HTTP, and the injected kizen client.

Only Python steps exist. There is no JavaScript workflow step; JavaScript on records is a record action, which is a different surface with different wiring.

Directory layout#

src/
  automationSteps/
    <stepDirectory>/
      config.json      # the step declaration — required
      script.py        # the step body — required

Steps are discovered by directory convention. They are not listed anywhere in kizen.json. (The actions array inside a setup assistant is for record actions, not steps.)

script.py is packaged verbatim — unlike JavaScript surfaces, Python step bodies are not minified. Comments and formatting survive into the published step, so in-file comments are a legitimate place to document the step's contract.

config.json — step-level fields#

Full example, referenced field by field below:

{
  "name": "Send Notification",
  "api_name": "send_notification",
  "action_description": "Posts a plain-text message to the given channel using the connected workspace. Channels may be given by name (#general) or by id.",
  "runtime": "python 3.13",
  "secrets": ["api_key"],
  "when": "Boolean({{config.enableNotifications}})",
  "inputs": [
    {
      "name": "channel",
      "label": "Channel",
      "data_type": "string",
      "required": true,
      "input_source": "variable",
      "hint_field_name": "channel",
      "hint_related_object_field_name": null
    },
    {
      "name": "message",
      "label": "Message",
      "data_type": "string",
      "required": true,
      "input_source": "variable"
    }
  ],
  "outputs": [
    {
      "name": "delivered_at",
      "label": "Delivered At",
      "data_type": "datetime",
      "required": false,
      "input_source": "object_field",
      "hint_field_name": "last_notified_at",
      "hint_related_object_field_name": null,
      "conflict_resolution": "overwrite",
      "create_field_options": false
    }
  ]
}

Four step-level keys older configs carry — action_type, script_alias, plugin_description and overall_description — are removed from the publish contract: each one now fails the build.

The fields with a rule in Build-time validation — runtime, secrets, the shape of inputs/outputs, and the removed keys — are checked before anything is sent. api_name is checked too, by the structure/api-name-format, structure/duplicate-api-name and structure/reserved-api-name rules (see api_name). name, action_description, when and step_history_template are not checked at build time. Publish does reject an empty action_description, and the packager sends "" when the key is absent, so set it on every step.

name#

The label a workflow author sees in the action picker after choosing your plugin. Free text; make it a verb phrase describing what the step does ("Send Notification", "Read Data"), not the plugin name — the plugin name is already shown.

api_name#

The step's real primary key. After publish, a step is resolved by the pair (plugin app, action_step_api_name), and that pair is the only uniqueness constraint. Everything downstream — the wired step inside a saved workflow, config lookup at run time — keys on it.

Always set it explicitly. When omitted, the packager derives it from the directory name by lowercasing it, collapsing hyphen and whitespace runs to _, then dropping characters outside [a-z0-9_]. That collapses camelCase — sendNotification becomes sendnotification — and turns send-notification into send_notification. Underscores are preserved, so send_notification comes through unchanged. The reason to set it anyway is that a later directory rename silently changes the key — and the key is the step's primary identity.

It must not be a reserved name. The packager carries a list of 46 names that publish rejects with "API Name is reserved" — field type names (files, email, status, …), default Contact fields (first_name, tags, …), default object fields (name, owner, stage, …) and notes. A step whose api_name (explicit or derived from the directory) is on that list fails the build with structure/reserved-api-name. Prefix it (example_plugin_status) instead.

Renaming a published api_name is a breaking change: existing workflows that reference the old name hard-fail at run time with a "config not found" error. Treat it as immutable once shipped (see release and publish).

plugin_description#

Removed. Setting it — or overall_description, the name it used to be packaged under — fails the build with automation-step/removed-field. Delete the key.

The plugin-level blurb above the action picker is now authored once, as the top-level agentic_description in kizen.json. Step-specific text belongs in action_description.

action_description#

The per-step description, shown after the workflow author selects this action. This is the field that should carry step-specific detail: what it does, what the inputs mean, what it writes.

This is the highest-leverage documentation surface a step has — it is the only text the workflow author reads before wiring the step up. Use it to state input semantics and any asymmetry (for example, that a reference-typed value only survives a round trip if it is a valid id for the target field).

action_type#

Removed. An api-name string that older configs set and nothing ever read at runtime. It is no longer part of the publish contract, and setting it fails the build with automation-step/removed-field. Delete it; the step's identity is api_name.

runtime#

Selects the Python runtime. Two runtimes exist, Python 3.13 and Python 3.12, and the authored string must normalize to python-3-12 or python-3-13 or the build fails with automation-step/runtime. A runtime that does not exist, such as "python 3.7", fails there.

Omitting runtime gives you Python 3.13. The packager's fallback is python-3-13, applied both when the field is absent and when config.json itself is missing. The fallback was python-3-12 before @kizenapps/packager 0.8.0, so a step that never declared a runtime moves to 3.13 the next time the plugin is published — pin "python 3.12" explicitly if the script depends on it.

Normalization replaces whitespace and dots with hyphens, and it runs before the check, so every spelling that collapses to the packaged value is accepted: "python 3.13", "python-3-13" and "python-3.13" all package as script_runtime: "python-3-13", and the 3.12 spellings likewise. Author the spaced form — "python 3.13" — which is what the rule's own error message names.

secrets#

An array of bare integration-secret names this step is allowed to read:

"secrets": ["api_key", "connection_json"]

Each name must also appear in the manifest's base_config.secrets. A step secret that is not declared there fails the build with automation-step/undeclared-secret, and the publish pipeline checks the same subset rule again. See auth, secrets and services.

Declaring a secret here is what gets its value into the step's secrets dict at run time. A secret the plugin declares but this step does not list is not readable from this step.

Inside the script the key is namespaced — see Secrets.

when#

A template expression over install config that gates whether the step is offered at all:

"when": "Boolean({{config.enableNotifications}})"

Same scoping rules as other artifact configs: {{config.key}} for business-level setup-assistant values and {{userConfig.key}} for user-level ones. (Setup-assistant-internal when expressions use bare {{key}} — a different scope; see setup assistants.)

A when on a step makes the host load install config before evaluating visibility (the packaged block_loading_for_setup flag). That is expected, not a defect.

A missing config key resolves to null, so the expression evaluates false and the step silently disappears from the picker. Setup-assistant default values reach when evaluation but do not reach this.config/the runtime — do not rely on a default to keep a step visible if the user never saved the assistant.

step_history_template#

The publish API accepts this as an optional step-level string — but the packager never reads it, so a value in config.json is dropped during packaging and never reaches the publish payload. There is no authoring path to it from a plugin repo today, and no template syntax to document. Setting it is harmless and inert — and since no authored value can reach the backend, its rendering semantics in workflow run history are moot from a plugin's point of view.

script#

Ignored. A "script" key in config.json is not a filename override and not an inline script body — the packager unconditionally reads script.py from the step directory. Some existing configs carry "script": "script.py"; it is decoration.

Authored → packaged rename map#

Two fields are renamed on the way from your config.json into the published step definition. This matters when reading platform-side documentation, publish payloads, or the workflow-builder API, which all use the packaged names.

plugin_description no longer maps to overall_description: both names are rejected at build time, and the plugin-level blurb is authored as agentic_description in kizen.json instead.

The packaged collection itself is automation_action_configs.

Which authored fields are read, and where#

Parameter reference (inputs and outputs)#

Inputs and outputs share one shape. Outputs add conflict_resolution and create_field_options; both on an input fail the build with automation-step/output-only-option.

The fields with a rule in Build-time validation — data_type, input_source, conflict_resolution, create_field_options, output_target and script_alias — are checked before anything is sent. name must not be a reserved name (structure/reserved-api-name; see name). label, required and the two hint_* keys pass through unvalidated; hint, allowed_values and default are dropped at publish altogether.

{
  "name": "channel",
  "label": "Channel",
  "data_type": "string",
  "required": true,
  "input_source": "object_field",
  "hint_field_name": "channel",
  "hint_related_object_field_name": null,
  "allowed_values": ["general", "alerts"],
  "default": true
}

name#

The runtime accessor. inputs.<name> reads an input; outputs.<name> = value writes an output. Publish requires an api-name-shaped string (lowercase letters, digits and underscores, starting with a letter or underscore); the build does not check the format.

The build does check that name is not a reserved name: files, email, status, name, first_name, owner, notes and the rest of the same list as api_name fail with structure/reserved-api-name. Prefix it (s3_files) and rename inputs.<name> / outputs.<name> in script.py to match — the runtime binds by name.

Output names are validated on write-back: assigning to a name that is not a declared output raises inside the runtime and fails the step.

label#

What the workflow author sees next to the mapping control in the builder. Free text.

hint#

Dropped at publish, exactly like default. It is not in the fixed parameter field set the publish endpoint accepts, and the workflow builder reads only hint_field_name, so a hint is discarded server-side without a warning and never shown to anyone.

Format expectations that do not fit the label belong in the step's action_description, which the builder does render, or in the plugin's own documentation. Some existing configs carry a hint; it is decoration.

data_type#

The variable type of the value, and the one field every parameter must carry. See data_type reference — this is the field most likely to be set wrong, because the valid values are variable type names, not custom-field type names. A missing, non-string, or out-of-enum value fails the build with automation-step/data-type.

required#

Boolean, default false.

• On an input: pre-execution validation refuses to run the step when a required input is unmapped or resolves to null, and the builder blocks saving the workflow with it unmapped. A required input is therefore guaranteed present in the script — read it directly with inputs.<name>.
• On an output: marks the mapping as mandatory in the builder. It does not force your script to assign the output.

input_source#

Where the value comes from (inputs) or goes to (outputs). Publishable values:

A value outside this table fails the build with automation-step/input-source, and so does null; omit the key instead.

input_source is used for both inputs and outputs — outputs use the same key, not output_target.

One more source exists in the builder but cannot be pre-declared: business_plugin_config. When a workflow author picks it, the platform injects the plugin's install config as a JSON string under that input's name (the cleaned config, when present). Your script must json.loads it.

allowed_values#

An array of literals intended to constrain a static_value input.

It is dropped at publish and never reaches the builder. Publishing accepts a fixed field set for step parameters — name, label, data_type, required, input_source, hint_field_name, hint_related_object_field_name, conflict_resolution, create_field_options — and silently discards anything else. allowed_values is not on that list, so it is stripped server-side without a warning: the workflow author sees a free-text control, not a picker, and nothing validates their input against your list.

Keep it in config.json if it documents intent for the next maintainer, but do not design around it. The value your script receives is whatever the author typed, so the unexpected-value branch is not an edge case — it is the normal path:

mode = inputs.failure_mode

if mode == "retry":
    ...
elif mode == "skip":
    ...
else:
    # Reachable: nothing constrains this value at any layer.
    raise ValueError(f"Unknown mode {mode!r}. Expected 'retry' or 'skip'.")

default#

Intended to pre-fill the builder's mapping control (most often on a boolean static-value input).

It is dropped at publish too, by the same fixed field set described above. It does not pre-fill the builder, and it does not substitute a value at run time for an input the author left empty. An input you expected to default to True arrives as whatever the empty control produced.

If a step needs a default, apply it in the script. An input the author left unconfigured arrives as None, so compare against None:

retry_on_conflict = inputs.retry_on_conflict
if retry_on_conflict is None:
    retry_on_conflict = True

hint_field_name#

The auto-mapping hook, and the reason well-authored steps feel pre-configured.

When the workflow author selects your action, the builder walks your declared inputs and outputs and, for each one carrying a hint_field_name, looks for a field on the workflow's object whose api name matches, and pre-populates the mapping. It works for variable sources too — the builder pre-selects the variable with that name.

Two properties to design around:

1. Matching is by canonical api name. "hint_field_name": "email" pre-maps to a field whose api name is email. Pick the names a typical business actually uses (name, email, mobile_phone), not your internal vocabulary.
2. There is no type check on the hint. The pre-fill happens whether or not the field's type is compatible with your data_type. A wrong data_type therefore looks fine when the hint happens to match a field, and only fails later — at workflow save — with "X" is not a valid choice. The manual dropdown, by contrast, does enforce variable-type compatibility, which is why the same step can work when mapped by hand and break when auto-mapped.

hint_related_object_field_name#

The related-object counterpart of hint_field_name, used with input_source: "related_object_field". Set it to null when unused; keeping the key present with an explicit null is the house style.

script_alias#

Removed, on parameters and at step level alike. The Python runtime always bound inputs.<name> by the parameter's name and never consulted script_alias; the key is now rejected, so a config that still carries it fails the build with automation-step/removed-field.

Older configs set it on every input and output. Delete all of them — nothing referenced it, so there is nothing to migrate.

conflict_resolution (outputs only)#

How the write-back reconciles with the field's current value.

The builder surfaces overwrite, update_if_blank, and add_only. The other two are accepted by the platform.

A value outside this table fails the build with automation-step/conflict-resolution; null and an absent key are both fine. On an input the key itself is an error — automation-step/output-only-option.

create_field_options (outputs only)#

Boolean. When true, writing a value that does not match an existing dropdown option creates the option rather than failing. Default false.

Leave it false unless the step's whole purpose is to seed an option list from an external system — a typo in an upstream feed becomes a permanent dropdown option otherwise.

A non-boolean value fails the build with automation-step/create-field-options, and the key on an input fails with automation-step/output-only-option.

output_target — do not use#

Not a field. Some published configs carry "output_target": "object_field" on outputs; it used to be silently dropped at publish and now fails the build with automation-step/output-target. The real key on outputs is input_source, exactly as on inputs.

Build-time validation#

Step config.json files are validated by the packager, so a wrong value is a build error rather than a step that publishes and then misbehaves in the workflow builder. npx --yes @kizenapps/cli build and every dev rebuild run the whole rule set and print the failing rule id along with the step and the parameter at fault.

Every rule is an error — there are no warnings in this set, and none of them can be waived.

Where each gate runs#

• Build and dev. data_type is validated against the type list bundled in the packager, and every other rule against the fixed lists above.
• PR check and publish. Both the PR's Plugin Validation check and the deploy re-validate data_type against the live list read from the first publishable environment in the manifest's release_environments (go if none resolves), via GET /api/automation2/automations/metadata, keyed by variables.variable_from_field. If that list cannot be read the publish fails as a pipeline problem, not a plugin problem: retry the publish rather than editing the step.
• Already-published apps are never re-validated. An app that shipped with action_type, script_alias, or a files data_type keeps running exactly as it did. Only the next publish is gated — which is also when those keys have to come out.

data_type reference#

data_type names a variable type, not a custom-field type. The enum is nine values, and every input and output must carry one of them, as a string — anything else fails the build with automation-step/data-type:

string | boolean | number | date | datetime | phone_number | employee | entity | uuid

There's no file value. A parameter that will be mapped to a files field is declared as string - see Types that are rejected.

Wire encoding#

Values cross the boundary into the runtime as a tagged envelope, {"t": <type code>, "v": <value>}, and are rebuilt into typed Python objects before your script runs. You never handle the envelope directly, but the type codes explain what you receive:

The italicised rows are not data_type values you can write in config.json — they are codes the platform derives from the field or variable an author maps. Only the nine named values above are authorable, and the build rejects anything else.

Note the near-collision between employee and entity: employee is the bare code e, while entity is the parameterized e<object_id>. They are distinguished by the angle brackets, not by the letter.

Scalar vs reference types — the asymmetry to design for#

string, boolean, number, date, datetime, and uuid round-trip unmodified: what you read is what you can write back.

employee, entity, and phone_number are asymmetric. On the read side your script sees a plain scalar — an id, or an E.164 string. On the write side the platform re-hydrates a record or re-validates the number, so the value only survives if it is a valid id or number for the target field. Writing an entity id belonging to a different object than the destination relationship field expects fails at write-back, after your script has already succeeded.

A practical consequence for employee: an id is rarely what an external system wants. Resolve it:

resp = kizen.api.get(f"/team/{inputs.owner}")
resp.raise_for_status()
owner_email = resp.json().get("email")

FieldOption#

Dropdown-style values deserialize to a FieldOption object with name, order, and code. It is UUID-compatible, so it interpolates and compares as the option id where an id is expected — but when you want the human-readable label, read .name:

state_label = inputs.state.name       # dropdown option label

Do not assume a bare string from a dropdown-backed input.

Stage#

Pipeline stage values deserialize to a Stage object:

Types that are rejected#

Custom-field type names are the mistake authors reach for, and they no longer get past the build. files, file, integer, decimal, money, text, email and emails all fail with automation-step/data-type, naming the step and the parameter:

file and files have no valid equivalent because data_type enumerates variable types, and no variable type carries a file. Declare string: a string parameter mapped to a files field arrives in the script as a list of file objects, not a str, so handle it as a list.

Before the build gate, these values published and surfaced only later — "No Options" in the builder's field dropdown, then "X" is not a valid choice at workflow save. Those remain the symptoms for an app published before the gate; a build that passes today cannot produce them.

The gate checks the enum, not whether the type you picked matches the field you meant, so still wire every new step into a workflow and save it before calling it done.

What the workflow builder user sees#

Understanding the builder UI is what separates a step that feels native from one that feels like a form to fill in.

1. Plugin description. The plugin's manifest agentic_description is rendered above the action picker — once, plugin-wide. No step field feeds it any more; an app published before the fields were removed shows no plugin-level description until it republishes with agentic_description.
2. Action select. Every step your plugin publishes, by name, that passes its when gate.
3. Action description. Once an action is selected, its action_description appears.
4. Pre-configured input and output lists. This is the key difference from a generic code step: the input and output rows are fixed. The workflow author maps each declared row to a field, a variable, or a static value — they cannot add rows, delete rows, or rename them. Your config.json is the form.
5. Auto-mapping on selection. The moment the action is picked, the host tries to pre-populate every mapping by matching hint_field_name / hint_related_object_field_name against the workflow object's field names. Good hints mean the step arrives already wired.
6. Script and runtime are copied at selection time. The step snapshots script_runtime when the author picks the action; the script body itself is resolved from the currently-published plugin config at run time, never from user input. A workflow author cannot edit your Python.

Validation blocks saving the workflow on:

• a required input left unmapped,
• a mapped field or secret that has since been deleted,
• the plugin no longer being installed — the step renders "The Plugin is no longer available" and is marked invalid. Uninstalling a plugin therefore invalidates every workflow step that used it; it does not silently no-op.

Removing or renaming a published step is the destructive change here: existing workflow steps referencing it hard-fail at run time with a config-not-found error. Adding a step, or adding an optional input, is safe.

The Python runtime contract#

Execution environment#

• Runtimes: Python 3.13 and Python 3.12 (3.13 is the packager's default when runtime is omitted).

• Isolation: each execution runs in a fresh container with /tmp wiped beforehand, and the user code runs in a subprocess.

• Memory: 1 GB. An out-of-memory kill is surfaced as a step error.

• Execution time: the hard ceiling is 55 seconds. Past it the execution environment kills the step outright — your code gets no exception, no chance to clean up, and no control over the message the workflow records. Design to about 30 seconds instead. That is not the limit, it is the working target: it leaves a full retry's worth of headroom inside the real ceiling, so a single slow upstream call or one Retry-After sleep does not push the run into the kill. A step that cannot finish in 30 seconds should do less per run.

• Identity: the step runs as the plugin's own service account for that business — a non-human employee, authenticated by a short-lived signed API key. That service account is the actor recorded on timeline events your step causes.

• Pre-installed packages beyond the standard library: requests, psycopg, pymysql, oracledb, sqlalchemy, paramiko, pypdf, fpdf2, lxml, numpy, msgpack, pyjwt, bcrypt, pytz, dateutil, tzdata. There is no way to add a dependency; a step that needs something else must call it over HTTP or fall back to the standard library (http.client keeps a step zero-dependency).

Injected globals#

The script body is exec'd with four names already bound. They are runtime-injected, so linters will report them as undefined — those warnings are expected and safe to ignore.

There is no main() and no entry-point convention: the module body is the step. Top-level statements execute in order.

inputs#

Attribute access on a typed object built from your declared parameters.

An unmapped optional input is None. Every declared input is present on inputs; one the workflow author left unmapped is sent as None, so plain attribute access never raises. Check for None wherever the script needs a value:

channel = inputs.channel                  # required: always mapped
tag = inputs.connection_secret_tag        # optional: None when unmapped

getattr(inputs, "name", fallback) does not apply a default — the attribute exists, so it returns None, not fallback. Use an explicit is None check instead.

outputs#

Assign attributes to write results back:

outputs.delivered_at = datetime.now(timezone.utc)
outputs.message_id = str(response_body["ts"])

The object validates names: assigning to an undeclared output raises and fails the step. Outputs are written back after the script completes successfully — a step that raises writes nothing, even for outputs it had already assigned. There are no partial writes.

Each written value goes through the destination's conflict_resolution, under a plugin timeline initiator.

outputs.log(message)#

Appends to the step's run log, visible in the workflow's execution history alongside the resolved inputs, the outputs, and a summary of the HTTP requests the step made.

Limits: 8 KB per message, 100 KB total per execution. Overflow is discarded silently — a step that logs a full API response per loop iteration will lose the log lines that actually mattered.

outputs.log(f"Resolved channel {channel!r} to id {channel_id}")

Log identifiers and status codes, never secret values. Log content is stored in run history and visible to anyone who can view the workflow.

HTTP request logging#

Outbound HTTP is tracked automatically and summarized in the run history: request count, plus per-request detail for the first 100 requests, with the request body truncated at 8 KB, the response body at 1 KB, and headers redacted. This is free instrumentation — a step that mysteriously fails against a third-party API can usually be diagnosed from run history without adding a single log line.

secrets (inside a step)#

secrets is a flat dict, keyed by the namespaced name {plugin_api_name}__{secret_name} — even though both kizen.json and the step's config.json declare the bare name:

// config.json declares the bare name
"secrets": ["api_key"]

# the runtime key is namespaced
auth_header = f"Basic {secrets['example_plugin__api_key']}"

Hardcoding the full key is fine when the plugin's api_name is fixed. When the same source is published under more than one api_name (per-environment variants, sandbox builds that suffix the api_name), match on the suffix instead:

key = next((k for k in secrets if k.endswith("connection_json")), None)
if not key:
    raise ValueError("No connection_json secret is configured for this plugin.")
connection_raw = secrets[key]

Secret values are obfuscated in the step's error output, but nothing stops your own outputs.log from leaking one. Never log a secret, and never echo a response that reflects your credentials back.

kizen.api — calling Kizen and declared services#

kizen.api is a pre-authenticated HTTP client injected into every step. It is a thin wrapper over a requests session whose headers already carry the plugin service account's business id, user id, and signed API key. You never handle credentials.

Methods#

kizen.api.get(path, params=None, headers=None, **kwargs)
kizen.api.post(path, data=None, json=None, headers=None, **kwargs)
kizen.api.patch(path, data=None, json=None, headers=None, **kwargs)
kizen.api.put(path, data=None, json=None, headers=None, **kwargs)
kizen.api.delete(path, **kwargs)
kizen.api.head(path, **kwargs)
kizen.api.options(path, **kwargs)

Because it is a requests session, the usual keyword arguments (timeout, params, data, json, headers) work as you expect.

Response object#

Every method returns a real requests.Response:

Nothing raises automatically on a non-2xx: kizen.api returns the response and leaves the decision to you.

Relative paths → the Kizen REST API#

A path starting with / is resolved against the platform API base, so worker code calls exactly the same public REST API documented in platform API — authenticated as the plugin's service account:

# read a team member
resp = kizen.api.get(f"/team/{inputs.owner}")

# update a record
resp = kizen.api.patch(
    f"/records/example_object/{record_id}",
    json={"fields": [{"name": "example_plugin_status", "value": "active"}]},
)

# search
resp = kizen.api.post(
    "/records/example_object/search",
    json={"field_names": ["name", "email"], "query": [...]},
)

Absolute URLs pass through unchanged, but there is no reason to send one — use plain requests for external hosts so the authenticated session is never pointed at a third party.

Service-proxy paths → your declared services#

Python steps can call the plugin's declared services[] through the generic proxy. Build the path by hand — there is no getServiceUrl helper in Python — and the proxy resolves it against the service's base_service_url and injects auth server-side:

BASE_URL = "/external-integrations/proxy/example_plugin/example_service"

resp = kizen.api.post(
    f"{BASE_URL}/messages",
    json={"channel": channel_id, "text": inputs.message},
)

The shape is /external-integrations/proxy/{plugin_api_name}/{service_name}/{remaining path}; everything after the service name is appended to the service's base URL.

Proxy behavior a Python step must handle:

• The proxy wraps the upstream response. The upstream payload is nested — extract it: body = resp.json().get("body") or {}.
• A disconnected OAuth service returns HTTP 503 from the proxy. That is a configuration problem, not a transient one: raise with a message telling the user to reconnect the integration in the plugin's setup assistant.
• API-level errors can hide behind HTTP 200. Several vendors always return 200 and put the real outcome in the body. Check the body's own success flag, not just resp.ok.
• The proxy sends exactly three headers upstream, and none of them are yours. It forwards Accept: application/json (forced, not dropped — you cannot negotiate a different response content type), the injected Authorization, and on POST/PUT/PATCH/DELETE with a body, Content-Type: application/json. Your own headers do not survive, including Content-Type: the body is re-serialized as JSON regardless of what you asked for. A vendor endpoint that only accepts application/x-www-form-urlencoded cannot be called through the proxy with a form body — call it with plain requests instead, or have the vendor accept JSON.
• Service scope. A service declared scope: "service-account-only" can be called only by plugin code steps (the plugin service account); JavaScript surfaces get 403. The inverse, user-account-only, blocks code steps. See auth, secrets and services.

kizen.api vs bare requests#

Never mix the two for the same host.

Rate limits and retries#

Three graduated patterns, in increasing order of rigor. Pick the simplest one that fits the API you are calling, and remember the whole step shares one execution budget: a sleep is spent from the same seconds your HTTP calls need, against a 55-second hard ceiling.

1. Retry loop with Retry-After and exponential fallback#

For an API that returns a plain integer Retry-After:

import time
import requests

MAX_RETRIES = 3
auth_header = f"Basic {secrets['example_plugin__api_key']}"


def call_with_retry(url):
    attempt = 0
    response = requests.get(url, headers={"Authorization": auth_header}, timeout=10)

    while response.status_code == 429 and attempt < MAX_RETRIES:
        attempt += 1
        retry_after = response.headers.get("Retry-After")
        delay = int(retry_after) if retry_after and retry_after.isdigit() else 2**attempt

        outputs.log(f"{url} -> 429; retry {attempt}/{MAX_RETRIES} after {delay}s")
        time.sleep(delay)

        response = requests.get(url, headers={"Authorization": auth_header}, timeout=10)

    return response

2. Budget-aware single retry#

The version to prefer when the upstream can ask for a long wait. It retries once, and only if the requested wait fits the execution budget; otherwise it fails fast with a message the workflow owner can act on:

import time

MAX_SLEEP_SECONDS = 20   # keep well inside the step's execution budget


def request_with_retry(method, url, **kwargs):
    """Call the proxy, retrying once on 429 when Retry-After fits the budget."""
    try:
        resp = method(url, **kwargs)
    except Exception as exc:
        raise Exception(
            "Example Plugin error: service_unavailable — the proxy is unreachable. "
            f"Check the integration connection in the plugin setup assistant. ({exc})"
        )

    if resp.status_code == 429:
        retry_after = resp.headers.get("Retry-After")
        if not retry_after or int(retry_after) > MAX_SLEEP_SECONDS:
            raise Exception(
                "Example Plugin error: rate_limited — the API rate limit was reached. "
                f"Retry after {retry_after or 'unknown'} seconds."
            )
        time.sleep(int(retry_after))
        resp = method(url, **kwargs)

    if not resp.ok:
        detail = resp.json().get("error", "") if resp.content else ""
        raise Exception(
            f"Example Plugin error: proxy_error — HTTP {resp.status_code} from the proxy"
            f"{f': {detail}' if detail else ''}"
        )

    # The proxy wraps the upstream payload.
    return resp.json().get("body") or {}

Call it by passing the bound method: request_with_retry(kizen.api.post, url, json=payload).

3. RFC-2822-aware Retry-After#

Retry-After may legally be an HTTP-date rather than a number. Handle both:

import time
from email.utils import parsedate_to_datetime
from datetime import datetime, timezone

DEFAULT_RETRY_DELAY = 1


def parse_retry_after(value):
    """Parse Retry-After (seconds or HTTP-date) into a delay in seconds."""
    if value is None:
        return DEFAULT_RETRY_DELAY
    try:
        return int(value)
    except (ValueError, TypeError):
        pass
    try:
        retry_at = parsedate_to_datetime(value)
        delay = (retry_at - datetime.now(timezone.utc)).total_seconds()
        return max(delay, DEFAULT_RETRY_DELAY)
    except Exception:
        return DEFAULT_RETRY_DELAY

Helpers are per-step, by design#

Steps are isolated units. There is no shared module, no import path between steps, and no way to factor a helper out of one script.py into another. Copying request_with_retry into every step in a plugin is the correct pattern, not a smell — the only factoring available is local functions inside one file. (This is specific to Python steps; the plugin's JavaScript scripts can share code — 19-sharing-code-between-scripts.md.)

Database-connector steps#

A recurring shape: a plugin whose whole job is to run SQL against a customer's database. The pattern below is worth copying wholesale.

One JSON secret, optionally keyed by environment#

Rather than four secrets for host/port/user/password, declare one secret holding a JSON document. That also gives you a free multi-environment mechanism: the document is either flat, or nested by environment tag, and an optional connection_secret_tag input selects the nesting key.

// flat
{ "host": "db.example.com", "port": 5432, "user_name": "reporting", "password": "…" }

// keyed by environment
{
  "production_db": { "host": "…", "port": 5432, "user_name": "…", "password": "…" },
  "staging_db":    { "host": "…", "port": 5432, "user_name": "…", "password": "…" }
}

connection_secret_tag is not a manifest or platform field — it is a conventional input name. Declare it as an optional string input.

Normalize smart quotes before parsing#

Users paste connection JSON out of documents and chat clients, which substitute curly quotes. json.loads rejects them with an unhelpful error. Translate first:

SMART_QUOTE_MAP = str.maketrans({
    "“": '"', "”": '"', "„": '"', "‟": '"',
    "‘": "'", "’": "'", "‛": "'",
})
connection = json.loads(raw_secret.translate(SMART_QUOTE_MAP))

Split read and write into two steps, and guard the read#

A read step and a write step are separate published steps with separate api_names. The read step rejects anything that is not a SELECT, both with a regex pre-check and by forcing the session read-only; the write step deliberately has no guardrail.

The regex is a guardrail, not a security boundary — the query text is executed verbatim, so a step that takes a query from a workflow variable is by construction an arbitrary-SQL surface. Say so in the action_description, restrict the database user's grants, and never assemble SQL from record data.

Complete read step#

src/automationSteps/dbRead/config.json:

{
  "name": "Read Data",
  "api_name": "db_read",
  "action_description": "Connects to the configured database and runs a read-only query. SELECT only — write statements are rejected. With Return Single Value on, the query must return exactly one row and one column.",
  "runtime": "python 3.13",
  "secrets": ["connection_json"],
  "inputs": [
    { "name": "database", "label": "Database", "data_type": "string", "required": true, "input_source": "static_value" },
    { "name": "query", "label": "Query", "data_type": "string", "required": true, "input_source": "variable" },
    { "name": "return_single_value", "label": "Return Single Value", "data_type": "boolean", "required": true, "input_source": "static_value" },
    { "name": "connection_secret_tag", "label": "Connection Secret Tag", "data_type": "string", "required": false, "input_source": "static_value" }
  ],
  "outputs": [
    { "name": "result", "label": "Result", "data_type": "string", "required": true, "input_source": "variable", "conflict_resolution": "overwrite", "create_field_options": false }
  ]
}

No default on return_single_value: the key is dropped at publish, and the input is required: true, so the builder always supplies a value and the script reads inputs.return_single_value directly. An optional flag would arrive as None when unmapped, so it would need an explicit if flag is None: flag = True.

src/automationSteps/dbRead/script.py:

# Example Plugin · Agentic Workflow Step · Read Data
#
# Runs a read-only query against the database described by the connection_json secret.
# The secret is either a flat {host, port, user_name, password} document or a map of
# those documents keyed by environment tag, selected with the connection_secret_tag input.

import json
import re

import psycopg
from psycopg.rows import dict_row

WRITE_STATEMENT = r"^\s*(INSERT|UPDATE|DELETE|DROP|ALTER|TRUNCATE|CREATE|GRANT|REVOKE|COPY|CALL|DO)\b"
REQUIRED_KEYS = ("host", "port", "user_name", "password")

SMART_QUOTE_MAP = str.maketrans({
    "“": '"', "”": '"', "„": '"', "‟": '"',
    "‘": "'", "’": "'", "‛": "'",
})


def load_connection():
    """Resolve the connection document, tolerating smart quotes and the nested-by-tag shape."""
    key = next((k for k in secrets if k.endswith("connection_json")), None)
    if not key:
        raise ValueError("No connection_json secret is configured for this plugin.")

    document = json.loads(secrets[key].translate(SMART_QUOTE_MAP))

    tag = inputs.connection_secret_tag
    if tag:
        if tag not in document:
            raise ValueError(f"Connection Secret Tag {tag!r} is not present in the connection secret.")
        document = document[tag]

    missing = [k for k in REQUIRED_KEYS if k not in document]
    if missing:
        raise ValueError(f"The connection secret is missing required key(s): {', '.join(missing)}")

    return document


connection_info = load_connection()
query = inputs.query

# Guardrail: reject obvious write statements before opening a connection. The session is also
# forced read-only below — neither is a substitute for restricting the database user's grants.
if re.match(WRITE_STATEMENT, query, re.IGNORECASE):
    raise ValueError("Write statements are not allowed by this step. Use the Write Data step.")

outputs.log(f"Connecting to {connection_info['host']}:{connection_info['port']}/{inputs.database}")

try:
    with psycopg.connect(
        host=connection_info["host"],
        port=connection_info["port"],
        dbname=inputs.database,
        user=connection_info["user_name"],
        password=connection_info["password"],
        row_factory=dict_row,
        connect_timeout=10,
    ) as connection:
        with connection.cursor() as cursor:
            cursor.execute("SET SESSION CHARACTERISTICS AS TRANSACTION READ ONLY")
            cursor.execute("SET default_transaction_read_only = on")
            cursor.execute(query)
            rows = cursor.fetchall()

            if not rows:
                outputs.log("Query returned no rows.")
                outputs.result = ""
            elif inputs.return_single_value:
                if len(rows) != 1 or len(rows[0]) != 1:
                    raise ValueError(
                        "Return Single Value is on, but the query returned "
                        f"{len(rows)} row(s) and {len(rows[0]) if rows else 0} column(s)."
                    )
                outputs.result = str(next(iter(rows[0].values())))
            else:
                outputs.log(f"Query returned {len(rows)} row(s).")
                outputs.result = str(rows)

except psycopg.Error as exc:
    raise ValueError(f"Database error: {exc}")

The stringification limitation is real and worth stating in your action_description. Because a parameter cannot express a list or a structured value, multi-row results come back as the Python repr of a list of dicts in a single string output. Downstream steps have to parse that. Prefer return_single_value mode, or a query that already aggregates to one value, whenever the result feeds another step.

Failing cleanly#

Raising is how a step fails. There is no error callback, no toast, no partial success. An uncaught exception marks the step failed, writes no outputs, and puts the message and traceback in the workflow's run history — which is what the workflow owner reads when they investigate.

That makes the exception message a user-facing string. Write it accordingly:

# Bad — the reader learns nothing.
raise Exception(resp.text)

# Good — what failed, why, and what to do.
raise Exception(
    f"Example Plugin error: channel_not_found — no channel named '#{name}' exists. "
    "Check the name, and make sure the bot has been invited to private channels."
)

Conventions that make run history readable:

• Prefix with the plugin and a machine-ish code, then a human sentence: "Example Plugin error: {code} — {what the user should do}".
• Wrap connection failures with the remedy. A proxy 503 means the integration is disconnected; say "reconnect the integration in the plugin setup assistant," not "503."
• Validate at the boundary, not everywhere. Check external API responses and secret documents; do not re-check what the platform already guarantees (a required: true input is present).
• Distinguish hard from soft failures deliberately. A step that does three things can raise on the essential one and outputs.log a warning for a cosmetic one. Decide per operation, and say which is which in the action_description.
• Do not swallow. A bare except: pass turns a failed step into a successful one that wrote nothing — the worst outcome, because the workflow continues on stale data.

The three failure shapes worth being able to recognize in run history:

# 1. Explicit raise — the common, intentional failure.
raise ValueError("Example Plugin: start page must be at least 1.")

# 2. Unhandled non-2xx — check and raise with the status and body.
response = requests.get(url, timeout=10)
if not response.ok:
    raise Exception(f"Example Plugin: upstream returned {response.status_code} — {response.text[:500]}")

# 3. Network timeout — always pass an explicit timeout; without one a hung
#    connection burns the whole execution budget and the step dies with no message.
requests.get(url, timeout=10)

Always pass timeout= to every outbound call. It is the difference between a step that fails in ten seconds with a clear timeout error and one that is killed mid-flight with nothing useful in the log.

Complete worked example#

A step that calls a declared OAuth service through the proxy, resolves a channel by name, posts a message, and writes the delivery timestamp back to a record field.

src/automationSteps/sendNotification/config.json:

{
  "name": "Send Notification",
  "api_name": "send_notification",
  "action_description": "Posts a plain-text message to a channel using the connected workspace. Channels may be given by name (#general) or by id. Writes the delivery timestamp back to the record.",
  "runtime": "python 3.13",
  "inputs": [
    {
      "name": "channel",
      "label": "Channel",
      "data_type": "string",
      "required": true,
      "input_source": "variable"
    },
    {
      "name": "message",
      "label": "Message",
      "data_type": "string",
      "required": true,
      "input_source": "variable"
    },
    {
      "name": "notify_owner",
      "label": "Notify Record Owner",
      "data_type": "employee",
      "required": false,
      "input_source": "object_field",
      "hint_field_name": "owner",
      "hint_related_object_field_name": null
    }
  ],
  "outputs": [
    {
      "name": "delivered_at",
      "label": "Delivered At",
      "data_type": "datetime",
      "required": false,
      "input_source": "object_field",
      "hint_field_name": "last_notified_at",
      "hint_related_object_field_name": null,
      "conflict_resolution": "overwrite",
      "create_field_options": false
    },
    {
      "name": "message_id",
      "label": "Message Id",
      "data_type": "string",
      "required": false,
      "input_source": "variable",
      "conflict_resolution": "overwrite",
      "create_field_options": false
    }
  ]
}

src/automationSteps/sendNotification/script.py:

# Example Plugin · Agentic Workflow Step · Send Notification
#
# Posts a message to a channel through the example_service proxy service. Auth is injected
# server-side by the proxy; this script never sees a token.
#
# An optional input the workflow author left unmapped arrives as None.

import time
from datetime import datetime, timezone

BASE_URL = "/external-integrations/proxy/example_plugin/example_service"
MAX_SLEEP_SECONDS = 20


def call_service(method, path, **kwargs):
    """Call the service through the proxy, retrying once on 429, and unwrap the payload."""
    url = f"{BASE_URL}{path}"
    try:
        resp = method(url, **kwargs)
    except Exception as exc:
        raise Exception(
            "Example Plugin error: service_unavailable — the integration proxy is unreachable. "
            f"Check the workspace connection in the plugin setup assistant. ({exc})"
        )

    if resp.status_code == 429:
        retry_after = resp.headers.get("Retry-After")
        if not retry_after or not retry_after.isdigit() or int(retry_after) > MAX_SLEEP_SECONDS:
            raise Exception(
                "Example Plugin error: rate_limited — the API rate limit was reached. "
                f"Retry after {retry_after or 'unknown'} seconds."
            )
        outputs.log(f"429 from {path}; sleeping {retry_after}s before one retry.")
        time.sleep(int(retry_after))
        resp = method(url, **kwargs)

    if resp.status_code == 503:
        raise Exception(
            "Example Plugin error: not_connected — the workspace is not connected. "
            "Reconnect the integration from the plugin's setup assistant."
        )

    if not resp.ok:
        raise Exception(
            f"Example Plugin error: proxy_error — HTTP {resp.status_code} from the proxy: "
            f"{resp.text[:500]}"
        )

    # The proxy nests the upstream response under "body".
    return resp.json().get("body") or {}


def resolve_channel_id(name):
    """Resolve a channel name (without the leading #) to its id, paging through results."""
    cursor = None
    while True:
        params = {"limit": 200, "exclude_archived": "true"}
        if cursor:
            params["cursor"] = cursor

        body = call_service(kizen.api.get, "/channels.list", params=params)
        if not body.get("ok"):
            code = body.get("error", "unknown_error")
            raise Exception(f"Example Plugin error listing channels: {code}")

        for channel in body.get("channels", []):
            if channel.get("name") == name:
                return channel["id"]

        cursor = body.get("response_metadata", {}).get("next_cursor")
        if not cursor:
            break

    raise Exception(
        f"Example Plugin error: channel_not_found — no channel named '#{name}' exists. "
        "Check the name, and make sure the bot has been invited to private channels."
    )


# --- main -------------------------------------------------------------------

channel_input = inputs.channel
channel_id = (
    resolve_channel_id(channel_input.lstrip("#"))
    if channel_input.startswith("#")
    else channel_input
)

text = inputs.message

# An `employee` input arrives as a team-member id (uuid.UUID). Resolve it to something the
# external system understands before sending it.
owner_id = inputs.notify_owner
if owner_id:
    owner_resp = kizen.api.get(f"/team/{owner_id}")
    if owner_resp.ok:
        owner_email = owner_resp.json().get("email")
        if owner_email:
            text = f"{text}\n(owner: {owner_email})"
    else:
        outputs.log(f"Could not resolve owner {owner_id}: HTTP {owner_resp.status_code} — sending without it.")

body = call_service(
    kizen.api.post,
    "/messages.post",
    headers={"Content-Type": "application/x-www-form-urlencoded"},
    data={"channel": channel_id, "text": text},
)

# This API always returns HTTP 200 — success lives in the body.
if not body.get("ok"):
    code = body.get("error", "unknown_error")
    raise Exception(f"Example Plugin error: {code} — the message was not delivered.")

outputs.delivered_at = datetime.now(timezone.utc)
outputs.message_id = str(body.get("ts", ""))
outputs.log(f"Delivered message {outputs.message_id} to channel {channel_id}.")

Gotchas#

• An unmapped optional input is None. Every declared input is present on inputs, so inputs.optional_thing never raises — but getattr(inputs, "optional_thing", default) returns None, not default. Apply defaults with an explicit is None check.
• data_type takes variable type names, not field type names. files, file, integer, decimal, money, text, email and emails fail the build with automation-step/data-type. Use number and string; files fields take string and arrive as a list of file objects.
• hint_field_name prefills with no type check. A wrong data_type looks correct when the hint happens to match a field name and only breaks at save time — which is why the same step can work when mapped by hand and fail when auto-mapped.
• Always set api_name explicitly. The directory-name fallback lowercases, so sendNotification collapses to sendnotification (underscores themselves are preserved), and a later directory rename silently changes the step's primary key.
• script_alias, action_type, plugin_description and overall_description are rejected. All four are out of the publish contract; a config that still carries one fails the build with automation-step/removed-field. The runtime always bound inputs by name.
• output_target fails the build. Outputs use input_source, same as inputs.
• A "script" key in config.json is ignored. script.py is always read from disk.
• allowed_values is not enforced. Any string can reach a static_value input; always handle the else branch.
• The plugin-level blurb comes from the manifest, not the step. The builder shows agentic_description from kizen.json above the action picker; step detail goes in action_description.
• Omitting runtime now means Python 3.13. It used to mean 3.12. Pin "python 3.12" explicitly if a step depends on it.
• Secrets are namespaced at runtime. config.json declares "api_key"; the script reads secrets["example_plugin__api_key"]. Suffix-match when the api_name varies across environments or sandbox builds.
• A secret must be listed in the step's secrets array, not just in base_config.secrets, to appear in the step's secrets dict.
• The proxy wraps the upstream response. resp.json()["body"] holds the real payload; a disconnected OAuth service returns 503; and some vendors report failure inside a HTTP 200 body, so check the body's own success flag.
• The proxy forwards only Content-Type. Accept and every other request header is dropped and replaced — you cannot negotiate a response format through it.
• Outputs are all-or-nothing. A step that raises writes no outputs, including ones it already assigned. There are no partial writes.
• Assigning an undeclared output name raises and fails the step.
• outputs.log silently truncates. 8 KB per message and 100 KB per execution; overflow is discarded. Logging a full response per loop iteration loses the lines that mattered.
• Budget the whole step, sleeps included. The hard ceiling is 55 seconds and 1 GB; design to 30 seconds so a retry fits. Cap any Retry-After sleep well below the budget and fail fast with an actionable message rather than sleeping into a kill.
• Always pass timeout=. A call without one can consume the execution budget and die with no usable error.
• File presigned URLs expire in about 10 minutes and reject an added Authorization header. Download at the top of the step, with plain requests.
• No shared helpers between Python steps. Each script.py is an isolated unit; duplicating a retry helper into every step is the correct pattern, not a smell. (JavaScript scripts can share code via import — 19; script.py cannot.)
• Uninstalling a plugin invalidates every workflow step that used it ("The Plugin is no longer available"), and a removed or renamed step api_name hard-fails at run time with a config-not-found error. Additive changes are safe; removals and renames are breaking.
• Step configs are validated at build time. The automation-step/* rules (Build-time validation) fail build and dev on a bad data_type, input_source, conflict_resolution, create_field_options, runtime, an undeclared secret, an output-only key on an input, output_target on an output, or a removed field. What they cannot check is whether the type you declared matches the field you meant, so still wire a new step into a real workflow and save it before calling it done.
• Already-published apps are not re-validated. An app that shipped before these rules keeps running; the gate applies to its next publish, which is when removed keys have to come out.