core/docs/STARLARK-REFERENCE.md

Starlark Reference

Armature extensions can include Starlark scripts for server-side logic. Starlark is a Python-like language designed for configuration and embedding — see the official spec.

Sandbox constraints

• No while loops — use for with bounded ranges.
• No load() — use lib.require() for library dependencies.
• Max steps: 1,000,000 bytecode operations per execution.
• No filesystem or OS access — all I/O goes through gated modules.
• Deterministic — same input produces same output (no random, no time).

Always-available modules

These modules are injected into every script with no permission required.

json

Standard Starlark JSON module.

data = json.decode('{"key": "value"}')
text = json.encode({"key": "value"})

settings

Read resolved package settings (global → team → user cascade).

val = settings.get("theme", "light")
# Returns the resolved value, or the default if unset.

The cascade respects the user_overridable flag from the package manifest. See Permissions & Groups for details.

lib

Load exported functions from library packages.

helpers = lib.require("my-utils")
result = helpers.format_date("2026-01-15")

Requirements:

• The library must be declared in your package manifest's dependencies.
• The library must be type library, status active, tier starlark.
• Circular dependencies are detected and rejected.
• Results are cached per execution (calling require twice returns the same object).

Permission-gated modules

These modules are only available if the package has the corresponding permission granted in Admin > Packages.

secrets

Permission: secrets.read

Read admin-configured secrets for this package.

api_key = secrets.get("OPENAI_KEY")  # str or None
all_keys = secrets.list()             # list of key names

Secrets are set in Admin > Packages > Secrets and scoped per package.

notifications

Permission: notifications.send

Send in-app notifications to users.

notifications.send(
    user_id,        # str — target user UUID
    title,          # str — notification title
    body="",        # str — optional body text
    type="extension.notify"  # str — notification type
)

db

Permission: db.read (queries) or db.write (mutations)

Read and write extension data tables. All tables are automatically namespaced as ext_{package_id}_{table_name}.

Read operations

# Query with filters, ordering, and pagination
rows = db.query(
    "tasks",                    # table name (without prefix)
    filters={"status": "open"}, # equality WHERE clauses
    order="-created_at",        # column name (prefix - for DESC)
    limit=50,                   # max 1000
    before={"created_at": ts},  # range: column < value
    after={"created_at": ts},   # range: column > value
    search_like={"title": "%bug%"}  # LIKE/ILIKE search
)

# Read from system views (read-only)
users = db.view("users", filters={"display_name": "Alice"}, limit=10)
channels = db.view("channels", limit=100)

# List all tables owned by this package
tables = db.list_tables()

Available views: users, channels.

Aggregate operations

# Count rows matching filters
count = db.count("tasks", filters={"status": "open"})

# Aggregate a column (sum, avg, min, max, count)
total = db.aggregate("orders", "amount", "sum", filters={"status": "paid"})
# Returns int, float, or None (if no matching rows)

# Batch multiple queries in a single call
results = db.query_batch([
    {"table": "tasks", "filters": {"status": "open"}, "limit": 10},
    {"table": "logs", "order": "-created_at", "limit": 5},
])
# Returns list of result lists. Max 10 queries per batch.
# Each query spec supports: table (required), filters, order, limit, before, after, search_like

Write operations

row = db.insert("tasks", {"title": "Fix bug", "status": "open"})
# Returns the inserted row dict (with generated id, created_at)

db.update("tasks", row_id, {"status": "closed"})
# Returns True on success

db.delete("tasks", row_id)
# Returns True on success

http

Permission: api.http

Make outbound HTTP requests.

resp = http.get("https://api.example.com/data", headers={"Authorization": "Bearer ..."})
resp = http.post(url, body='{"key": "val"}', headers={"Content-Type": "application/json"})
resp = http.put(url, body="...", headers={})
resp = http.delete(url, headers={})
resp = http.request("PATCH", url, body="...", headers={})

Response dict:

{
    "status": 200,
    "headers": {"content-type": "application/json"},
    "body": "..."   # capped at 1 MB
}

Batch requests

responses = http.batch([
    {"method": "GET", "url": "https://api.example.com/a"},
    {"method": "POST", "url": "https://api.example.com/b", "body": "{}", "headers": {"Content-Type": "application/json"}},
])
# Returns list of response dicts (same shape as individual calls).
# Individual failures return {"status": 0, "body": "error: ...", "headers": {}}.
# Max 10 requests per batch. Dispatched concurrently.

Security:

• Private/loopback IPs are blocked (SSRF protection).
• Packages can declare network_access.allow (allowlist) or network_access.block (blocklist) in their manifest.
• Max 10 redirects. 10-second timeout. 1 MB response body limit.

realtime

Permission: realtime.publish

Publish WebSocket events to subscribed clients.

realtime.publish(
    "my-channel",     # channel name
    "item.updated",   # event label
    {"id": "abc"}     # payload dict (max 7 KB)
)

The payload is automatically tagged with _pkg: package_id.

connections

Permission: connections.read

Read external connection configurations (secrets are decrypted).

conn = connections.get("postgres", "main-db")
# Returns dict with id, type, name, scope, plus flattened config fields
# Returns None if not found

all_pg = connections.list("postgres")
# Returns list of connection dicts

Connections are resolved via scope chain: personal → team → global.

workflow

Permission: workflow.access

Read workflow definitions and instances (read-only from Starlark; mutations go through the HTTP API).

defn = workflow.get_definition(workflow_id)
# Returns dict: id, name, slug, entry_mode, is_active, version, stages[]

inst = workflow.get_instance(instance_id)
# Returns dict: id, workflow_id, current_stage, status, stage_data, ...

instances = workflow.list_instances(workflow_id, status="active")
# Returns list of instance dicts

batch

Permission: batch.exec

Run multiple callables concurrently. Each callable gets its own execution thread with an independent step budget.

jira = lib.require("jira-client")
confluence = lib.require("confluence-client")

results, errors = batch.exec([
    lambda: jira.create_issue(issue_data),
    lambda: confluence.create_page(page_data),
    lambda: send_notification(user_id),
], timeout=15)

# results[i] = return value of callables[i], or None on error
# errors[i]  = None on success, or error string on failure
# All three ran concurrently.

Constraints:

• Max 8 callables per call. Dispatched concurrently via goroutines.
• timeout (optional): 1–30 seconds per branch (default 10).
• lib.require() is not available inside branch callables. Load libraries before the batch.exec call.
• batch.exec() cannot be called from within a branch (no nesting).
• print() output from branches is discarded.

---

Example: automated stage hook

A simple hook that reads a setting, queries data, and advances:

def on_run(ctx):
    threshold = settings.get("approval_threshold", 1000)
    amount = ctx["stage_data"].get("amount", 0)

    if amount > threshold:
        notifications.send(
            ctx["started_by"],
            "High-value submission",
            body="Amount %d exceeds threshold." % amount,
        )
        return {"advance": True, "data": {"needs_review": True}}

    return {"advance": True, "data": {"needs_review": False}}