Typed input

Every pipeline starts with a Zod schema. The schema validates untrusted input before the first stage and determines the type received by that stage.

1. Define the workflow contract

import { Pipeline } from '@anvia/core/pipeline';
import { z } from 'zod';
const TicketInput = z.object({
    customer: z.string().min(1),
    subject: z.string().min(1),
    body: z.string().min(1),
});
const normalizeTicket = new Pipeline({
    id: 'normalize-ticket',
    inputSchema: TicketInput,
})
    .step({
    id: "step-1",
    run: ({ input: ticket }) => ({
        customer: ticket.customer.trim(),
        subject: ticket.subject.trim(),
        body: ticket.body.trim().replace(/\s+/g, ' '),
    })
});

TypeScript infers the parsed schema type for ticket. Invalid input rejects run() with a ZodError before the step executes.

2. Run the pipeline

const result = await normalizeTicket.run({
    input: {
        customer: 'Acme Co.',
        subject: 'Checkout failure',
        body: 'Checkout fails after payment authorization.',
    }
});
console.log(result.output.subject);

run() resolves to a { runId, output, trace? } result. output holds the last stage's return value, so its type is inferred from that stage; runId identifies the execution in logs and traces, and trace is present only when the run was traced. Keep route-specific transport validation in the route, and use the pipeline schema for the workflow contract every caller must satisfy.

3. Use schema transforms and defaults

The first stage receives the Zod output, which may differ from the accepted input:

const searchPipeline = new Pipeline({
    id: 'search',
    inputSchema: z.object({
        query: z.string().transform((value) => value.trim()),
        limit: z.number().int().positive().default(10),
    }),
})
    .step({
    id: "step-1",
    run: ({ input: input }) => (({ query, limit }) => searchIndex.query(query, limit))(input)
});

Parsing runs once for each pipeline execution.

4. Add workflow metadata

const pipeline = new Pipeline({
  id: 'ticket-triage',
  name: 'Ticket triage',
  description: 'Normalize and route support tickets.',
  metadata: { owner: 'support' },
  inputSchema: TicketInput,
})

The ID is required and must be non-empty. Names, descriptions, and JSON metadata appear in graph inspection and help operational tooling explain the workflow.

Next, add pipeline steps.