Write a custom source

When the catalog doesn’t have what a motion needs, such as a client’s internal API or a public registry, write a custom source. It’s a TypeScript file in the tenant repository. Steps call it like a built-in one, and it runs sandboxed, reaching only the hosts it declares.

The file

Each source is a folder under sources/, named after the source’s id, with an index.ts that exports defineSource:

• Directorysources/

  • Directorybrasilapi/

    • index.ts
    • index.test.ts

sources/brasilapi/index.ts

import { defineSource, SourceError } from "@motionskit/sdk";
import * as z from "zod/mini";

type Reply = { razao_social: string; nome_fantasia?: string | null; cnae_fiscal_descricao?: string | null; municipio?: string | null; uf?: string | null };

/** Brazil's public company registry, through BrasilAPI. Free and keyless, so it declares no secrets. */
export default defineSource({
  id: "brasilapi",
  title: "BrasilAPI",
  hosts: ["brasilapi.com.br"],
  ops: {
    cnpj: {
      input: z.object({ cnpj: z.string() }),
      ttl: "7d",
      estimateUsd: () => 0,
      async call({ cnpj }, { fetch }) {
        const digits = cnpj.replace(/\D/g, "");
        if (digits.length !== 14) throw new SourceError("bad_input", `a CNPJ has 14 digits, got ${digits.length}`);
        const res = await fetch(`https://brasilapi.com.br/api/cnpj/v1/${digits}`);
        if (res.status === 404) throw new SourceError("not_found", `no company has CNPJ ${digits}`);
        if (res.status === 429) throw new SourceError("rate_limited", "BrasilAPI is rate limiting");
        if (!res.ok) throw new SourceError(res.status >= 500 ? "transient" : "bad_input", `BrasilAPI answered ${res.status}`);
        const body = (await res.json()) as Reply;
        return {
          data: {
            legal_name: body.razao_social,
            trade_name: body.nome_fantasia || null,
            activity: body.cnae_fiscal_descricao ?? null,
            city: body.municipio ?? null,
            uf: body.uf ?? null,
          },
        };
      },
    },
  },
});

Field
id	The source’s name in steps: brasilapi.cnpj. Lowercase, and the same as the folder’s name.
title	What the app shows.
hosts	The only hosts fetch may reach, over HTTPS.
secrets	The names of the secrets it uses, set in Settings → Secrets.
ops	One entry per operation.

Each operation has:

Field
input	A schema for the request, with zod or any Standard Schema library. A call that doesn’t match fails with bad_input before your code runs.
output	Optional: a schema for the data, which also types it for steps.
call(input, { fetch })	Does the work, and returns { data }. Add usd for what the call cost, and cache: false for an answer that mustn’t be cached.
ttl	How long an answer is reused, such as "7d" or "12h". Without it, nothing is cached.
estimateUsd(input)	The most a call can cost. Motions checks it against the step’s cap before calling.

Errors

Throw a SourceError with a class, and Motions treats it like a built-in source’s error:

throw new SourceError("not_found", `no company has CNPJ ${digits}`);      // the step gets null
throw new SourceError("rate_limited", "slow down", 30_000);                // retried, after 30 seconds
throw new SourceError("transient", `the API answered ${res.status}`);      // retried
throw new SourceError("bad_input", "a CNPJ has 14 digits");               // fails the step

Any other exception fails the step with its message.

Secrets

Declare each secret by name, and put {{secrets.NAME}} where it goes in the URL, a header or the body. Motions fills it in on the way out; your code never sees the value.

export default defineSource({
  id: "crmplus",
  hosts: ["api.crmplus.example"],
  secrets: ["CRMPLUS_TOKEN"],
  ops: {
    account: {
      input: z.object({ domain: z.string() }),
      ttl: "1d",
      async call({ domain }, { fetch }) {
        const res = await fetch(`https://api.crmplus.example/accounts?domain=${encodeURIComponent(domain)}`, {
          headers: { authorization: "Bearer {{secrets.CRMPLUS_TOKEN}}" },
        });
        if (!res.ok) throw new SourceError(res.status >= 500 ? "transient" : "bad_input", `crmplus answered ${res.status}`);
        return { data: await res.json() };
      },
    },
  },
});

After the push, set the value in Settings → Secrets. Until it’s set, calls that use it fail, and the page lists it under Missing secrets.

Use it from a step

Add the operation to the step’s sources, then call it:

    sources: [brasilapi.cnpj]

const registry = await ctx.sources.brasilapi.cnpj({ cnpj: company.core.tax_id! });

motions check writes .motions/sources.d.ts, which types ctx.sources.brasilapi from the source’s own schemas. Include .motions/*.d.ts in tsconfig.json, and add .motions/ to .gitignore: it’s regenerated on every check.

Publish it

motions push publishes each source before the motions, as a new version when it changed. Like motions, a source version is a draft until it’s live: push with --live, or choose Make live on the source’s page under Sources. Steps call the live version.

Test it

A source is a plain object, so test its operations directly, with a stand-in fetch:

sources/brasilapi/index.test.ts

import { expect, test } from "vitest";
import brasilapi from "./index.ts";

/** A stand-in for the Egress fetch: answers every request with `status` and `body`, and remembers the URLs. */
function fakeFetch(status: number, body: unknown) {
  const urls: string[] = [];
  const fetch = (async (url: string | URL | Request) => {
    urls.push(String(url));
    return new Response(JSON.stringify(body), { status });
  }) as typeof globalThis.fetch;
  return { fetch, urls };
}

const lookup = brasilapi.ops.cnpj;

test("looks a company up by its CNPJ digits", async () => {
  const { fetch, urls } = fakeFetch(200, { razao_social: "EMPRESA EXEMPLO LTDA", nome_fantasia: "", municipio: "SAO PAULO", uf: "SP" });
  const out = await lookup.call({ cnpj: "11.222.333/0001-81" }, { fetch });
  expect(urls).toEqual(["https://brasilapi.com.br/api/cnpj/v1/11222333000181"]);
  expect(out.data).toEqual({ legal_name: "EMPRESA EXEMPLO LTDA", trade_name: null, activity: null, city: "SAO PAULO", uf: "SP" });
});

test("refuses a malformed CNPJ without calling out", async () => {
  const { fetch, urls } = fakeFetch(200, {});
  await expect(lookup.call({ cnpj: "123" }, { fetch })).rejects.toMatchObject({ cls: "bad_input" });
  expect(urls).toEqual([]);
});

test("a CNPJ BrasilAPI doesn't know is not_found, which a step reads as null", async () => {
  const { fetch } = fakeFetch(404, { message: "CNPJ não encontrado" });
  await expect(lookup.call({ cnpj: "11222333000181" }, { fetch })).rejects.toMatchObject({ cls: "not_found" });
});

Run source tests with vitest

motions test only runs the tests under motions/*/tests/. Run a source’s tests with npx vitest run sources.

Limits

• Network: HTTPS to the declared hosts only. Anything else is refused.
• Per call: 10 seconds of CPU and 50 requests.
• Fixture mode doesn’t apply: a custom source calls its real API even when the tenant is in fixture mode. In step tests, runStep answers custom sources from fixtures like any other.