workflow-builder

Workflow Builder

When the user asks you to create, build, or generate a workflow, use this skill to produce a correct .py workflow file and save it to the workflows directory.

Workflow file format

Every workflow file has:

1. A top-level meta dict with name, description, when_to_use, and optional phases list.
2. An async def run() entry point.
3. DSL globals injected at runtime — never import them.

Available DSL globals:

Global	Purpose
await agent(prompt, schema=None, system=None, tools=None, resume=False, stall_ms=180000, max_retries=5, model=None, provider=None)	Full subagent turn with tool execution. Returns str or Pydantic instance.
await classify(prompt, *, options=None, schema=None, system=None, model=None, provider=None)	Single direct LLM call — no tool loop. Use for routing/labelling. Returns str (options) or Pydantic instance (schema).
await parallel(*thunks, concurrency=5, return_exceptions=False)	Run zero-arg async callables concurrently.
await pipeline(items, *stages, concurrency=5)	Pass each item through staged transforms.
await workflow(name, args=None)	Run another workflow inline (one level deep).
async with phase("name"):	Label the current phase in run status.
log("message")	Append timestamped line to run log.
budget	.spent() / .remaining() / .exhausted() / .tokens_spent() — call count + token tracking.
args	dict of invocation arguments.

Pattern catalogue

Pick the right pattern based on the user's description. Generate the full .py file for the chosen pattern.

---

1. Classify-and-act

Route input to specialized handlers based on a cheap single-call classification.

meta = {
    "name": "classify-and-act",
    "description": "Route a task to a specialized handler based on its type.",
    "when_to_use": "Input could be one of several distinct task types each needing different handling.",
    "phases": [
        {"name": "classify", "description": "Determine task type"},
        {"name": "execute",  "description": "Run the appropriate handler"},
    ],
}

async def run():
    task = args.get("task", "")

    async with phase("classify"):
        kind = await classify(
            f"Classify this task into one category.\nTask: {task}",
            options=["research", "code", "summarize", "other"],
            model="claude-haiku-4-5",
        )
        log(f"classified as: {kind}")

    async with phase("execute"):
        if kind == "research":
            return await agent(f"Research the following thoroughly:\n{task}")
        elif kind == "code":
            return await agent(f"Write code to accomplish:\n{task}")
        elif kind == "summarize":
            return await agent(f"Summarize the following:\n{task}")
        else:
            return await agent(task)