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.

# Check if a runtime capability is available
if settings.has_capability("pgvector"):
    # Use native vector search
    ...

has_capability(name) returns True if the named environment capability is detected by the kernel. Detected capabilities: pgvector, workspace, object_storage, s3, postgres.

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).

permissions

Check whether a user has a specific permission.

if permissions.check(user_id, "image-gen.use"):
    # User is authorized
    ...

Returns True if the user has the permission, False otherwise (including when the user is not found). Resolves the user's groups and merges granted permissions — works for both kernel and extension-declared permissions.

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

Vector similarity search

# Find rows with the most similar embeddings (cosine distance)
rows = db.query_similar(
    "documents",        # table name
    "embedding",        # vector column name
    vector=[0.1, 0.2, ...],  # query vector (list of floats)
    limit=10,           # max results (default 10, max 100)
    filters={"active": True},  # optional equality filters
    metric="cosine",    # only "cosine" supported
)
# Returns rows ordered by ascending _distance (0.0 = identical, 1.0 = orthogonal)
# Each row dict includes an injected "_distance" float key.

Vector columns are declared as "vector(N)" in the manifest db_tables block (N = dimension, 1–4096). Storage varies by backend:

Backend	Column type	Search
Postgres + pgvector	vector(N) with HNSW index	Native <=> operator
Postgres (no pgvector)	JSONB	Go-side cosine computation
SQLite	TEXT	Go-side cosine computation

Insert vectors as lists: db.insert("docs", {"embedding": [0.1, 0.2, 0.3]}).

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.

---

files

Permissions: files.read, files.write

Store and retrieve files via the kernel ObjectStore (PVC or S3). All keys are scoped to ext/{packageID}/ — extensions cannot access each other's files.

# Store a file with optional metadata
files.put("reports/q1.pdf", pdf_bytes,
          content_type="application/pdf",
          metadata={"quarter": "Q1", "year": 2026})

# Read a file
result = files.get("reports/q1.pdf")
# result = {"content": b"...", "content_type": "application/pdf",
#           "size": 12345, "metadata": {"quarter": "Q1", "year": 2026}}

# Metadata only (no content transfer)
meta = files.meta("reports/q1.pdf")

# List files by prefix
entries = files.list(prefix="reports/", limit=50)
# entries = [{"name": "reports/q1.pdf", "size": 12345, "content_type": "..."}]

# Check existence
if files.exists("reports/q1.pdf"):
    files.delete("reports/q1.pdf")

# Bulk delete
files.delete_prefix("temp/")

Notes:

• content accepts string or bytes; get() returns bytes.
• Maximum file size: 50 MB (configurable via EXT_FILES_MAX_SIZE).
• Metadata is stored as a companion JSON object, not in the file itself.
• files.list() automatically filters out internal metadata companions.

---

workspace

Permission: workspace.manage

Managed disk directories for extensions that need a real filesystem (git clones, compilers, media tools). Each workspace is scoped to {WORKSPACE_ROOT}/{packageID}/{name}/.

# Create a workspace (idempotent)
path = workspace.create("my-repo")
# Returns the absolute path to the directory

# Get the path (None if workspace doesn't exist)
path = workspace.path("my-repo")

# List all workspaces owned by this extension
names = workspace.list()  # ["my-repo", "cache"]

# Delete a workspace and all its contents
workspace.delete("my-repo")

# Get disk usage in bytes (10-second timeout)
size = workspace.usage("my-repo")  # 1048576

Constraints:

• Names must match ^[a-z][a-z0-9_]{0,62}$ (lowercase, no spaces or separators).
• Path traversal and symlink escape are blocked.
• Quota enforcement via WORKSPACE_QUOTA_MB env var (0 = unlimited).
• Module not available if WORKSPACE_ROOT is unset or not writable. Use settings.has_capability("workspace") to check availability.

---

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}}