Workflows

A Workflow is a named, versioned orchestration definition — a sequence of steps that run in order (or in parallel) with inputs, outputs, conditions, and error handling baked in. Trigger a workflow on demand, on a schedule, or in response to an EDM event.

Workflows run server-side. Each run produces a Job that you can monitor and query.

Workflow Schema

Field	Type	Description
`id`	`string`	Prefixed ID: `wfl-{ulid}`
`name`	`string`	Unique name for the workflow
`description`	`string`	What this workflow does
`version`	`number`	Auto-incremented on each update
`status`	`enum`	`active` \| `draft` \| `disabled`
`trigger`	`object`	How the workflow starts (manual, schedule, or event)
`steps`	`array`	Ordered list of step definitions
`createdAt`	`string`	ISO 8601 UTC

List & Get Workflows

import { CpodClient } from '@cpod/sdk'
const sdk = CpodClient.fromEnv()
 
const workflows = await sdk.workflows.list({ status: 'active' })
for (const wf of workflows.items) {
  console.log(wf.name, wf.version, wf.trigger.type)
}
 
const workflow = await sdk.workflows.get('wfl-abc123')

from cpod import CpodClient
sdk = CpodClient.from_env()
 
workflows = await sdk.workflows.list(status="active")
for wf in workflows.items:
    print(wf.name, wf.version)
 
workflow = await sdk.workflows.get("wfl-abc123")

import "github.com/cpod-ai/cpod-sdk-go/workflows"
 
client := cpod.New(os.Getenv("CPOD_API_KEY"))
 
result, err := client.Workflows.List(ctx, workflows.ListOptions{Status: "active"})
workflow, err := client.Workflows.Get(ctx, "wfl-abc123")

Create a Workflow

const workflow = await sdk.workflows.create({
  name: 'quarterly-access-review',
  description: 'Run access review skill, notify owners, export report.',
  trigger: {
    type: 'schedule',
    cron: '0 9 1 */3 *',  // 09:00 UTC on the 1st of every quarter
  },
  steps: [
    {
      id: 'run-review',
      type: 'skill',
      skill: 'access-review',
      input: { groupId: 'grp-all-staff', lookbackDays: 90 },
    },
    {
      id: 'notify-owners',
      type: 'webhook',
      url: 'https://hooks.acme.com/access-review',
      method: 'POST',
      body: { jobId: '{{steps.run-review.jobId}}' },
      dependsOn: ['run-review'],
    },
  ],
})
// workflow.id → "wfl-01HXYZ..."

workflow = await sdk.workflows.create(
    name="quarterly-access-review",
    description="Run access review skill, notify owners, export report.",
    trigger={"type": "schedule", "cron": "0 9 1 */3 *"},
    steps=[
        {
            "id": "run-review",
            "type": "skill",
            "skill": "access-review",
            "input": {"group_id": "grp-all-staff", "lookback_days": 90},
        },
        {
            "id": "notify-owners",
            "type": "webhook",
            "url": "https://hooks.acme.com/access-review",
            "method": "POST",
            "body": {"job_id": "{{steps.run-review.job_id}}"},
            "depends_on": ["run-review"],
        },
    ],
)

workflow, err := client.Workflows.Create(ctx, workflows.CreateInput{
    Name:        "quarterly-access-review",
    Description: "Run access review skill, notify owners.",
    Trigger: workflows.Trigger{
        Type: "schedule",
        Cron: "0 9 1 */3 *",
    },
    Steps: []workflows.Step{
        {
            ID:    "run-review",
            Type:  "skill",
            Skill: "access-review",
            Input: map[string]any{"groupId": "grp-all-staff", "lookbackDays": 90},
        },
    },
})

Trigger a Workflow

// Trigger immediately with optional input overrides
const run = await sdk.workflows.trigger('wfl-abc123', {
  input: { groupId: 'grp-engineering' },
})
// run.jobId → "job-01HXYZ..."
 
// Wait for the run to finish
const result = await sdk.jobs.wait(run.jobId)
// result.output → { steps: { 'run-review': {...}, 'notify-owners': {...} } }

run = await sdk.workflows.trigger("wfl-abc123", input={"group_id": "grp-engineering"})
 
result = await sdk.jobs.wait(run.job_id)

run, err := client.Workflows.Trigger(ctx, "wfl-abc123", workflows.TriggerInput{
    Input: map[string]any{"groupId": "grp-engineering"},
})
 
result, err := client.Jobs.Wait(ctx, run.JobID, 120000)

Monitor Runs

// List recent runs for a workflow
const runs = await sdk.workflows.runs('wfl-abc123', { limit: 10 })
for (const run of runs.items) {
  console.log(run.jobId, run.status, run.startedAt, run.completedAt)
}
 
// Update (disable a workflow)
await sdk.workflows.update('wfl-abc123', { status: 'disabled' })
 
// Delete
await sdk.workflows.delete('wfl-abc123')

runs = await sdk.workflows.runs("wfl-abc123", limit=10)
for run in runs.items:
    print(run.job_id, run.status, run.started_at)
 
await sdk.workflows.update("wfl-abc123", status="disabled")
await sdk.workflows.delete("wfl-abc123")

runs, err := client.Workflows.Runs(ctx, "wfl-abc123", 10, "")
for _, run := range runs.Items {
    fmt.Println(run.JobID, run.Status)
}
 
_, err = client.Workflows.Update(ctx, "wfl-abc123", workflows.UpdateInput{Status: "disabled"})
err = client.Workflows.Delete(ctx, "wfl-abc123")

Workflow run history is retained for 30 days. Workflow definitions persist until explicitly deleted.

Skills Jobs