@clamator/codegen

CLI plus library that turns a Zod contract module into TypeScript and Python client/server wrappers for clamator.

Install

npm install -D @clamator/codegen
npm install zod @clamator/protocol

⚠️ Required: declare zod and @clamator/protocol in your own package.json.

Codegen reads your contract source files, which import defineContract/defineMethod/defineNotification from @clamator/protocol and type-parameterize over zod. Both are peer dependencies. Add "zod": "^3.23.0" and "@clamator/protocol" (matching version) to your package's dependencies for type-dedupe across your workspace. See @clamator/protocol's install section for the rationale.

CLI usage

npx @clamator/codegen \
  --src <contracts-dir> \
  --out-ts <ts-output-dir> \
  --out-py <py-output-dir> \
  --manifest <manifest.json> \
  --ts-contract-import <import-path>

The interop test runner invokes the CLI like this:

  const args = [
    codegenCli,
    '--src', contractsSrc,
    '--out-ts', outTs,
    '--out-py', outPy,
    '--manifest', manifestPath,
    '--ts-contract-import', '../../contracts/index.js',
  ];

(Verbatim from tests/interop/lib/runner.ts:291-298. codegenCli is the path to dist/cli.js of this package.)

Pass --out-py only when you want Python output. The Python emitter requires the datamodel-code-generator Python tool on PATH.

Flag conditionality. --ts-contract-import is required only when --out-ts is set (it controls the import path written into emitted TS wrappers); omit it for Python-only runs. --manifest is optional — the manifest file is only useful for the drift-detection workflow described below; codegen emits TS / Py wrappers correctly whether or not it's passed.

Contract input shape

A contract module exports one or more contracts via defineContract from @clamator/protocol:

import { z } from 'zod';
import { defineContract, defineMethod } from '@clamator/protocol';

export const arithContract = defineContract('arith', {
  add: defineMethod({
    params: z.object({ a: z.number().int(), b: z.number().int() }),
    result: z.object({ sum: z.number().int() }),
  }),
  divide: defineMethod({
    params: z.object({ a: z.number().int(), b: z.number().int() }),
    result: z.object({ q: z.number(), r: z.number().int() }),
  }),
});

(Verbatim from ts/packages/codegen/tests/fixtures/contracts/arith.ts.)

The codegen scans every .ts file in --src for defineContract calls and emits one wrapper file per contract.

Output layout

Given a --src directory containing contract modules and an --out-ts <dir> and --out-py <dir>:

• <out-ts>/<service>.ts — typed client and server wrappers for each contract; importable from a TS package.
• <out-py>/<service>.py — typed client and server wrappers for each contract; importable from a Python package.
• <manifest>.json — content-addressed schema hashes per method/notification, used by interop tests to detect drift.

The --ts-contract-import flag controls the import path written into the emitted TS wrappers — supply the path that resolves to your contract module from the directory the wrappers will be imported from.

Emitted output

Codegen run against the contract above emits one TypeScript file and one Python file per service. Both files are fully typed; the consumer imports <Service>Client, <Service>Service, the per-method param/result types, and (Py only) the runtime <service>_contract: Contract object you pass to register_service(...).

TypeScript output

// AUTO-GENERATED by @clamator/codegen v0.1.4 from arith.ts.
// DO NOT EDIT. Re-run codegen to update.
import { arithContract } from '../contracts/arith.js';
import type { ClamatorClient } from '@clamator/protocol';
import type { z } from 'zod';

export type AddParams = z.infer<typeof arithContract.methods.add.params>;
export type AddResult = z.infer<typeof arithContract.methods.add.result>;
export type PingParams = z.infer<typeof arithContract.methods.ping.params>;

export class ArithClient {
  constructor(private client: ClamatorClient) {}
  add(params: AddParams, opts?: { timeoutMs?: number }): Promise<AddResult> {
    return this.client.call('arith', 'add', params, opts);
  }
  ping(params: PingParams): Promise<void> {
    return this.client.notify('arith', 'ping', params);
  }
}

export interface ArithService {
  add(params: AddParams, opts?: { timeoutMs?: number }): Promise<AddResult>;
  ping(params: PingParams): Promise<void>;
}

(Verbatim from ts/packages/over-redis/tests/generated/arith.ts:1-24.)

The TS file does not export the contract itself — the source contract module already exports arithContract, and the generated client imports it from there. Method names that are camelCased in the contract (e.g., addEvent) are kept camelCased on the TS side. Notifications return Promise<void> and route through client.notify(...) rather than client.call(...).

Python output

# AUTO-GENERATED by @clamator/codegen v0.1.4 from arith.ts.
# DO NOT EDIT. Re-run codegen to update.
from __future__ import annotations
from abc import ABC, abstractmethod
from clamator_protocol import ClamatorClient, Contract, MethodEntry

from pydantic import BaseModel, ConfigDict


class AddParams(BaseModel):
    model_config = ConfigDict(
        extra="forbid",
    )
    a: float
    b: float


class AddResult(BaseModel):
    model_config = ConfigDict(
        extra="forbid",
    )
    sum: float


class PingParams(BaseModel):
    model_config = ConfigDict(
        extra="forbid",
    )


class ArithClient:
    def __init__(self, client: ClamatorClient) -> None:
        self._client = client

    async def add(self, params: AddParams, *, timeout_ms: int | None = None) -> AddResult:
        raw = await self._client.call("arith", "add", params.model_dump(mode='json', by_alias=True), timeout_ms=timeout_ms)
        return AddResult.model_validate(raw)

    async def ping(self, params: PingParams) -> None:
        await self._client.notify("arith", "ping", params.model_dump(mode='json', by_alias=True))


class ArithService(ABC):
    @abstractmethod
    async def add(self, params: AddParams) -> AddResult: ...

    @abstractmethod
    async def ping(self, params: PingParams) -> None: ...


METHODS = {
    "add": MethodEntry(params_model=AddParams, result_model=AddResult, handler_attr="add"),
    "ping": MethodEntry(params_model=PingParams, result_model=None, handler_attr="ping"),
}

arith_contract = Contract(
    service="arith",
    methods=METHODS,
)

(Verbatim from py/packages/over-redis/tests/generated/arith.py:1-60.)

The Py file exports <Method>Params / <Method>Result Pydantic models (with extra="forbid" for strict validation), an <Service>Client proxy class that calls model_dump(by_alias=True) on params before sending and model_validate on the reply before returning, an <Service>Service ABC with one abstract method per contract method, and the runtime <service>_contract: Contract object you pass to server.register_service(...).

Identifier conventions. The contract identifier is <service>_contract — the wire-side service name lowercased and snake-cased, with a _contract suffix. The ABC and Client class names are PascalCase versions of the service name. Method names that are camelCased in the contract (e.g., addEvent) become snake_case on the Py side (e.g., add_event); the handler_attr in each MethodEntry matches the snake_cased Py method name. Subclass the ABC to register a service: class MyService(ArithService): async def add(self, params): ....

Discriminated-union result example

When a method's result is a Zod discriminatedUnion — the canonical "failure as data" pattern (see @clamator/protocol's "Failure as data vs. RpcError" section) — codegen wraps the variants in a pydantic.RootModel and the proxy returns that wrapper. Given a launch.ts contract whose start.result is z.discriminatedUnion('ok', [{ok: true, runId}, {ok: false, reason}]), codegen emits:

class StartResult1(BaseModel):
    model_config = ConfigDict(
        extra="forbid",
    )
    ok: Literal[True]
    run_id: str = Field(..., alias="runId")


class StartResult2(BaseModel):
    model_config = ConfigDict(
        extra="forbid",
    )
    ok: Literal[False]
    reason: Literal["not-launchable", "already-running", "not-found"]


class StartResult(RootModel[StartResult1 | StartResult2]):
    root: StartResult1 | StartResult2


class LaunchClient:
    def __init__(self, client: ClamatorClient) -> None:
        self._client = client

    async def start(self, params: StartParams, *, timeout_ms: int | None = None) -> StartResult:
        raw = await self._client.call("launch", "start", params.model_dump(mode='json', by_alias=True), timeout_ms=timeout_ms)
        return StartResult.model_validate(raw)

(Verbatim from ts/packages/codegen/tests/fixtures/expected/py/launch.py:19-45.)

The variant classes are <Method>Result1, <Method>Result2, ... in source order. The wrapper <Method>Result is a pydantic.RootModel over their union — model_validate(raw) returns it, and result.root.ok discriminates the variant. The handler's return type is the same wrapper; return either variant directly and result_model.model_validate(...) accepts it. Both pyright and mypy enforce exhaustive matching against the union when consumers branch on result.root.ok.

Drift detection via the manifest

--manifest <path> writes a JSON file with content-addressed schema hashes per method and notification. The codegen CLI does not have a --check mode; drift detection is a pattern you run in CI:

1. Keep manifest.json checked into your repo, alongside the committed generated outputs.
2. In CI, regenerate the codegen output to a temporary path:

   npx @clamator/codegen --src contracts --out-ts /tmp/ts --out-py /tmp/py --manifest /tmp/manifest.json

3. Byte-compare against the checked-in copy:

   diff manifest.json /tmp/manifest.json

4. Any mismatch means the contract source diverged from the committed generated artifacts. Fail the CI step.

The interop suite uses this exact pattern (regenerate twice into separate tmp directories and compare manifests byte-for-byte) to verify codegen determinism — see tests/interop/lib/runner.ts:415-425.

This drift detection is for your contract source vs. your committed generated wrappers. clamator framework version drift across the seven published packages is impossible by construction: every package is released in lockstep at the same X.Y.Z, and the release-verification workflow runs the cross-language interop suite on every tag (see @clamator/protocol for the version-compatibility statement). Pin all clamator packages to one X.Y.Z on both sides and framework drift cannot occur.

The diff-against-committed pattern works because codegen output is deterministic: the AUTO-GENERATED header carries the codegen version (no timestamp), and the emitted file content is a function of the contract IR. Identical contracts on the same codegen version produce byte-identical output across runs.

Monorepo integration

A polyglot monorepo with both TS and Py consumers of the same contract benefits from a single canonical location for the contract source and one CI pipeline that emits both languages' wrappers.

Recommended layout.

• Contract source lives in a dedicated TS package (e.g., packages/<name>-contracts/src/<service>.ts). That package depends on zod and @clamator/protocol (both peer-deps if it's library-shaped, both regular deps if it's an internal-only package).
• The contracts package owns a codegen npm script that runs the CLI with --src pointing at its own contract sources, --out-ts pointing at the package's dist/generated/ (or wherever the TS consumers want them), --out-py pointing at the Python consumer's source tree (e.g., packages/<name>-engine/src/<name>_engine/_generated/), and --manifest for the drift-detection workflow.
• Python consumers import the emitted _generated/<service>.py files from their package source as if they were any other Python module. They do not need npm available at runtime — only the committed generated artifacts.
• Per the "Drift detection via the manifest" section above, CI runs the codegen script and diffs the manifest against the committed copy on every change.

Implication for Python-only consumers. Codegen runs in Node, so a Python-only deployment still needs Node available wherever codegen runs (typically a developer machine or CI runner). The committed generated .py files have no Node dependency at runtime — Python installs and consumes them the same way as any vendored library code.

Avoid running codegen in pip install / uv build. The generated outputs are committed artifacts; treat regeneration as a contract-update event, not a per-install step. CI's drift-detection diff catches stale outputs before they merge.

Browser consumers

The source contract file (the one calling defineContract(...)) and the generated TS wrapper both import @clamator/protocol at runtime. @clamator/protocol uses Node-only APIs (node:crypto) and cannot be loaded in a browser bundle.

If a consumer wants to share types with browser-side code — for example, exposing failure-reason enums or result shapes to a UI — those types must live in a file that does not import from @clamator/protocol. The recommended pattern is to keep browser-shareable Zod schemas (enums, shared object schemas) in a separate module (e.g. engine-failure-reasons.ts), and have the contract source file import from there. Browser-facing barrels (index.ts) can then re-export from the schema-only file safely.

Links

• Protocol packages: @clamator/protocol, clamator-protocol
• Transports:
  • @clamator/over-memory
  • @clamator/over-redis
  • clamator-over-memory
  • clamator-over-redis
• Design spec: docs/2026-05-07-clamator-design.md
• Agent rules: AGENTS.md