Functions

Function registration, handlers, invocation patterns, and HTTP functions

Overview

Functions are the core building blocks of the III SDK. They are named, remotely-invocable units of work that can be called by any worker connected to the Engine. Functions support typed inputs/outputs, automatic tracing, and both synchronous and asynchronous invocation patterns.

Function Registration

Local Functions

Register a function with a handler that executes in your worker:

const functionRef = iii.registerFunction(
  {
    id: 'myservice::processOrder',
    description: 'Process customer order and return confirmation',
    request_format: {
      name: 'OrderRequest',
      type: 'object',
      body: [
        { name: 'orderId', type: 'string', required: true },
        { name: 'items', type: 'array', required: true },
        { name: 'customerId', type: 'string', required: true }
      ]
    },
    response_format: {
      name: 'OrderConfirmation',
      type: 'object',
      body: [
        { name: 'confirmationId', type: 'string' },
        { name: 'status', type: 'string' },
        { name: 'estimatedDelivery', type: 'string' }
      ]
    },
    metadata: { version: '1.0', team: 'orders' }
  },
  async (input) => {
    // Function implementation
    const { orderId, items, customerId } = input
    
    // Process order logic...
    
    return {
      confirmationId: crypto.randomUUID(),
      status: 'confirmed',
      estimatedDelivery: new Date(Date.now() + 86400000).toISOString()
    }
  }
)

Function registration message structure:

// Source: packages/node/iii/src/iii-types.ts:93-116
type RegisterFunctionMessage = {
  message_type: MessageType.RegisterFunction
  id: string                              // Function path (use :: for namespacing)
  description?: string                    // Human-readable description
  request_format?: RegisterFunctionFormat // Input schema
  response_format?: RegisterFunctionFormat// Output schema
  metadata?: Record<string, unknown>      // Custom metadata
  invocation?: HttpInvocationConfig       // For HTTP functions
}

Function IDs use :: as namespace separator (e.g., service::resource::action). This convention helps organize functions and avoid naming conflicts.

Function Naming Best Practices

1. Use Namespaces

   Prefix functions with service name:

   iii.registerFunction({ id: 'payments::charge' }, handler)
   iii.registerFunction({ id: 'payments::refund' }, handler)
   iii.registerFunction({ id: 'inventory::reserve' }, handler)
   
2. Action-Oriented Names

   Use verbs that describe the operation:

   'orders::create'      // Good
   'orders::order'       // Avoid
   'users::authenticate' // Good
   'users::auth'         // Less clear
   
3. Reserved Prefixes

   Avoid these Engine-reserved prefixes:

   • engine::* - Internal Engine functions
   • stream::* - Stream API functions
   • log::* - Logging functions

HTTP Functions

Register external HTTP endpoints (Lambda, Cloudflare Workers, etc.) as functions:

const httpFn = iii.registerHttpFunction(
  'external::stripe::createCharge',
  {
    url: 'https://api.stripe.com/v1/charges',
    method: 'POST',
    timeout_ms: 10000,
    headers: {
      'Content-Type': 'application/x-www-form-urlencoded'
    },
    auth: {
      type: 'bearer',
      token_key: 'STRIPE_SECRET_KEY'  // Resolved from Engine env
    }
  }
)

HTTP authentication types:

Bearer Token

auth: {
  type: 'bearer',
  token_key: 'API_TOKEN'  // Env var name in Engine
}

API Key Header

auth: {
  type: 'api_key',
  header: 'X-API-Key',
  value_key: 'SERVICE_API_KEY'
}

HMAC Signature

auth: {
  type: 'hmac',
  secret_key: 'WEBHOOK_SECRET'
}

HTTP functions are invoked by the Engine, not the worker. The Engine handles authentication, retries, and timeout enforcement.

Function Handlers

Handler Signature

Handlers are async functions that receive input and return output:

// Source: packages/node/iii/src/types.ts:15
type RemoteFunctionHandler<TInput = any, TOutput = any> = 
  (data: TInput) => Promise<TOutput>

Context Access

Access request context (logger, trace) using getContext():

import { getContext } from 'iii-sdk'

iii.registerFunction(
  { id: 'service::analyzeData' },
  async (input) => {
    const ctx = getContext()
    
    ctx.logger.info('Starting analysis', { recordCount: input.records.length })
    
    try {
      const result = await processRecords(input.records)
      
      // Add custom trace attributes
      ctx.trace?.setAttribute('records.processed', result.count)
      ctx.trace?.setAttribute('analysis.duration_ms', result.durationMs)
      
      ctx.logger.info('Analysis complete', { result })
      return result
    } catch (error) {
      ctx.logger.error('Analysis failed', { error: error.message })
      throw error
    }
  }
)

Context structure:

// Source: packages/node/iii/src/context.ts:6-10
type Context = {
  logger: Logger                    // Structured logger with trace correlation
  trace?: Span                      // Active OpenTelemetry span
}

See packages/node/iii/src/context.ts for implementation.

Automatic Tracing

All function handlers are automatically wrapped in OpenTelemetry spans:

// Source: packages/node/iii/src/iii.ts:202-217
handler: async (input, traceparent?, baggage?) => {
  if (getTracer()) {
    const parentContext = extractContext(traceparent, baggage)
    
    return context.with(parentContext, () =>
      withSpan(`call ${message.id}`, { kind: SpanKind.SERVER }, async span => {
        const traceId = currentTraceId() ?? crypto.randomUUID()
        const spanId = currentSpanId()
        const logger = new Logger(traceId, message.id, spanId)
        const ctx = { logger, trace: span }
        
        return withContext(async () => await handler(input), ctx)
      })
    )
  }
  // Fallback without tracing...
}

No manual span creation needed! Every function call gets a server span with trace correlation, parent/child relationships, and automatic context propagation.

Function Invocation

Synchronous Call (await response)

Wait for function result with optional timeout:

const result = await iii.call<InputType, OutputType>(
  'service::function',
  { param: 'value' },
  5000  // Optional timeout in ms (default: 30000)
)

Implementation:

// Source: packages/node/iii/src/iii.ts:289-328
trigger = async <TInput, TOutput>(
  function_id: string,
  data: TInput,
  timeoutMs?: number
): Promise<TOutput> => {
  const invocation_id = crypto.randomUUID()
  const traceparent = injectTraceparent()
  const baggage = injectBaggage()
  const effectiveTimeout = timeoutMs ?? this.invocationTimeoutMs
  
  return new Promise<TOutput>((resolve, reject) => {
    const timeout = setTimeout(() => {
      const invocation = this.invocations.get(invocation_id)
      if (invocation) {
        this.invocations.delete(invocation_id)
        reject(new Error(`Invocation timeout after ${effectiveTimeout}ms`))
      }
    }, effectiveTimeout)
    
    this.invocations.set(invocation_id, {
      resolve: (result: TOutput) => {
        clearTimeout(timeout)
        resolve(result)
      },
      reject: (error: unknown) => {
        clearTimeout(timeout)
        reject(error)
      },
      timeout
    })
    
    this.sendMessage(MessageType.InvokeFunction, {
      invocation_id, function_id, data, traceparent, baggage
    })
  })
}

Asynchronous Call (fire-and-forget)

Invoke without waiting for response:

iii.callVoid('notifications::sendEmail', {
  to: '[e-mail]',
  subject: 'Order Confirmation',
  body: '...'
})

Key differences:

Aspect	call()	callVoid()
Returns	Promise<TOutput>	void
Invocation ID	Generated UUID	None (message omits field)
Timeout	Yes (configurable)	No
Use Case	Request-response	Fire-and-forget notifications

callVoid() provides no delivery confirmation or error handling. Use for non-critical operations only.

Invocation Errors

Functions can fail in several ways:

1. Function Not Found

   // InvocationResult message
   {
     invocation_id: '...',
     function_id: 'missing::function',
     error: {
       code: 'function_not_found',
       message: 'Function not found'
     }
   }
   
2. Handler Exception

   {
     invocation_id: '...',
     function_id: 'service::function',
     error: {
       code: 'invocation_failed',
       message: 'Database connection timeout'
     }
   }
   
3. Timeout

   Rejected locally by SDK after timeout expires:

   try {
     await iii.call('slow::function', {}, 1000)
   } catch (error) {
     // Error: Invocation timeout after 1000ms: slow::function
   }
   

Function Discovery

List Available Functions

Query all registered functions across workers:

const functions = await iii.listFunctions()

functions.forEach(fn => {
  console.log(`${fn.function_id}: ${fn.description}`)
  console.log('  Input:', fn.request_format)
  console.log('  Output:', fn.response_format)
  console.log('  Metadata:', fn.metadata)
})

FunctionInfo structure:

// Source: packages/node/iii/src/iii-types.ts:164-170
type FunctionInfo = {
  function_id: string
  description?: string
  request_format?: RegisterFunctionFormat
  response_format?: RegisterFunctionFormat
  metadata?: Record<string, unknown>
}

React to Function Changes

Subscribe to function availability events:

const unsubscribe = iii.onFunctionsAvailable((functions) => {
  console.log(`Function registry updated: ${functions.length} functions available`)
  
  const hasPayments = functions.some(f => f.function_id.startsWith('payments::'))
  if (hasPayments) {
    console.log('Payment service is online')
  }
})

// Later: unsubscribe()

Implementation:

// Source: packages/node/iii/src/iii.ts:393-428
onFunctionsAvailable = (callback: FunctionsAvailableCallback): (() => void) => {
  this.functionsAvailableCallbacks.add(callback)
  
  if (!this.functionsAvailableTrigger) {
    const function_id = `engine.on_functions_available.${crypto.randomUUID()}`
    
    this.registerFunction(
      { id: function_id },
      async ({ functions }: { functions: FunctionInfo[] }) => {
        this.functionsAvailableCallbacks.forEach(handler => {
          handler(functions)
        })
        return null
      }
    )
    
    this.functionsAvailableTrigger = this.registerTrigger({
      type: EngineTriggers.FUNCTIONS_AVAILABLE,
      function_id,
      config: {}
    })
  }
  
  return () => {
    this.functionsAvailableCallbacks.delete(callback)
    if (this.functionsAvailableCallbacks.size === 0) {
      this.functionsAvailableTrigger?.unregister()
      this.functionsAvailableTrigger = undefined
    }
  }
}

Unregistering Functions

Functions can be unregistered dynamically:

const fnRef = iii.registerFunction({ id: 'temp::function' }, handler)

// Later...
fnRef.unregister()

Unregister message:

// Source: packages/node/iii/src/iii.ts:231-235
{
  unregister: () => {
    this.sendMessage(MessageType.UnregisterFunction, { id: message.id }, true)
    this.functions.delete(message.id)
  }
}

Channel Arguments

Functions can accept streaming channels as arguments:

import type { ChannelReader } from 'iii-sdk'

iii.registerFunction(
  { id: 'data::process' },
  async (input: { dataStream: ChannelReader }) => {
    const chunks: Buffer[] = []
    
    for await (const chunk of input.dataStream.stream) {
      chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk))
    }
    
    const data = JSON.parse(Buffer.concat(chunks).toString('utf-8'))
    return { processedRecords: data.length }
  }
)

The Engine automatically resolves channel references:

// Source: packages/node/iii/src/iii.ts:769-786
private resolveChannelValue(value: unknown): unknown {
  if (isChannelRef(value)) {
    return value.direction === 'read'
      ? new ChannelReader(this.address, value)
      : new ChannelWriter(this.address, value)
  }
  if (Array.isArray(value)) {
    return value.map(item => this.resolveChannelValue(item))
  }
  if (value !== null && typeof value === 'object') {
    const out: Record<string, unknown> = {}
    for (const [k, v] of Object.entries(value)) {
      out[k] = this.resolveChannelValue(v)
    }
    return out
  }
  return value
}

See Channels for complete channel API.

Multi-Language Examples

TypeScript

import { init } from 'iii-sdk'

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

iii.registerFunction(
  { id: 'greeting::hello' },
  async (input: { name: string }) => {
    return { message: `Hello, ${input.name}!` }
  }
)

const result = await iii.call('greeting::hello', { name: 'World' })
console.log(result.message)  // "Hello, World!"

Python

from iii import III

iii = III('ws://localhost:8080')
await iii.connect()

async def hello_handler(input: dict) -> dict:
    return {"message": f"Hello, {input['name']}!"}

iii.register_function(
    {"id": "greeting::hello"},
    hello_handler
)

result = await iii.call("greeting::hello", {"name": "World"})
print(result["message"])  # "Hello, World!"

Rust

use iii::{III, RegisterFunctionMessage};
use serde_json::{json, Value};

let iii = III::init("ws://localhost:8080", Default::default()).await?;

iii.register_function(
    RegisterFunctionMessage {
        id: "greeting::hello".into(),
        ..Default::default()
    },
    |input: Value| async move {
        let name = input["name"].as_str().unwrap();
        Ok(json!({ "message": format!("Hello, {}!", name) }))
    }
)?;

let result: Value = iii.call("greeting::hello", json!({"name": "World"})).await?;
println!("{}", result["message"]);  // "Hello, World!"

Next Steps

Triggers

Learn how to invoke functions from HTTP, events, and schedules

Channels

Stream data between functions with channels