Documentation / Pipelines

Pipeline SDK (ctx) Reference

The definitive reference for the injected ctx runtime context — adapters, secrets, SSRF-safe HTTP, the internal workspace, files, structured logging, child runs, the Result object, and pipeline constraints.

A Nexus pipeline is plain Python source stored on the server and run by a sandboxed runtime. It defines exactly one entrypoint — async def run(ctx) — and reaches every capability through the injected ctx SDK. You never import database drivers, HTTP clients, or adapters: they arrive on ctx. This page is the definitive reference for that surface — every attribute, the shape of the Result a run returns, the constraints the runtime enforces, and a realistic multi-step example. It is for developers and analysts authoring automations inside their own Nexus tenant.

For orientation and the authoring lifecycle, start at the Pipelines Overview. For scheduling and monitoring, see Scheduling Pipelines and Monitoring & Auditing Pipelines.

The entrypoint

async def run(ctx):
    rows = await ctx.adapters.monday.read({"collection": "deals", "limit": 50})
    await ctx.logger.info("read rows", count=len(rows))
    return {"rows": len(rows)}

The module must define async def run(ctx). A synchronous def run or a missing run is rejected at save time with a clear error (PipelineMissingEntrypointError), never at 3 a.m. mid-schedule.
Return a dict (or any value) and it becomes the run result; return None for an empty success. You may also return a Result object explicitly (see below).
Raise on failure. Any exception thrown out of run is caught, the run is marked failed, and the complete Python traceback — exception type, message, and the offending line — is persisted to the run record. You debug from real tracebacks, never a silent error. The executor never raises out of run, so callers and schedulers stay stable regardless of what your code does.

The `ctx` surface

Every attribute below is a real attribute of the runtime context. Optional capabilities that are not wired in your environment fall back to a loud stub: attribute access works, but calling a method raises a clear UNCONFIGURED_* error rather than a NoneType failure — so a misconfiguration reads as an explicit message, not a mysterious crash.

Attribute	Purpose
`ctx.adapters.<name>`	Read / write / discover your connected data sources (bridges)
`ctx.secrets`	Read-only Key Vault access (`get`, `get_required`)
`ctx.http`	SSRF-safe, allowlisted HTTP client for external APIs
`ctx.workspace`	The tenant-private internal Postgres workspace
`ctx.files`	Blob outputs and named file references
`ctx.nexus`	Invoke child pipelines (orchestration)
`ctx.tools.<name>`	Tenant-registered external-API tools / connectors
`ctx.logger`	Structured, persisted run logging
`ctx.params`	Per-invocation input dict the run was started with
`ctx.variables`	Per-run scratch namespace, captured into the result
`ctx.pipeline_id` / `ctx.run_id` / `ctx.tenant_id`	Read-only identity strings

Everything on ctx that performs I/O is a coroutine — always await it.

ctx.adapters — data sources

Each adapter exposes read, write, and discover. The available names are your tenant's enabled bridges (for example monday, netsuite, mssql, mysql, postgres). Adapter calls are bound to the run's identity by the host — a pipeline cannot reach another user's credentials, and every read and write is RBAC-gated against the run's principal.

rows   = await ctx.adapters.monday.read({"collection": "boards", "limit": 50})
schema = await ctx.adapters.postgres.discover()
await ctx.adapters.netsuite.write([{"id": 1, "amount": 99}], op="upsert")

read takes a query dict (collection, filters, limit); the engine pushes what the adapter can handle down to the source and applies the rest in memory, so filters behave consistently even on adapters with limited pushdown. Which operators each connector pushes down varies — see the connector capability notes and the UQL operators reference. Writes are additionally gated by each adapter's writability rules (read-only collections, views, and computed columns are blocked).

ctx.secrets — Key Vault (read-only)

key = await ctx.secrets.get("monday_api_key")          # str | None
key = await ctx.secrets.get_required("monday_api_key")  # str, raises if absent

Secrets resolve from your tenant's Key Vault. Pipelines never write or rotate secrets — the surface is read-only by design. Prefer get_required for mandatory secrets so a missing value fails the run loudly with a named error rather than propagating a None into an HTTP header. Secret values are never written into logs, the result, or tracebacks by the runtime.

ctx.http — SSRF-safe HTTP client

resp = await ctx.http.get(url, headers={"Authorization": f"Bearer {key}"})
resp = await ctx.http.post(url, json={"name": "x"})
resp = await ctx.http.request("PATCH", url, json=body, timeout=30.0)

Methods: get, post, put, delete, and the generic request(method, url, *, headers=, params=, json=, data=, timeout=). This is the only sanctioned outbound-HTTP path — third-party clients like httpx and requests are not importable. Network access is governed on every call:

Each URL is validated against the tenant's egress allowlist at the application layer, on top of the network-layer NSG egress rules on the tenant VNet.
Every request runs an SSRF check — hostnames resolve and are pinned, and requests to link-local, loopback, and internal metadata addresses are refused, defeating DNS-rebind style attacks.
Calls are per-tenant rate limited.

Use ctx.http for any endpoint not covered by a native adapter. For a first-class, reusable integration with auth resolution and request/response scripting, register a generic HTTP/REST connector instead and call it through ctx.tools.

On backoff: the runtime does not automatically retry a pipeline on failure. Some adapters do apply bounded backoff to HTTP 429 responses and to child-run polling — that is adapter-level behavior, not run-level retry. If you want a call retried, write the retry loop in your pipeline (idempotently). See Retries & Idempotency.

ctx.workspace — internal Postgres workspace

A tenant-private Postgres database with its own managed identity, structurally isolated from Nexus's own operational tables — it is a separate database reached by a separate MI, so pipeline code literally cannot address the operational tables. It is the natural place to stage, join, and accumulate intermediate data between steps or between runs. Create tables first (via the desktop Workspace page or the workspace tooling), then read and write rows from a pipeline.

await ctx.workspace.insert("daily_metrics", [{"day": "2026-06-14", "total": 42}])
await ctx.workspace.update("daily_metrics", {"total": 43}, {"day": "2026-06-14"})
await ctx.workspace.delete("daily_metrics", {"day": "2026-06-14"})
rows = await ctx.workspace.read("daily_metrics", limit=100)

Both update and delete require their where clause, and update requires a non-empty set map — there are no accidental full-table writes. The workspace supports full SQL pushdown (filter, order, group, aggregate) when you query it through UQL. See Internal Workspace for its isolation model and DDL surface.

ctx.files — outputs and references

out   = await ctx.files.write("report", rows, file_type="csv")  # OUTPUT blob
prior = await ctx.files.read_output("report")                   # newest output, parsed
text  = await ctx.files.read("monthly_template")                # a named reference's text

write(name, data, *, file_type="csv") writes a list of dicts as an output blob (csv, json, xlsx), stored in your tenant and returned as a file reference.
read_output(name) finds the most recent output this pipeline produced for that name and returns its parsed data — how you feed an earlier run's output back into a later run.
read(name) returns the text of a named inline reference — for templates, SQL, and small config. See Files.

ctx.nexus — child runs

ctx.nexus lets a parent pipeline invoke other pipelines as children and compose multi-step workflows. Child runs execute through the same sandboxed executor, carry their own run records, and roll their cost and identity up into the parent's record, so an orchestration reads as one auditable tree.

async def run(ctx):
    extract = await ctx.nexus.run_pipeline("extract-invoices", params={"since": ctx.params["since"]})
    await ctx.nexus.run_pipeline("load-warehouse", params={"batch": extract["batch_id"]})
    return {"orchestrated": 2}

Child-run polling uses bounded backoff internally — this is the one place "backoff" legitimately appears, and it is polling, not run-level retry.

ctx.tools — registered connectors and tools

await ctx.tools.slack.post_message(channel="#ops", text="pipeline done")
await ctx.tools.invoke("slack", "post_message", channel="#ops", text="hi")

ctx.tools reaches tenant-registered external-API tools and generic HTTP/REST connectors, with their auth resolved by the bridge. Use it when you have a reusable, governed integration rather than an ad hoc ctx.http call.

ctx.logger — structured logging

await ctx.logger.info("fetched rows", count=len(rows), bridge="monday")
await ctx.logger.warning("record skipped", id=rec_id, reason="missing amount")
await ctx.logger.error("downstream rejected batch", status=resp.status_code)

debug, info, warning, and error are all async — await them. The first argument is the message; keyword arguments become structured fields that stay filterable in the run record (log structured fields rather than string-formatting them into the message). Logs are persisted as they are emitted and are preserved even when the pipeline later raises — if you logged three steps and then failed on the fourth, all three log lines remain attached to the failed run alongside the traceback.

ctx.params and ctx.variables

ctx.params is the dict the run was started with — REST input, an MCP invocation payload, or the parameters attached to a schedule. Read inputs from it, ideally with defaults:

since = ctx.params.get("since", "2026-01-01")

ctx.variables is a per-run scratch namespace. Whatever you leave in it is captured into the run result at the end, so it doubles as a place to surface intermediate state for diagnostics.

Identity fields

ctx.pipeline_id, ctx.run_id, and ctx.tenant_id are read-only strings identifying the current pipeline, this specific run, and the tenant. Use them to tag outbound requests or correlate your own logging.

The Result object

Every run resolves to a Result. Return a plain value and the runtime wraps it; return a Result and it is used as-is. The persisted record exposes:

Field	Meaning
`status`	`"succeeded"` or `"failed"`
`exception_text`	The full Python traceback when `status == "failed"`; empty on success
`rows_read` / `rows_created` / `rows_updated` / `rows_deleted`	Row counters accumulated across adapter and workspace operations
`log_entries`	The ordered, structured `ctx.logger` emissions
`variables`	The final `ctx.variables` dict
Return payload	Whatever `run` returned (wrapped if it wasn't already a `Result`)

The executor never raises out — every failure path returns a Result with status="failed" and exception_text populated. If you construct a failed Result yourself but omit exception_text, the runtime backfills a placeholder so a failure is never blank. This is the core reliability property of the runtime: a run always resolves to a record you can read, whether it succeeded or failed.

Pipeline constraints and timeouts

The runtime wraps every run in defense-in-depth limits so a runaway or buggy pipeline can't run unbounded or scribble across a system:

Per-run wall-clock timeout — a default ceiling (3600 seconds) enforced by the worker wrapper, with a hard maximum of 6 hours. Override it per pipeline with a frontmatter directive:
```
# nexus:timeout 1800

async def run(ctx):
    ...
```
A run that exceeds its timeout is stopped and recorded as failed.
Row-count constraints — caps on the maximum rows a single run may read, create, update, or delete. These are a blast-radius guardrail: they turn "a bug just tried to update every row" into a bounded, recorded failure instead of a mass mutation.

These limits, the durable run queue, concurrency controls, and the (rolling-out) env-scrubbed isolated-worker execution model are treated in depth in Timeouts & Limits and Execution & Concurrency.

What you can import

The sandbox checks imports at save/validate time. Allowed are the runtime SDK plus pure-data standard-library modules — json, datetime, decimal, math, re, collections, itertools, functools, uuid, hashlib, base64, urllib.parse, and asyncio, among others. Modules like os, subprocess, socket, pathlib, httpx/requests, and database drivers are not allowed — reach those capabilities through ctx instead. The builtins open, eval, exec, compile, input, breakpoint, and __import__ are stripped from the runtime namespace, so even a runtime call to open(...) raises NameError rather than touching disk. Extending the allowlist is a governed, security-reviewed change, not a config toggle.

A realistic multi-step pipeline

The pattern below is the one most production pipelines follow: extract → stage in the workspace → transform → load → export, staging intermediate state in the tenant-private workspace so each step is inspectable and the transform is pure SQL over data you control.

# nexus:trigger scheduled 0 6 * * *
# nexus:timeout 1800

async def run(ctx):
    since = ctx.params.get("since", "2026-01-01")

    # 1. EXTRACT from the source system
    invoices = await ctx.adapters.netsuite.read(
        {"collection": "invoices", "updated_after": since, "limit": 5000}
    )
    await ctx.logger.info("extracted invoices", count=len(invoices), since=since)

    # 2. STAGE raw rows in the tenant-private workspace
    staged = [
        {"id": inv["id"], "customer": inv["entity"], "amount": inv["total"], "day": since}
        for inv in invoices
    ]
    await ctx.workspace.delete("stg_invoices", {"day": since})   # idempotent re-run
    await ctx.workspace.insert("stg_invoices", staged)

    # 3. TRANSFORM: keep only material invoices
    material = [r for r in staged if r["amount"] and r["amount"] >= 100]
    await ctx.logger.info("filtered material invoices", kept=len(material), dropped=len(staged) - len(material))

    # 4. LOAD into the reporting system
    written = await ctx.adapters.mssql.write(material, op="upsert")
    await ctx.logger.info("loaded to mssql", rows=written)

    # 5. EXPORT a downloadable artifact
    await ctx.files.write("invoices_export", material, file_type="csv")

    return {"extracted": len(invoices), "loaded": written}

Notice the idempotency pattern in step 2: the run deletes the day's staged rows before inserting, so re-running the pipeline for the same since converges to the same state rather than double-writing. Because Nexus does not automatically retry failed runs, writing idempotent pipelines like this is how you make a manual rerun safe — the recommended pattern is spelled out in Retries & Idempotency.

Remember

await everything on ctx: adapters, secrets, http, workspace, files, and logger are all coroutines.
Raise a clear exception to fail a run — the full traceback is captured for you; the executor never raises out.
Read and write only through ctx; there is no direct file I/O, no third-party HTTP client, and no way to reach the operational database.
Design writes to be idempotent, and set a realistic # nexus:timeout for long jobs.