SDK

import { defineStep, defineBatchStep, drop, hold, soft, defineSource, SourceError, choice, score, noul } from "@motionskit/sdk";
import { runStep, loadFixtures } from "@motionskit/sdk/testing";

Steps

defineStep

function defineStep<S extends StandardSchema | undefined>(def: {
  output?: S;
  run(company: Company, ctx: StepContext): StepReturn<Infer<S>> | void | Promise<StepReturn<Infer<S>> | void>;
}): CompanyStepDef<S>;

The default export of a company step’s file. output is any Standard Schema (zod 4, valibot, arktype); when given, facts are typed by it and checked against it after the run.

defineBatchStep

function defineBatchStep(def: {
  run(companies: Company[], ctx: StepContext): Record<string, StepReturn> | void | Promise<Record<string, StepReturn> | void>;
}): BatchStepDef;

The default export of a batch step’s file. Returns a result per company key. See Batch steps.

Company

type Company = {
  key: string;
  core: {
    name: string;
    domain: string | null;
    country: string | null;
    tax_id: string | null;
    crm: { system: string; object: "company"; id: string; owner_id: string | null };
  };
  fields: Record<string, any>;   // crm.yaml's custom fields, by short name
  facts: Record<string, any>;    // earlier steps' facts, by step id
  answers: Record<string, string>; // this step's answered questions: type → option id
};

StepReturn

type StepReturn<F = JsonInput> = {
  facts?: F;
  outcome?: { lane: "continue" | "hold" | "drop" | "done"; reason: string };
  questions?: QuestionInput[];   // non-blocking unless blocking: true
  events?: { type: string; data?: Json }[];
};

const drop: (reason: string) => StepReturn<never>;  // { outcome: { lane: "drop", reason } }
const hold: (reason: string) => StepReturn<never>;  // { outcome: { lane: "hold", reason } }

Keys whose value is undefined are dropped from facts.

StepContext

type StepContext = {
  batch: { id: string; tenant: string; motion: string; label: string; openedAt: Date };
  step: string;
  rules: Record<string, any>;               // rules.yaml
  sources: Sources;                         // ctx.sources.<provider>.<op>(request), or ctx.sources.call(op, request)
  crm: Crm;
  calendar: Record<string, number>;         // the motion's calendar, as epoch milliseconds
  ask(question: QuestionInput): Promise<string>;
  answer(type: string, company?: string): string | undefined;
  rerun: string[];                          // the company keys this run's results count for
  wait(every: string, reason: string): never;
  now(): Date;
};

Member
ask	Returns the answer’s option id if the question was answered; otherwise stops the step until it is. A low-risk suggest answers it at once. Not for batch steps.
answer	This step’s answer to a question type, without asking. In a batch step, pass the company key.
wait	Stops the step and runs it again after every, such as "10m". The company is parked until then.
now	When this run was dispatched. Date.now() and new Date() return the same.

QuestionInput

type QuestionInput = {
  type: string;            // the queue
  prompt: string;
  options: { id: string; text: string; effect?: "continue" | "hold" | "drop" }[];  // 2 to 4
  recommended?: string;    // an option id
  evidence?: { source: string; shows: string; ref?: string }[];
  risk?: "low" | "high";   // default "low"
  blocking?: boolean;      // for questions in a return value; ctx.ask always blocks
  suggest?: { answer: string; risk: "low" | "high"; evidence?: Evidence[]; note?: string };
};

suggest.answer is an option id, or $hold, $drop or $continue.

soft

function soft<T>(call: Promise<T>): Promise<{ ok: true; value: T } | { ok: false; error: string }>;

Turns a refused source call (no key, not allowed, bad input) into a value instead of failing the step. Budget, retry, ask and wait signals pass through.

CRM

interface Crm {
  companies: {
    get(id: string): Promise<CrmCompany | null>;
    search(q: { tax_id?: string; tax_root?: string; domain?: string; name?: string; limit?: number }): Promise<CrmCompany[]>;
    update(id: string, values: { fields: Record<string, any> }): Promise<OpRef>;
  };
  contacts: {
    list(q: { company: string }): Promise<CrmContact[]>;
    search(q: { email?: string; linkedin?: string; name?: string }): Promise<CrmContact[]>;
    create(values: ContactWrite): Promise<OpRef>;
    update(id: string | OpRef, values: ContactWrite): Promise<OpRef>;
  };
  deals: { list(q: { company: string; open?: boolean; limit?: number }): Promise<CrmDeal[]> };
  owners: { get(id: string): Promise<CrmOwner | null>; list(): Promise<CrmOwner[]> };
  notes: {
    list(q: { company?: string; contact?: string }): Promise<CrmNote[]>;
    create(values: { on: string | OpRef; body: string }): Promise<OpRef>;
    pin(note: string | OpRef): Promise<OpRef>;
  };
  associate(contact: string | OpRef, company: string | OpRef): Promise<OpRef>;
}

type CrmCompany = { id: string; name: string; domain: string | null; tax_id: string | null; owner_id: string | null; fields: Record<string, any> };
type CrmContact = { id: string; first_name: string | null; last_name: string | null; email: string | null; title: string | null;
                    linkedin: string | null; owner_id: string | null; company_ids: string[]; fields: Record<string, any> };
type CrmDeal = { id: string; name: string; stage: string | null; amount: number | null; is_closed: boolean; owner_id: string | null;
                 company_ids: string[]; fields: Record<string, any> };
type CrmOwner = { id: string; name: string; email: string | null };
type CrmNote = { id: string; body: string; pinned: boolean; on: { object: string; id: string } };
type ContactWrite = { first_name?: string | null; last_name?: string | null; email?: string | null; title?: string | null;
                      linkedin?: string | null; fields?: Record<string, any> };
type OpRef = { $op: string };   // a proposed operation, usable wherever an id goes

Reads need the object in the step’s crm.read; writes need the action in crm.propose. See Read and propose CRM changes.

Built-in sources

Typed requests and results for the catalog:

ctx.sources.rdap.lookup({ domain })
  // → { domain, registrant: string | null, registeredAt: string | null, changedAt: string | null, status: string[], server } | null
ctx.sources.cnpj.match({ name?, cnpj?, domain?, city?, uf? })
  // → { status: "matched" | "review" | "unresolved", cnpj?, cnpj_root?, razao_social?, confidence?, reason, candidates: [...] }
ctx.sources.cnpj.company({ id, include?: ("establishments" | "socios")[] })   // → registry record | null
ctx.sources.cnpj.search({ q, uf?, municipio?, limit? })                      // → { results: [...] }
ctx.sources.exa.search({ query, numResults?, type?: "auto" | "fast" | "neural" | "keyword" })
  // → { results: { title, url, publishedDate?, text? }[] }
ctx.sources.llm.complete({ key?, model?, system?, prompt?, messages?, maxTokens?, temperature? })
  // → { text, stopReason, model }
ctx.sources.typesafe.ask({ state, questions: { name: choice(...) | score(...) | noul(...) }, model? })
  // → per question: choice → { choice, probabilities, confidence }; score → { score, probabilities, confidence }; noul → number
ctx.sources.harvestapi.lead_search(req)         // LinkedIn lead search, one page
ctx.sources.harvestapi.profile(req)             // a LinkedIn profile
ctx.sources.fullenrich.enrich_bulk({ name, contacts })        // → { enrichment_id }
ctx.sources.fullenrich.enrich_bulk_get({ enrichment_id })    // → { status: "pending" | "done", contacts? }
ctx.sources.apollo.people_match({ first_name?, last_name?, name?, domain?, organization_name?, linkedin_url? })
  // → { email, title, linkedin_url, ... } | null

Your editor shows the full types of each request and result.

Custom sources

defineSource

function defineSource(def: {
  id: string;                    // the folder's name
  title?: string;
  hosts: readonly string[];      // the only hosts fetch may reach, over https
  secrets?: readonly string[];   // names used as {{secrets.NAME}}
  ops: Record<string, {
    input: StandardSchema;
    output?: StandardSchema;
    ttl?: `${number}${"s" | "m" | "h" | "d"}` | number;   // cache time; default: no cache
    estimateUsd?(input): number;                          // the most a call can cost
    call(input, ctx: { fetch: typeof fetch }): Promise<{ data: unknown; usd?: number; cache?: false }>;
  }>;
}): SourceDefinition;

SourceError

new SourceError(cls: "not_found" | "bad_input" | "not_allowed" | "no_credit" | "budget" | "rate_limited" | "transient" | "config",
                message: string, retryAfterMs?: number)

Thrown from call to fail with that class. See Write a custom source.

Testing

From @motionskit/sdk/testing.

runStep

function runStep(step: CompanyStepDef, opts: RunOptions & { company?: TestCompany }):
  Promise<StepRunResult & { calls: SourceCallRecord[] }>;
function runStep(step: BatchStepDef, opts: RunOptions & { companies: TestCompany[]; rerun?: string[] }):
  Promise<{ results: Record<string, StepRunResult>; calls: SourceCallRecord[] }>;

type RunOptions = {
  fixtures?: FixtureEntry[] | string;  // entries, or a fixtures/sources folder; default: the tenant's
  crm?: { companies?: CrmCompany[]; contacts?: CrmContact[]; deals?: CrmDeal[]; owners?: CrmOwner[]; notes?: CrmNote[] };
  rules?: Record<string, Json>;
  calendar?: Record<string, number>;
  now?: Date | string | number;        // default: the real time
  step?: string;                       // the step id the run reports
  fixtureCalls?: Map<string, number>;  // share between runs to walk a fixture sequence
};

type TestCompany = { key?: string; core?: Partial<Company["core"]>; fields?: Record<string, Json>;
                     facts?: Record<string, Json>; answers?: Record<string, string> };

type StepRunResult =
  | { status: "ok"; facts?; outcome?; questions?; events?; proposals?; logs; ms }
  | { status: "asked"; question: QuestionInput; logs; ms }
  | { status: "waiting"; until: number; reason: string; logs; ms }
  | { status: "parked"; reason: string; logs; ms }
  | { status: "failed"; error: string; transient: boolean; logs; ms };

type SourceCallRecord = { op: string; request: Json; reply: { ok: true; data: Json } | { ok: false; error: { class: string; message: string } } };

loadFixtures

function loadFixtures(dir?: string): FixtureEntry[];

Reads a fixtures/sources folder, by default the tenant’s, in the order calls match them.