FlowOS Docs
Flow SDK

NodeContext API

Complete reference for the ctx object injected into every Flow SDK artifact — scripts, custom nodes, scheduled jobs, and webhook handlers.

Every Flow SDK artifact receives a ctx (NodeContext) object as its first argument. It provides access to tables, HTTP, events, notifications, logging, secrets, and flow control. All async methods return Promises and should be awaited.

example-node.ts
import type { NodeContext } from '@flowos/sdk'

export async function run(ctx: NodeContext) {
  const record = await ctx.tables.find('incidents', ctx.trigger.payload.id)
  await ctx.notify.send({
    channel: 'slack',
    to: record.assigned_to,
    message: `Incident ${record.number} updated`
  })
  return { updated: true }
}

ctx.tables

Full CRUD access to any table in your workspace. All operations respect the token's role permissions.

findmethod
ctx.tables.find(table: string, id: string): Promise<Record>
ReturnsPromise<Record>

Fetch a single record by its ID. Throws if the record does not exist.

typescript
const incident = await ctx.tables.find('incidents', 'inc_01HX...')
// incident.title, incident.status, etc.
querymethod
ctx.tables.query(table: string, params?: QueryParams): Promise<QueryResult>
ReturnsPromise<QueryResult>

Query records with filters, sorting, and pagination. Returns { data: Record[], total: number, page: number, pageSize: number }.

typescript
const result = await ctx.tables.query('incidents', {
  filter: { status: 'open', severity: 'P1' },
  sort: [{ field: 'created_at', dir: 'desc' }],
  page: 1,
  pageSize: 50
})
for (const inc of result.data) { ... }
createmethod
ctx.tables.create(table: string, data: Record<string,unknown>): Promise<Record>
ReturnsPromise<Record>

Create a new record. System fields (_id, created_at, etc.) are set automatically. Triggers any record-event workflows listening to this table.

typescript
const problem = await ctx.tables.create('problems', {
  title: 'Database connection drops under load',
  status: 'open',
  category: 'database',
  impact: 'high'
})
updatemethod
ctx.tables.update(table: string, id: string, data: Partial<Record>): Promise<Record>
ReturnsPromise<Record>

Update fields on a record. Only the provided fields are changed. Returns the full updated record.

typescript
await ctx.tables.update('incidents', incidentId, {
  status: 'resolved',
  resolution: 'Rolled back deployment 4.2.1',
  resolved_at: new Date().toISOString()
})
deletemethod
ctx.tables.delete(table: string, id: string): Promise<void>
ReturnsPromise<void>

Permanently delete a record. For system tables this may be blocked by data retention policies — check the table's settings.

typescript
await ctx.tables.delete('catalog_requests', requestId)
upsertmethod
ctx.tables.upsert(table: string, match: Record<string,unknown>, data: Record<string,unknown>): Promise<{ record: Record; created: boolean }>
ReturnsPromise<{ record, created }>

Create a record if none matches the filter, otherwise update the first match. Returns the record and a boolean indicating whether it was created.

typescript
const { record, created } = await ctx.tables.upsert(
  'cmdb_items',
  { external_id: server.hostname },
  { name: server.hostname, class: 'server', status: 'active', attributes: server }
)
sqlmethod
ctx.tables.sql(query: string, params?: unknown[]): Promise<unknown[]>
ReturnsPromise<unknown[]>

Execute a parameterized read-only SQL query against your workspace's DB. Only SELECT statements are allowed. Use $1, $2 placeholders for parameters.

typescript
const rows = await ctx.tables.sql(
  'SELECT id, number, title FROM incidents WHERE severity = $1 AND created_at > $2',
  ['P1', '2026-01-01']
)

ctx.http

A pre-configured HTTP client for calling external APIs. Requests automatically include workspace proxy headers, respect rate limits, and log to the integration transaction log.

get / post / put / patch / deletemethod
ctx.http.get(url: string, options?: HttpOptions): Promise<HttpResponse>
ReturnsPromise<HttpResponse>

Make an HTTP request. Returns { status, headers, body } where body is auto-parsed as JSON if content-type is application/json. Throws on network errors; non-2xx responses do not throw unless options.throwOnError is true.

typescript
const res = await ctx.http.post('https://api.pagerduty.com/incidents', {
  headers: {
    Authorization: `Token token=${await ctx.secrets.get('PAGERDUTY_TOKEN')}`
  },
  body: { incident: { type: 'incident', title: ctx.trigger.payload.title } }
})
ctx.log.info('PagerDuty incident created', { id: res.body.incident.id })
connectormethod
ctx.http.connector(connectorId: string): ConnectorClient
ReturnsConnectorClient

Get a pre-authenticated client for a configured Integration Studio connector. The client exposes .get(), .post(), .put(), .patch(), .delete() with auth headers and base URL already set.

typescript
const jira = ctx.http.connector('jira-prod')
const issue = await jira.post('/rest/api/3/issue', {
  body: { fields: { project: { key: 'OPS' }, summary: title, issuetype: { name: 'Bug' } } }
})

ctx.events

Publish messages to the event bus or read from an event topic within the current run context.

publishmethod
ctx.events.publish(topic: string, payload: unknown, options?: PublishOptions): Promise<{ eventId: string }>
ReturnsPromise<{ eventId }>

Publish an event to a topic. Any workflow with a matching event-bus trigger and subscribers on this topic will receive it. Options: { delay?: number (ms), dedup?: string (deduplication key) }.

typescript
await ctx.events.publish('incident.resolved', {
  incidentId: incident.id,
  number: incident.number,
  resolvedBy: ctx.user?.id
})

ctx.notify

Send real-time notifications to users through any configured channel.

sendmethod
ctx.notify.send(options: NotifyOptions): Promise<{ notificationId: string }>
ReturnsPromise<{ notificationId }>

Send a notification. Channels: in_app | email | slack | teams | webhook | sms. Use a template ID or inline message. The to field accepts a user ID, email, or team slug.

typescript
await ctx.notify.send({
  channel: 'email',
  to: incident.assigned_to,
  template: 'incident-assigned',
  variables: {
    incidentNumber: incident.number,
    incidentTitle: incident.title,
    severity: incident.severity
  }
})
broadcastmethod
ctx.notify.broadcast(options: BroadcastOptions): Promise<{ sent: number }>
ReturnsPromise<{ sent: number }>

Send to all members of a team or all users with a given role. Returns the count of notifications sent.

typescript
await ctx.notify.broadcast({
  to: { role: 'on-call-engineer' },
  channel: 'slack',
  message: `P1 incident declared: ${incident.title}`
})

ctx.log

Structured logging that appears in the Run Lifecycle trace panel and is queryable via the SDK Execution Log page.

typescript
ctx.log.debug('Fetching CI details', { ciId })       // stripped in production by default
ctx.log.info('Connector call succeeded', { status })
ctx.log.warn('Retry attempt', { attempt, error: e.message })
ctx.log.error('Hard failure', { error: e.message, stack: e.stack })

All log methods accept a message string and an optional structured data object. Logs are retained for the same duration as the run record (default 90 days).

ctx.secrets

getmethod
ctx.secrets.get(name: string): Promise<string>
ReturnsPromise<string>

Read a secret from the workspace vault. The name is the secret's key slug. Throws if the secret does not exist or the artifact's access rule doesn't permit it.

typescript
const apiKey = await ctx.secrets.get('DATADOG_API_KEY')
const res = await ctx.http.post('https://api.datadoghq.com/api/v1/events', {
  headers: { 'DD-API-KEY': apiKey },
  body: { title: 'Deployment completed', text: ctx.trigger.payload.version }
})

ctx.cache

A fast workspace-scoped key-value cache backed by Redis. Useful for avoiding redundant API calls within a workflow or across short-interval runs.

typescript
// Write with optional TTL (seconds)
await ctx.cache.set('last-check', JSON.stringify({ ts: Date.now() }), 300)

// Read — returns null if missing or expired
const raw = await ctx.cache.get('last-check')
const data = raw ? JSON.parse(raw) : null

// Atomic increment
const count = await ctx.cache.incr('daily-alert-count')

// Delete
await ctx.cache.del('last-check')

ctx.flow

Control flow utilities — set workflow variables, pause for approvals, and trigger sub-workflows.

typescript
// Set a workflow-scoped variable accessible by downstream nodes
ctx.flow.setVar('resolvedIncidentId', incident.id)

// Read a variable set by an upstream node
const priority = ctx.flow.getVar<string>('calculatedPriority')

// Pause and wait for human approval (returns once approved/rejected)
const { approved, approver, comment } = await ctx.flow.waitForApproval({
  title: `Approve change ${change.number}`,
  approvers: change.cab_reviewers,
  timeoutHours: 48
})

// Trigger a sub-workflow and wait for its result
const result = await ctx.flow.runWorkflow('wf_01HX_notify_oncall', {
  incidentId: incident.id
})

ctx.trigger

Read-only access to the payload that triggered this run.

typescript
// For record-event triggers
const { table, action, record, previous } = ctx.trigger.payload
// table: 'incidents', action: 'updated', record: { id, status, ... }, previous: { status: 'open' }

// For webhook triggers
const { headers, body, query } = ctx.trigger.payload

// For schedule triggers
const { scheduledAt, runNumber } = ctx.trigger.payload

// Trigger type
ctx.trigger.type // 'record_event' | 'webhook' | 'schedule' | 'itsm_event' | 'manual'

ctx.user & ctx.workspace

typescript
// The user who manually triggered the run (null for automated)
ctx.user?.id
ctx.user?.email
ctx.user?.name

// Workspace metadata
ctx.workspace.id
ctx.workspace.name
ctx.workspace.slug
ctx.workspace.timezone  // e.g. "America/New_York"
ctx.workspace.locale    // e.g. "en-US"

TypeScript Types

Full TypeScript types ship with @flowos/sdk. Import them for type safety in your artifacts:

typescript
import type {
  NodeContext,
  QueryParams,
  QueryResult,
  HttpOptions,
  HttpResponse,
  NotifyOptions,
  PublishOptions,
  Record as FlowRecord,
} from '@flowos/sdk'
Execution limits — SDK artifacts run with a default 30-second timeout. Long-running operations should use ctx.flow.runWorkflow to delegate to a sub-workflow that can run asynchronously. HTTP requests have a separate 10-second timeout; override with options.timeoutMs.