dynamic-workflows

name	dynamic-workflows
description	Orchestrate many Codex subagents deterministically from one JavaScript script. Use when a task splits into independent units that should run in parallel or through staged pipelines (reviewing/enriching/migrating a list, multi-angle research, fan-out then merge) and you want loops/concurrency/merging handled in code rather than by hand.

Dynamic Workflows

Call the run_workflow tool (from the dynamic-workflows MCP server) with either a script (plain JavaScript) or a script_file, plus optional args. Inline scripts are archived as <workflows_dir>/scripts/<run_id>.js and returned as scriptFile; script_file accepts a .js path or a bare name under workflows_dir. The script runs in an async context and orchestrates subagents. run_workflow runs in the background; use the returned runId with get_workflow_result until it returns status: "completed" or status: "failed".

The whole point of this tool is the agent() fan-out. Each agent() call spawns a fresh Codex subagent (via @openai/codex-sdk) that shares no context with other agents or with you, so pass everything an agent needs in its prompt. If your script would call agent() zero times, do not use this tool - answer or compute the result directly.

Helpers in scope

• agent(prompt, opts?) -> result. opts: { schema, model, label, phase, cd, agentType, isolation, resumeThread }. With schema (a JSON Schema) the agent must return matching JSON and you get the parsed object; without it you get the agent's final text. Throws if that agent fails.
• parallel(thunks) -> any[]. Runs thunks concurrently under a global cap; a throwing thunk becomes null (use .filter(Boolean)).
• pipeline(items, ...stages) -> any[]. Each item flows through all stages independently (no barrier). Stages receive (prev, originalItem, index). A throwing stage drops that item to null.
• phase(title) / log(msg) for progress. args is the tool's args input.

Shape

export const meta = { name: 'review', description: 'review each changed file' }

const SCHEMA = {
  type: 'object',
  properties: { file: { type: 'string' }, issues: { type: 'array', items: { type: 'string' } } },
  required: ['file', 'issues'],
}

phase('review')
const results = await parallel(
  args.files.map((f) => () =>
    agent(`Review ${f} for correctness bugs. Return JSON.`, { schema: SCHEMA, label: f })
  )
)
return results.filter(Boolean)

End the script with return <value>; that value is the workflow output returned by get_workflow_result. Default to parallel for independent work and pipeline when each item has multiple dependent stages.

When NOT to use

If you can produce the answer yourself - a single lookup, a calculation, formatting data you already have - do it directly; do not call run_workflow. A workflow that finishes with agentCount: 0 (no agent() calls) added nothing and is a misuse. Reach for run_workflow only when real subagent fan-out - parallel agent() calls, staged pipelines, or loops over many units - is worth expressing as code.