Runs and executors

fit_spec and score_bundle are plain functions. The platform does the same work as runs. A run is a JSON-only submission that goes to an executor. The executor sends progress as events. It also returns a status, and the status is the source of truth. The in-process executor uses that protocol inside your Python process. Thus notebooks, tests and the local-process and docker executors of the platform use one code path.

sequenceDiagram
    participant You as Notebook / API
    participant Ex as Executor
    participant H as Harness (fit_spec / score_bundle)
    You->>Ex: submit(RunSubmit{kind: fit, payload: spec}, nonce)
    Ex->>H: resolve dataset refs, fit
    H-->>Ex: progress events
    Ex-->>You: event stream (state, progress, complete)
    You->>Ex: wait(handle) → SUCCEEDED
    You->>Ex: submit(RunSubmit{kind: score, payload: request + bundle})
    Ex->>H: score_bundle
    Ex-->>You: result: xr.DataTree

Set up

The executor uses a resolver to find the datasets for the dataset:// references in specs and requests. Here, the resolver is a mapping to in-memory datasets. The built-in tabular plugin contains the store-demand data and the spec that this page uses.

import asyncio

from mapi_plugin_tabular.datasets import (
    STORE_DEMAND_SCHEMA,
    STORE_DEMAND_SPEC_YAML,
    store_demand_dataset,
)
from mapi_sdk import InprocExecutor
from mapi_spec.executor import ExecState, RunEventType, RunKind, RunSubmit

history = store_demand_dataset(stores=4, skus=2, weeks=60, seed=1)
recent = history.isel(week=slice(-8, None))  # a window of fitted weeks to re-score

executor = InprocExecutor(
    # dataset refs in specs and requests resolve to in-memory datasets
    resolver={
        "dataset://retail/store_sku_weekly@latest": history,
        "dataset://retail/last_8_weeks@v1": recent,
    },
)
print(executor.caps().isolation.name)  # NONE: your own code, in your own process

Submit, wait, score

async def main() -> None:
    fit = await executor.submit(
        RunSubmit(
            run_id="fit-001",
            kind=RunKind.FIT,
            payload={
                "spec": STORE_DEMAND_SPEC_YAML,
                "raw_schema": STORE_DEMAND_SCHEMA.model_dump(mode="json"),
            },
        ),
        nonce="nonce-fit-0001",  # resubmitting the same nonce returns the same run
    )
    status = await executor.wait(fit, timeout=300)
    print(status.state)  # succeeded
    for event in executor.events(fit):  # the same event stream a cloud run will produce
        if event.type is RunEventType.PROGRESS:
            print(event.seq, event.payload.get("message"))  # 2 validated the spec and the data

    executor.put_bundle("store-demand@1", executor.result(fit))
    score = await executor.submit(
        RunSubmit(
            run_id="score-001",
            kind=RunKind.SCORE,
            payload={
                "bundle": "store-demand@1",
                "request": {
                    "apiVersion": "mapi/v1",
                    "kind": "ScoreRequest",
                    "model": "endpoint://retail/store-demand-xgb",
                    "mode": "predict",
                    "data": "dataset://retail/last_8_weeks@v1",
                },
            },
        ),
        nonce="nonce-score-0001",
    )
    assert (await executor.wait(score, timeout=300)).state is ExecState.SUCCEEDED
    tree = executor.result(score)  # an xr.DataTree, exactly what score_bundle returns
    print(dict(tree["summary"].to_dataset()["y_hat"].sizes))
    # {'stat': 3, 'store_id': 4, 'sku': 2, 'week': 8}


asyncio.run(main())  # in Jupyter: `await main()`

succeeded
2 validated the spec and the data
3 split the data
4 fitted quantile model 1
5 persisting the fitted estimators
6 fitted on the training data
7 evaluated the holdout through the score path
8 fitted quantile model 1
9 persisting the fitted estimators
10 fit complete
{'stat': 3, 'store_id': 4, 'sku': 2, 'week': 8}

Look at these points:

• Payloads are JSON. The spec goes as YAML text or as a dict. The raw schema goes as a dict. Data goes only as references. This is exactly what the API will store and what a sandboxed runner will receive.
• Submissions are idempotent. If you submit again with the same nonce, you get the same handle. Thus an API call that you try again never starts a second fit. adopt(nonce) finds a run after a restart.
• Status is the source of truth. Events are for progress bars. status() and wait() tell you if a run succeeded. Failures come back as RFC 7807 problems with a stable code.
• Cancellation is cooperative. The executor checks cancel(handle) between the phases of the pipeline and inside long fits.

Isolation levels

Each executor declares how strongly it isolates a run. The platform does not send untrusted code to an executor that is not sufficiently strong.

Executor	Isolation	For	Status
InprocExecutor	NONE	Notebooks, tests and your own trusted code	available
Local process / Docker	PROCESS / GVISOR	Local platform, CI	v0.2
Cloud Run Jobs (gen2)	MICROVM	Hosted runs, one new sandbox for each run	v0.3

Never run untrusted code in-process

The in-process executor runs plugins in your interpreter, with your permissions. Use it only for code that you wrote or that you trust. Public plugins run only in the sandboxed executors.

For the sandbox design, see the roadmap.