Node.js SDK - Motia

Installation

npm install iii-sdk

Quick Start

import { init } from 'iii-sdk'

const iii = init('ws://localhost:49134')

iii.registerFunction({ id: 'myFunction' }, (req) => {
  return { status_code: 200, body: { message: 'Hello, world!' } }
})

iii.registerTrigger({
  type: 'http',
  function_id: 'myFunction',
  config: { api_path: '/hello', http_method: 'POST' },
})

const result = await iii.call('myFunction', { param: 'value' })

Initialization

init()

Initialize the III SDK and connect to the engine.

address

string

required

WebSocket address of the III Engine (e.g., ws://localhost:49134)

options

InitOptions

Optional configuration options

Show properties

workerName

string

Custom worker name (defaults to hostname:pid)

enableMetricsReporting

boolean

default:"true"

Enable automatic metrics reporting to the engine

invocationTimeoutMs

number

default:"30000"

Default timeout for function invocations in milliseconds

reconnectionConfig

ReconnectionConfig

WebSocket reconnection behavior configuration

Show properties

initialDelayMs

number

default:"1000"

Initial delay before first reconnection attempt

maxDelayMs

number

default:"30000"

Maximum delay between reconnection attempts

backoffMultiplier

number

default:"2.0"

Exponential backoff multiplier

jitterFactor

number

default:"0.3"

Random jitter factor (0-1)

maxRetries

number

default:"-1"

Maximum retry attempts (-1 for infinite)

otel

OtelConfig

OpenTelemetry configuration (enabled by default)

Show properties

enabled

boolean

default:"true"

Enable OpenTelemetry (can also set via OTEL_ENABLED env var)

serviceName

string

Service name for telemetry (defaults to worker name)

metricsEnabled

boolean

default:"true"

Enable metrics collection

tracesEnabled

boolean

default:"true"

Enable trace collection

logsEnabled

boolean

default:"true"

Enable log collection

telemetry

TelemetryOptions

Additional telemetry metadata

Show properties

language

string

User’s language/locale

project_name

string

Project name for analytics

framework

string

Framework name (e.g., “express”, “fastify”)

amplitude_api_key

string

Amplitude API key for analytics

import { init } from 'iii-sdk'

const iii = init('ws://localhost:49134', {
  workerName: 'my-worker',
  invocationTimeoutMs: 60000,
  reconnectionConfig: {
    maxRetries: 10,
    initialDelayMs: 2000,
  },
  otel: {
    enabled: true,
    serviceName: 'my-service',
  },
})

Function Registration

registerFunction()

message

object

required

Function registration metadata

Show properties

string

required

Unique function identifier (use :: for namespacing, e.g., external::my_lambda)

description

string

Human-readable function description

request_format

object

JSON schema for request validation

response_format

object

JSON schema for response validation

metadata

object

Additional metadata for the function

handler

function | HttpInvocationConfig

required

Function handler (local function) or HTTP invocation config (external HTTP endpoint)Local Handler:

(data: TInput) => Promise<TOutput>

HTTP Invocation Config:

Show properties

url

string

required

HTTP endpoint URL

method

string

default:"POST"

HTTP method (GET, POST, PUT, PATCH, DELETE)

timeout_ms

number

Request timeout in milliseconds

headers

object

Custom HTTP headers

auth

object

Authentication configuration (HMAC, Bearer, or API Key)

return

FunctionRef

Function reference with id and unregister() method

// Local handler
const ref = iii.registerFunction({ id: 'myFunction' }, async (req) => {
  return { status_code: 200, body: { message: 'Hello, world!' } }
})

// HTTP invocation (external endpoint)
iii.registerFunction(
  { id: 'external::lambda' },
  {
    url: 'https://my-lambda.example.com/invoke',
    method: 'POST',
    timeout_ms: 10000,
    headers: { 'X-Custom-Header': 'value' },
    auth: { type: 'bearer', token_key: 'MY_TOKEN_ENV_VAR' },
  }
)

// Unregister later
ref.unregister()

Function Invocation

call()

Invoke a function synchronously and wait for the result.

function_id

string

required

ID of the function to invoke

data

any

required

Input data to pass to the function

timeoutMs

number

Override the default invocation timeout

return

Promise<TOutput>

Promise that resolves with the function result

const result = await iii.call('myFunction', { param: 'value' })
console.log(result)

// With custom timeout
const result = await iii.call('slowFunction', { data: 'x' }, 60000)

callVoid()

Invoke a function asynchronously without waiting for a result (fire-and-forget).

function_id

string

required

ID of the function to invoke

data

any

required

Input data to pass to the function

iii.callVoid('myFunction', { param: 'value' })
// Function is invoked in the background, no result is returned

trigger()

Alias for call(). Invokes a function synchronously.

const result = await iii.trigger('myFunction', { param: 'value' })

triggerVoid()

Alias for callVoid(). Invokes a function asynchronously.

iii.triggerVoid('myFunction', { param: 'value' })

Trigger Management

registerTrigger()

trigger

object

required

Show properties

type

string

required

Trigger type (e.g., http, schedule, custom types)

function_id

string

required

ID of the function to invoke when triggered

config

object

required

Trigger-specific configuration (depends on trigger type)

return

Trigger

Trigger object with unregister() method

// HTTP trigger
const trigger = iii.registerTrigger({
  type: 'http',
  function_id: 'myFunction',
  config: {
    api_path: '/hello',
    http_method: 'POST',
  },
})

// Unregister later
trigger.unregister()

registerTriggerType()

triggerType

object

required

Show properties

string

required

Unique trigger type identifier

description

string

required

Human-readable description

handler

TriggerHandler

required

Trigger type handler with registerTrigger and unregisterTrigger methods

iii.registerTriggerType(
  {
    id: 'myTriggerType',
    description: 'My custom trigger type',
  },
  {
    registerTrigger: async (config) => {
      // Setup logic when a trigger of this type is registered
      console.log('Trigger registered:', config)
    },
    unregisterTrigger: async (config) => {
      // Cleanup logic when a trigger is unregistered
      console.log('Trigger unregistered:', config)
    },
  }
)

unregisterTriggerType()

Unregister a custom trigger type.

triggerType

object

required

string

required

Trigger type ID to unregister

iii.unregisterTriggerType({ id: 'myTriggerType' })

Service Registration

registerService()

message

object

required

Show properties

string

required

Service identifier

description

string

Service description

parent_service_id

string

Parent service ID for hierarchical organization

iii.registerService({
  id: 'my-service',
  description: 'My service description',
})

Engine Functions

listFunctions()

List all registered functions in the engine.

return

Promise<FunctionInfo[]>

Array of function information objects

const functions = await iii.listFunctions()
functions.forEach((fn) => {
  console.log(fn.function_id, fn.description)
})

listWorkers()

List all connected workers in the engine.

return

Promise<WorkerInfo[]>

Array of worker information objects

const workers = await iii.listWorkers()
workers.forEach((worker) => {
  console.log(worker.name, worker.status, worker.function_count)
})

Channels

createChannel()

Create a streaming channel pair for worker-to-worker data transfer.

bufferSize

number

default:"64"

Optional buffer size for the channel

return

Promise<Channel>

Channel object with writer, reader, writerRef, and readerRef

const channel = await iii.createChannel(128)

// Write data
await channel.writer.write({ data: 'hello' })
await channel.writer.close()

// Read data
for await (const item of channel.reader) {
  console.log(item)
}

// Pass channel refs to other functions
await iii.call('processData', {
  input: channel.readerRef,
  output: channel.writerRef,
})

Event Listeners

onFunctionsAvailable()

Subscribe to function availability events.

callback

function

required

Callback invoked when functions are registered/unregistered

(functions: FunctionInfo[]) => void

return

function

Unsubscribe function

const unsubscribe = iii.onFunctionsAvailable((functions) => {
  console.log('Available functions:', functions.map((f) => f.function_id))
})

// Unsubscribe later
unsubscribe()

onLog()

Subscribe to OpenTelemetry log events from the engine.

callback

function

required

Callback invoked when log events are received

(log: OtelLogEvent) => void

config

LogConfig

Optional log filtering configuration

Show properties

level

string

default:"all"

Minimum severity level: trace, debug, info, warn, error, fatal, all

return

function

Unsubscribe function

const unsubscribe = iii.onLog(
  (log) => {
    console.log(`[${log.severity_text}] ${log.body}`, log.attributes)
  },
  { level: 'warn' } // Only receive warnings and errors
)

// Unsubscribe later
unsubscribe()

onConnectionStateChange()

Monitor WebSocket connection state changes.

callback

function

required

Callback invoked when connection state changes

(state: 'disconnected' | 'connecting' | 'connected' | 'reconnecting' | 'failed') => void

return

function

Unsubscribe function

const unsubscribe = iii.onConnectionStateChange((state) => {
  console.log('Connection state:', state)
})

// Unsubscribe later
unsubscribe()

getConnectionState()

Get the current connection state.

return

string

Current state: disconnected, connecting, connected, reconnecting, or failed

const state = iii.getConnectionState()
console.log('Current state:', state)

Context & Logging

getContext()

Get the current execution context (available within function handlers).

return

Context

Context object with logger and optionally trace (OpenTelemetry span)

import { getContext } from 'iii-sdk'

iii.registerFunction({ id: 'myFunction' }, async (req) => {
  const { logger, trace } = getContext()
  
  logger.info('Processing request', { input: req })
  logger.warn('Something might be wrong')
  logger.error('An error occurred', { error: 'details' })
  
  if (trace) {
    trace.setAttribute('custom.attribute', 'value')
  }
  
  return { result: 'success' }
})

withContext()

Run a function with a custom context.

function

required

Function to execute

context

Context

required

Context to use during execution

import { withContext, Logger } from 'iii-sdk'

const logger = new Logger('trace-id', 'my-function')
const ctx = { logger }

await withContext(async () => {
  const { logger } = getContext()
  logger.info('Running with custom context')
}, ctx)

Lifecycle

shutdown()

Gracefully shut down the SDK, cleaning up all resources.

return

Promise<void>

Promise that resolves when shutdown is complete

await iii.shutdown()

Types

FunctionInfo

type FunctionInfo = {
  function_id: string
  description?: string
  request_format?: object
  response_format?: object
  metadata?: Record<string, unknown>
}

WorkerInfo

type WorkerInfo = {
  id: string
  name?: string
  runtime?: string
  version?: string
  os?: string
  ip_address?: string
  status: 'connected' | 'available' | 'busy' | 'disconnected'
  connected_at_ms: number
  function_count: number
  functions: string[]
  active_invocations: number
}

ApiRequest / ApiResponse

type ApiRequest<TBody = unknown> = {
  path_params: Record<string, string>
  query_params: Record<string, string | string[]>
  body: TBody
  headers: Record<string, string | string[]>
  method: string
  request_body: ChannelReader
}

type ApiResponse<TStatus = number, TBody = any> = {
  status_code: TStatus
  headers?: Record<string, string>
  body?: TBody
}

Examples

Basic HTTP API

import { init } from 'iii-sdk'

const iii = init('ws://localhost:49134')

iii.registerFunction({ id: 'api.hello' }, async (req) => {
  return {
    status_code: 200,
    body: { message: 'Hello, world!' },
  }
})

iii.registerTrigger({
  type: 'http',
  function_id: 'api.hello',
  config: { api_path: '/hello', http_method: 'GET' },
})

Calling Other Functions

iii.registerFunction({ id: 'processOrder' }, async (order) => {
  // Call payment service
  const payment = await iii.call('payment.charge', {
    amount: order.total,
    customer: order.customerId,
  })
  
  // Notify warehouse asynchronously
  iii.callVoid('warehouse.ship', { orderId: order.id })
  
  return { success: true, paymentId: payment.id }
})

Streaming Data with Channels

iii.registerFunction({ id: 'processStream' }, async (req) => {
  const channel = await iii.createChannel()
  
  // Process in background
  iii.callVoid('processData', { input: channel.readerRef })
  
  // Write data
  for (let i = 0; i < 100; i++) {
    await channel.writer.write({ value: i })
  }
  await channel.writer.close()
  
  return { status: 'processing' }
})

​Installation

​Quick Start

​Initialization

​init()

​Function Registration

​registerFunction()

​Function Invocation

​call()

​callVoid()

​trigger()

​triggerVoid()

​Trigger Management

​registerTrigger()

​registerTriggerType()

​unregisterTriggerType()

​Service Registration

​registerService()

​Engine Functions

​listFunctions()

​listWorkers()

​Channels

​createChannel()

​Event Listeners

​onFunctionsAvailable()

​onLog()

​onConnectionStateChange()

​getConnectionState()

​Context & Logging

​getContext()

​withContext()

​Lifecycle

​shutdown()

​Types

​FunctionInfo

​WorkerInfo

​ApiRequest / ApiResponse

​Examples

​Basic HTTP API

​Calling Other Functions

​Streaming Data with Channels

Installation

Quick Start

Initialization

init()

Function Registration

registerFunction()

Function Invocation

call()

callVoid()

trigger()

triggerVoid()

Trigger Management

registerTrigger()

registerTriggerType()

unregisterTriggerType()

Service Registration

registerService()

Engine Functions

listFunctions()

listWorkers()

Channels

createChannel()

Event Listeners

onFunctionsAvailable()

onLog()

onConnectionStateChange()

getConnectionState()

Context & Logging

getContext()

withContext()

Lifecycle

shutdown()

Types

FunctionInfo

WorkerInfo

ApiRequest / ApiResponse

Examples

Basic HTTP API

Calling Other Functions

Streaming Data with Channels