Script Components - Ensemble Edge

Cloudflare Workers Compatibility: Inline code (config.code) is not supported in Cloudflare Workers due to security restrictions. All code must be in script files that are bundled at build time.

Define reusable code scripts as components for consistent business logic across ensembles.

Overview

Script components enable you to:

Reuse code logic across multiple agents and ensembles
Version scripts with semantic versioning for reproducibility
Organize TypeScript code separately from YAML
Type-safe development with full TypeScript support
Test scripts independently with unit tests
Bundle scripts at build time for Workers compatibility

Quick Start

1. Create a Script File

Create a TypeScript file in your scripts/ directory:

// scripts/transform-data.ts
import type { AgentExecutionContext } from '@ensemble-edge/conductor'

interface TransformInput {
  raw_data: Array<{
    id: string
    value: number
    min: number
    max: number
    created_at: string
  }>
}

export default function transformData(context: AgentExecutionContext) {
  const { raw_data } = context.input as TransformInput

  const processed = raw_data.map(item => ({
    id: item.id,
    normalized_value: (item.value - item.min) / (item.max - item.min),
    timestamp: new Date(item.created_at).toISOString()
  }))

  return {
    processed,
    total_items: processed.length
  }
}

2. Reference in Your Ensemble

name: data-processor

agents:
  - name: transform
    operation: code
    config:
      script: scripts/transform-data
    input:
      raw_data: $input.raw_data

input:
  raw_data: []

output:
  result: ${transform.output}

3. Build & Deploy

Scripts are automatically discovered and bundled during the build:

npm run build    # Scripts are bundled automatically
npm run deploy   # Deploy to Cloudflare

URI Formats

Both formats are supported for referencing scripts:

Format	Example
Shorthand	`scripts/transforms/csv`
Full URI	`script://transforms/csv`
With version	`scripts/transforms/csv@v1.0.0`

# Shorthand format (recommended)
script: scripts/transform-data

# Full URI format
script: "script://transform-data"

# Nested paths
script: scripts/pipelines/etl/normalize

# With version (for edgit-managed components)
script: "script://transform-data@v1.0.0"

Script File Format

Scripts must export a default function that receives AgentExecutionContext:

// scripts/my-script.ts
import type { AgentExecutionContext } from '@ensemble-edge/conductor'

export default function myScript(context: AgentExecutionContext) {
  // Available in context:
  const {
    input,      // Input data from ensemble
    env,        // Environment bindings (KV, D1, etc.)
    logger,     // Logger instance
    request,    // HTTP request (if triggered via HTTP)
    cache,      // Cache utilities
  } = context

  // Your logic here
  const result = processData(input)

  // Return value becomes agent output
  return result
}

Async Scripts

Scripts can be async for operations that interact with storage:

// scripts/with-storage.ts
import type { AgentExecutionContext } from '@ensemble-edge/conductor'

export default async function withStorage(context: AgentExecutionContext) {
  const { env, input } = context

  // Access KV storage
  const cached = await env.KV?.get('my-key')
  if (cached) {
    return JSON.parse(cached)
  }

  const result = processData(input)
  await env.KV?.put('my-key', JSON.stringify(result))

  return result
}

TypeScript Types

Full TypeScript support with proper typing:

// scripts/typed-transform.ts
import type { AgentExecutionContext } from '@ensemble-edge/conductor'

interface InputData {
  items: Array<{
    id: string
    value: number
  }>
}

interface OutputData {
  processed: Array<{
    id: string
    normalized: number
  }>
  total: number
}

export default function typedTransform(
  context: AgentExecutionContext
): OutputData {
  const data = context.input as InputData

  const processed = data.items.map(item => ({
    id: item.id,
    normalized: item.value / 100
  }))

  return {
    processed,
    total: processed.length
  }
}

Directory Organization

Organize scripts by domain:

scripts/
├── transforms/
│   ├── csv.ts
│   ├── json.ts
│   └── normalize.ts
├── validators/
│   ├── email.ts
│   └── order.ts
├── pipelines/
│   ├── extract.ts
│   ├── transform.ts
│   └── aggregate.ts
├── auth/
│   ├── verify-token.ts
│   └── hash-password.ts
└── examples/           # Excluded with --no-examples flag
    ├── health-check.ts
    └── http/
        └── greet-user.ts

Multi-Step Workflows

Use multiple script agents for complex data pipelines:

name: data-pipeline

flow:
  - agent: extract
  - agent: transform
  - agent: validate
  - agent: aggregate

agents:
  - name: extract
    operation: code
    config:
      script: scripts/pipelines/extract
    input:
      data: $input.data

  - name: transform
    operation: code
    config:
      script: scripts/pipelines/transform
    input:
      extracted: $extract.output

  - name: validate
    operation: code
    config:
      script: scripts/pipelines/validate
    input:
      transformed: $transform.output

  - name: aggregate
    operation: code
    config:
      script: scripts/pipelines/aggregate
    input:
      validated: $validate.output.valid_items

input:
  data: []

output:
  result: ${aggregate.output}

With corresponding script files:

// scripts/pipelines/extract.ts
import type { AgentExecutionContext } from '@ensemble-edge/conductor'

export default function extract(context: AgentExecutionContext) {
  const { data } = context.input as { data: Array<Record<string, unknown>> }

  return data.map(item => ({
    id: item['ID'],
    name: item['Full Name'],
    amount: parseFloat(String(item['Amount']))
  }))
}

// scripts/pipelines/transform.ts
import type { AgentExecutionContext } from '@ensemble-edge/conductor'

interface ExtractedItem {
  id: string
  name: string
  amount: number
}

export default function transform(context: AgentExecutionContext) {
  const { extracted } = context.input as { extracted: ExtractedItem[] }

  return extracted.map(item => ({
    ...item,
    amount_cents: Math.round(item.amount * 100),
    processed_at: new Date().toISOString(),
    normalized_name: item.name.toUpperCase().trim()
  }))
}

// scripts/pipelines/validate.ts
import type { AgentExecutionContext } from '@ensemble-edge/conductor'

interface TransformedItem {
  id: string
  amount_cents: number
}

export default function validate(context: AgentExecutionContext) {
  const { transformed } = context.input as { transformed: TransformedItem[] }

  const valid = transformed.filter(item => item.id && item.amount_cents > 0)
  const invalid = transformed.filter(item => !item.id || item.amount_cents <= 0)

  return {
    valid_items: valid,
    invalid_items: invalid,
    success_rate: (valid.length / transformed.length) * 100
  }
}

// scripts/pipelines/aggregate.ts
import type { AgentExecutionContext } from '@ensemble-edge/conductor'

interface ValidatedItem {
  amount_cents: number
}

export default function aggregate(context: AgentExecutionContext) {
  const { validated } = context.input as { validated: ValidatedItem[] }

  return {
    total_items: validated.length,
    total_amount: validated.reduce((sum, item) => sum + item.amount_cents, 0),
    items: validated,
    processed_at: new Date().toISOString()
  }
}

Best Practices

1. Keep Scripts Focused

Each script should do one thing well:

// Good: Single responsibility
export default function calculateTotal(context: AgentExecutionContext) {
  const items = context.input as Array<{ price: number }>
  return { total: items.reduce((sum, item) => sum + item.price, 0) }
}

// Bad: Multiple responsibilities
export default function doEverything(context: AgentExecutionContext) {
  // Validate, transform, calculate, format... too much
}

2. Type Your Inputs

TypeScript helps catch errors at build time:

interface OrderInput {
  items: Array<{ price: number; quantity: number }>
  discount?: number
}

export default function processOrder(context: AgentExecutionContext) {
  const input = context.input as OrderInput
  // TypeScript helps catch errors
}

3. Handle Edge Cases

export default function processItems(context: AgentExecutionContext) {
  const { items } = context.input as { items?: unknown[] }

  // Handle missing or empty arrays
  if (!items || items.length === 0) {
    return { count: 0, items: [] }
  }

  return { count: items.length, items }
}

4. Add Error Handling

export default function parseJson(context: AgentExecutionContext) {
  const { json_string } = context.input as { json_string: string }

  try {
    const data = JSON.parse(json_string)
    return { success: true, data }
  } catch (error) {
    return {
      success: false,
      error: error instanceof Error ? error.message : 'Parse failed'
    }
  }
}

5. Don’t Mutate Input

// Good: Create new array
export default function sortItems(context: AgentExecutionContext) {
  const { items } = context.input as { items: number[] }
  const sorted = [...items].sort((a, b) => a - b)
  return { sorted }
}

// Bad: Mutates original
export default function sortItems(context: AgentExecutionContext) {
  const { items } = context.input as { items: number[] }
  items.sort() // Mutates input!
  return { sorted: items }
}

Testing Scripts

Test scripts with unit tests:

// scripts/__tests__/transform-data.test.ts
import transformData from '../transform-data'

describe('transformData', () => {
  it('normalizes values correctly', () => {
    const context = {
      input: {
        raw_data: [
          { id: '1', value: 50, min: 0, max: 100, created_at: '2024-01-01' }
        ]
      },
      env: {}
    } as any

    const result = transformData(context)

    expect(result.processed[0].normalized_value).toBe(0.5)
    expect(result.total_items).toBe(1)
  })
})

Integration with Other Operations

With Think Agents

agents:
  - name: prepare
    operation: code
    config:
      script: scripts/prepare-for-ai
    input:
      text: $input.text

  - name: analyze
    operation: think
    config:
      provider: anthropic
      model: claude-sonnet-4
      prompt: |
        Analyze this prepared data:
        ${prepare.output}

With Data Agents

agents:
  - name: fetch
    operation: data
    config:
      database: d1
      query: "query://get-users@v1"

  - name: process
    operation: code
    config:
      script: scripts/process-users
    input:
      users: $fetch.output.rows

output:
  stats: ${process.output}

Versioning with Edgit

For production deployments, use edgit to version and manage scripts:

# Add script to edgit
edgit components add transform-data scripts/transform-data.ts script

# Create version tag
edgit tag create transform-data v1.0.0

# Tag for production
edgit tag set transform-data production v1.0.0
edgit push --tags --force

Reference versioned scripts:

agents:
  - name: transform
    operation: code
    config:
      script: "script://transform-data@v1.0.0"

Using ctx API in Agents

When building custom agents with TypeScript handlers, you can access and execute other scripts through the ctx API:

ctx.scripts.get(name)

Get and execute a script by name:

// agents/orchestrator/index.ts
import type { AgentExecutionContext } from '@ensemble-edge/conductor'

export default async function orchestrate(ctx: AgentExecutionContext) {
  // Get and execute another script
  const transformScript = await ctx.scripts.get('transform-data')

  // Execute the script with custom input
  const result = await transformScript({
    ...ctx,
    input: { data: [1, 2, 3, 4, 5] }
  })

  return {
    transformed: result,
    original: ctx.input
  }
}

Composing Multiple Scripts

// agents/pipeline/index.ts
import type { AgentExecutionContext } from '@ensemble-edge/conductor'

export default async function runPipeline(ctx: AgentExecutionContext) {
  const { data } = ctx.input as { data: any[] }

  // Load multiple scripts
  const extract = await ctx.scripts.get('pipelines/extract')
  const transform = await ctx.scripts.get('pipelines/transform')
  const validate = await ctx.scripts.get('pipelines/validate')

  // Execute pipeline
  const extracted = await extract({ ...ctx, input: { data } })
  const transformed = await transform({ ...ctx, input: { extracted } })
  const validated = await validate({ ...ctx, input: { transformed } })

  return validated
}

Dynamic Script Loading

// agents/dynamic-processor/index.ts
import type { AgentExecutionContext } from '@ensemble-edge/conductor'

export default async function processWithScript(ctx: AgentExecutionContext) {
  const { processorType, data } = ctx.input as {
    processorType: 'csv' | 'json' | 'xml'
    data: string
  }

  // Load processor based on type
  const scriptName = `transforms/${processorType}`
  const processor = await ctx.scripts.get(scriptName)

  const result = await processor({ ...ctx, input: { data } })

  return {
    type: processorType,
    result
  }
}

Script with Shared Utilities

// agents/with-utilities/index.ts
import type { AgentExecutionContext } from '@ensemble-edge/conductor'

export default async function processWithUtils(ctx: AgentExecutionContext) {
  const { items } = ctx.input as { items: any[] }

  // Load utility scripts
  const validate = await ctx.scripts.get('utils/validate')
  const normalize = await ctx.scripts.get('utils/normalize')

  // Use utilities
  const validated = await Promise.all(
    items.map(item => validate({ ...ctx, input: { item } }))
  )

  const normalized = await Promise.all(
    validated.map(item => normalize({ ...ctx, input: { item } }))
  )

  return {
    processed: normalized,
    total: items.length
  }
}

Error Handling with Scripts

// agents/safe-processor/index.ts
import type { AgentExecutionContext } from '@ensemble-edge/conductor'

export default async function safeProcess(ctx: AgentExecutionContext) {
  const { data } = ctx.input as { data: any }

  try {
    // Try primary processor
    const processor = await ctx.scripts.get('processors/primary')
    const result = await processor({ ...ctx, input: { data } })

    return {
      success: true,
      result,
      processor: 'primary'
    }
  } catch (error) {
    // Fallback to backup processor
    const fallback = await ctx.scripts.get('processors/fallback')
    const result = await fallback({ ...ctx, input: { data } })

    return {
      success: true,
      result,
      processor: 'fallback',
      primaryError: error instanceof Error ? error.message : 'Unknown error'
    }
  }
}

Next Steps

Code Operation

Full code operation docs

Prompt Components

Reusable AI instructions

Query Components

SQL queries as components

Creating Agents

Build custom agents

​Overview

​Quick Start

​1. Create a Script File

​2. Reference in Your Ensemble

​3. Build & Deploy

​URI Formats

​Script File Format

​Async Scripts

​TypeScript Types

​Directory Organization

​Multi-Step Workflows

​Best Practices

​1. Keep Scripts Focused

​2. Type Your Inputs

​3. Handle Edge Cases

​4. Add Error Handling

​5. Don’t Mutate Input

​Testing Scripts

​Integration with Other Operations

​With Think Agents

​With Data Agents

​Versioning with Edgit

​Using ctx API in Agents

​ctx.scripts.get(name)

​Composing Multiple Scripts

​Dynamic Script Loading

​Script with Shared Utilities

​Error Handling with Scripts

​Next Steps

Code Operation

Prompt Components

Query Components

Creating Agents

Overview

Quick Start

1. Create a Script File

2. Reference in Your Ensemble

3. Build & Deploy

URI Formats

Script File Format

Async Scripts

TypeScript Types

Directory Organization

Multi-Step Workflows

Best Practices

1. Keep Scripts Focused

2. Type Your Inputs

3. Handle Edge Cases

4. Add Error Handling

5. Don’t Mutate Input

Testing Scripts

Integration with Other Operations

With Think Agents

With Data Agents

Versioning with Edgit

Using ctx API in Agents

ctx.scripts.get(name)

Composing Multiple Scripts

Dynamic Script Loading

Script with Shared Utilities

Error Handling with Scripts

Next Steps