JobContext

context is the only argument your run function gets. It's the doorway to everything the framework provides.

def run(context):
    context.entrypoint_key    # the entrypoint key, e.g. 'get_supplier'
    context.api          # HTTP client to the Nuntiq customer API
    context.logger       # structured logging
    context.secrets      # dict of secret values (passwords, tokens)
    context.config       # dict of non-secret settings
    context.job_params   # dict of per-run parameters
    context.customer_number   # tenant identifier
    context.work_dir     # per-run scratch directory (path, str)
    context.update_secret(name, value)   # store a secret the connector rotated itself
    context.files(mount)      # a folder on Nuntiq SFTP mounted for this run

The three input bags

The Nuntiq admin portal lets the operator configure three separate dictionaries per connector instance. They land on context exactly as configured.

context.secrets: encrypted at rest

Use for: passwords, API tokens, OAuth client secrets, private keys.

Stored encrypted at rest in the platform's secret store. Never logged by the framework, never displayed back in the admin UI after the operator saves them.

api_token = context.get_secret('api_token')
if not api_token:
    raise RuntimeError("secrets.api_token is required")

sftp_password = context.get_secret('sftp_password')

Use context.get_secret(key, default=None) rather than context.secrets[key] when you want a graceful fallback.

context.update_secret(): secrets the connector rotates itself

Some external systems replace a credential during a run, for example an OAuth server that returns a new refresh token with every access token. Store the new value at once:

context.update_secret('refresh_token', token['refresh_token'])
context.update_secrets({'refresh_token': rt, 'access_token': at})   # several, all or none

Allowed names	Only the secrets listed under Secrets the connector may update on the connector's Settings & Secrets tab (context.rotatable_secrets). Any other name raises SecretUpdateError (e.refused lists the names) and nothing is written.
Values	Non-empty strings of at most 64 KB.
Effect	Saved in the connector's secret store as a new version (the value before it is kept as the previous version); the next run reads it. context.secrets is updated in this run.
Log	The new value is masked in the run's output like every other secret. The customer's Change History records the key names, never the value, by "connector run" and the run's id.
Returns	The names whose value changed.
Toolkit	Writes the value into the secrets section of the settings file you run with; allowed names come from the file's rotatable_secrets.

Call it straight after the call that returned the new value, before other work, so a failure later in the run does not lose it. See OAuth 2.0 for the full pattern.

context.config: non-secret settings

Use for: URLs, hostnames, supplier codes, feature flags, anything that's sensitive enough to be customer-specific but not sensitive enough to encrypt.

api_url = context.get_config('api_url')                       # required
connector_name = context.get_config('connector_name')         # required

These survive across runs. They're set once at registration time and don't change unless the operator edits the connector instance.

The admin portal stores every configuration value and job parameter as a string ("false", "1", '{"a": "b"}'); a toolkit settings file carries real JSON types. Coerce in code so both behave the same:

include_zeros = str(context.get_config('include_zero_amount', default='false')).lower() in ('1', 'true')

Never test a flag with a bare if context.get_config('flag'):, because the string "false" is true in Python.

context.job_params: per-run parameters

Use for: backfill ranges, page limits, feature toggles you want the scheduler to override.

since_override = context.get_param('since_override')  # e.g. backfill from a date
limit = int(context.get_param('page_limit', default=100))

Set by the scheduler when it triggers a job. Different scheduled runs of the same entrypoint can pass different job_params. That's how you'd, say, run a connector with lookback_days=7 nightly but lookback_days=30 weekly for the same entrypoint.

context.api: the Nuntiq API

The authenticated HTTP client. Every call carries the job's short-lived token and the current customer_number header.

Two ways to use it:

# 1. Low-level: you know exactly which endpoint you want
suppliers = context.api.get('/v1/supplier', params={'limit': 100})   # {'data': [...], 'next_cursor': ...}

# 2. Typed entity Loads: recommended for everything CRUD-shaped
from lib.objects.supplier import SupplierLoad
load = SupplierLoad(context)
for s in load.get_all():
    print(s.supplier_number)

The entity Loads (SupplierLoad, InvoiceLoad, OrganizationLoad, etc.) are thin typed wrappers over the same HTTP calls. See BaseLoad for the patterns common to all of them.

Files go in with context.api.upload_files('capture' | 'einvoice', files), a role per file; see Sending files.

What the run's token may call depends on the entrypoint key:

Entrypoint	Operations
get_invoice	read and write invoices (upload_files('einvoice'), invoice_import), source documents (upload_files('capture')), read receiving inboxes
get_invoice_status	read and write invoices (upload_files('einvoice') too), submit results, lifecycle messages
push_invoice	read invoices, submit results, acknowledge and release, lifecycle messages, read organizations and suppliers
get_supplier, get_organization, get_purchaseorder, get_receipts, get_payment, get_fx_rates	their own object (plus the master data they map onto)
get_custom_N	customer data tables

Every key also gets the delta-state service. A call outside the list answers 403.

context.files(): Nuntiq SFTP files

context.files() takes no argument and returns the home folder of the run's SFTP user, read and write: the Data Integration SFTP user on Nuntiq's SFTP server chosen on the integration rule or scheduler task that started the run. A run without an SFTP user, or whose SFTP user is disabled, deleted or not a Data Integration user, raises ConfigError with the reason. See Nuntiq SFTP files.

context.logger: what shows up in the job log

context.logger.info("Starting pull")
context.logger.warn("API returned 0 rows", detail={'cursor': cursor})
context.logger.error("SFTP login failed", step='sftp_connect')

Output goes to stderr (which the worker captures as the job log file) and optionally to a structured log endpoint. See Logging for the detail.

context.customer_number: tenant scope

A string. Set by the worker when it constructs the job. Every API call carries it as a header automatically. You almost never read it directly. It's there when you need to include it in an external system call:

external_id = f"{context.customer_number}-{invoice.invoice_token}"

context.work_dir: scratch space

A per-run directory you can write to. Returns an absolute path as a string.

import os, csv

path = os.path.join(context.work_dir, 'export.csv')
with open(path, 'w') as f:
    csv.writer(f).writerows(rows)
context.logger.info(f"Wrote export to {path}")

Created lazily on first access. In production it's wiped when the container exits. In the toolkit it persists under <repo>/temp_work/ so you can inspect what your entrypoint produced. See Work dir for what to use it for.

Lifecycle

1. The framework constructs context exactly once, before importing your handler.py.
2. It's passed to run(context).
3. When run returns, the framework flushes any buffered log entries and exits.

You should treat context as a value, not something you can mutate. It's safe to pass to helper functions, store on classes, whatever. The framework doesn't peek inside it after you receive it.

What's next

• ApiClient: the methods available on context.api.
• BaseLoad: the read/write pattern shared by every entity.
• Logging: log levels, structured detail, the step argument.