Make a function reusable

Add typed inputs and outputs to a Python function, publish it once and call it from the CLI or Python.

A Program wraps a Python function with typed Pydantic input and output models. You publish the function once, then call it from the CLI or Python with new inputs whenever you need.

1. Write a typed Python function

Define a function that takes a Pydantic BaseModel and returns a BaseModel. Here is double from compute/platform/examples/programs/jobs.py:

python

from pathlib import Path
from pydantic import BaseModel, ConfigDict, Field

class Number(BaseModel):    model_config = ConfigDict(extra="forbid", strict=True, frozen=True)    value: int

def double(request: Number) -> Number:    result = Number(value=request.value * 2)    Path("results/value.txt").write_text(str(result.value))    return result

Keep these two rules in mind:

Return values vs. files: Return small structured data (like numbers, flags, or summaries) in the Pydantic model. Write large outputs (like trajectories or structures) to results/ as files.
Reject typos early: Setting extra="forbid" and strict=True on your model catches misspelled parameters before a worker starts running.

2. Publish and run from the CLI

Publish jobs:double as version 1 of example.double, then submit a JSON input:

bash

ct-cli program publish compute/platform/examples/programs \  --name example.double \  --version 1 \  --function jobs:double
printf '{"value": 21}\n' > /tmp/ct-number.jsonct-cli submit example.double@1 --params /tmp/ct-number.json --wait

The worker validates {"value": 21}, runs double, uploads results/value.txt, and returns the structured output:

3. Call the Program from Python

You can look up and submit any published Program from Python using your saved CLI login:

python

from chemistry_compute import Clientfrom jobs import Number, double
client = Client.from_login()program = client.lookup("example.double@1", double)
job = client.submit(program, Number(value=21))answer: Number = job.result(timeout=300)print(answer.value)  # 42

Note: Run this script from compute/platform/examples/programs/ (or with that folder on PYTHONPATH) so import jobs resolves. Client is a regular object, not a with context manager.

To see which Programs are already published in your project, run:

4. Publish new versions safely

Published Program versions are immutable:

Same code, new inputs: Call client.submit(program, Number(value=5))—no republish needed.
Changed code or dependencies: Publish a new version (--version 2 or example.double@2). Existing workflows pinned to example.double@1 continue to work without breaking.

Complete source for this tutorial (including the multi-step functions used in Build a scientific pipeline):

python

"""Tiny reusable functions and a compatible prepared-runtime implementation."""
from __future__ import annotations
from pathlib import Path
from pydantic import BaseModel, ConfigDict, Field
from chemistry_compute import RunnerContextfrom chemistry_compute.flows import Items

class Number(BaseModel):    model_config = ConfigDict(extra="forbid", strict=True, frozen=True)    value: int

class Count(BaseModel):    model_config = ConfigDict(extra="forbid", strict=True, frozen=True)    count: int = Field(ge=0, le=100)

def prepare(request: Count) -> Items[Number]:    return Items[Number](items=[Number(value=index) for index in range(request.count)])

def double(request: Number) -> Number:    result = Number(value=request.value * 2)    Path("results/value.txt").write_text(str(result.value))    return result

def total(request: Items[Number]) -> Number:    return Number(value=sum(item.value for item in request.items))

def increment(request: Number) -> Number:    return Number(value=request.value + 1)

class PreparedRunner:    def setup(self) -> None:        self.ready = True
    def run(self, request: Number, context: RunnerContext) -> Number:        if not self.ready or context.cancelled:            raise RuntimeError("runner is not ready")        return double(request)
    def reset(self) -> None:        pass
    def close(self) -> None:        self.ready = False