The Python client v0.2

mapi-client (mapi_client, Apache-2.0) is a typed client for the /v1 HTTP API. It uses httpx. Each request and each response is a Pydantic model from mapi_spec.api. The server uses the same models. Thus, if a field name has a typo, your editor shows the error before production. Errors come back as mapi_spec.errors.MapiError subclasses with the stable error code of the server.

The mapi CLI, make seed and the end-to-end tests use this client. This page shows one script, docs/snippets/client_tour.py. make e2e runs this script against a live stack. The outputs on this page come from that script. Some outputs are shorter than the real outputs.

uv add mapi-client              # or, for xr.Dataset uploads and xr.DataTree results:
uv add "mapi-client[xarray]"

The workspace already has the two packages. The examples need a live stack and the demo files:

1. Run make dev.
2. Run uv run python scripts/seed.py --export demo/.

Connect

from mapi_client import Client

api = Client()  # MAPI_URL / MAPI_API_URL, a profile, or http://127.0.0.1:8000
health = api.ready()
print(api.config.base_url, health.status, health.version)

http://127.0.0.1:8000 ok 0.15.1

Client() finds its settings in the same sequence as the CLI:

1. The arguments (Client("http://…", token=…, project=…)).
2. MAPI_URL (or MAPI_API_URL), MAPI_TOKEN, MAPI_PROJECT and MAPI_PROFILE.
3. A profile in ~/.config/mapi/config.toml.
4. http://127.0.0.1:8000.

api.config.sources tells where each value came from. Use the client as a context manager (with Client() as api:), or call close().

Upload data

versions = {}
for name in ("eu_media_daily", "fx_daily", "holidays_eu", "eu_media_future"):
    versions[name] = api.datasets.upload(f"demo/{name}.zarr.zip", dataset=name)
    print(versions[name].ref, versions[name].status.value, versions[name].sha256[:19])

dataset://default/eu_media_daily@v2 ready sha256:3062fa82622c
dataset://default/fx_daily@v2 ready sha256:7f27db4a2ac4
dataset://default/holidays_eu@v2 ready sha256:11abf028d5f5
dataset://default/eu_media_future@v2 ready sha256:2db88e1af79d

datasets.upload does all the steps of an upload:

1. If the name is new, it makes the dataset.
2. It opens an upload session.
3. It PUTs the bytes to the signed URL. It does not send your API credentials to that URL.
4. It starts the ingest run in the sandbox with a new idempotency key.
5. It waits until the version is ready. If the ingest fails, it raises the error of the ingest.

It accepts a path, bytes or an xr.Dataset (with the xarray extra). CSV and Parquet files need a mapping=IngestMapping(…). Zarr zips and netCDF files describe themselves. A zip of images needs IngestMapping(format="image_folder", …) (v0.14, see Image data). A folder of images goes up as a collection with datasets.upload_collection (v0.15, see below).

The version has the schema that the ingest run checked:

schema = versions["eu_media_daily"].data_schema  # the DataSchema the ingest run checked
print({dim: spec.policy.value for dim, spec in schema.dims.items()})
print({var: spec.role.value for var, spec in schema.vars.items()})

{'date': 'extend', 'market': 'new_levels', 'channel': 'fixed'}
{'spend_local': 'spend', 'revenue_local': 'target'}

Validate a spec

with open("demo/eu-geo-mmm.yaml", encoding="utf-8") as fh:
    spec_yaml = fh.read()

check = api.specs.validate(spec_yaml)
print("valid" if check.valid else [(i.code, i.pointer) for i in check.issues])
for stage in check.stages:  # the schema after every pipeline step, and where the split is
    if stage.data_schema is not None:
        print(stage.name, sorted(stage.data_schema.dims), "<- split" if stage.split else "")

valid
raw ['channel', 'geo', 'time']
to_eur ['channel', 'geo', 'time']
weekly ['channel', 'geo', 'time'] <- split
hols ['channel', 'geo', 'time']

Validation is synchronous and static. The server never imports plugin code. It returns all the issues at one time, each with a JSON pointer (/pipeline/2/with/over). It also returns the schema at each stage. Here, the schema uses the canonical names of the dataset (geo, time) after binding. The marketing-mix model (mmm.bayesian, on mmm-framework) scales the data and carries spend over. Thus, the pipeline stops at the currency, weekly and holiday steps.

Fit and follow the run

from mapi_client import RunEnd, RunEventFrame, RunLiveFrame

revision = api.specs.upsert(spec_yaml)  # creates the spec, or adds a revision if it changed
run = api.runs.fit(revision.spec_id, revision.revision)

for item in api.runs.stream_events(run.id):  # SSE; resumes with Last-Event-ID
    match item:
        case RunLiveFrame(name="progress", event=event):  # live only, never replayed
            print("  ", event.payload.get("message"))
        case RunEventFrame(name=name, event=event):  # persisted, with an event id
            p = event.payload  # state, warning, diagnostic, metric, complete, error
            print(name, p.get("code") or p.get("name") or "", p.get("message") or p.get("value"))
        case RunEnd(run=final):  # the stream closes after this frame
            print("end:", final.state.value)

run = api.runs.wait(run.id, raise_on_failure=True)
print(run.state.value, run.output_model_version_id)

state  bootstrap object read and deleted
warning DEV_MODE_ISOLATION worker isolation is reduced (development mode): …
state  downloading the dataset
state  fitting the pipeline and the model
   validated the spec and the data
   sampling (pymc)
   posterior ready
   fit complete
   posterior ready
state  rendering plots and views
   explorer: params
diagnostic RHAT 1.0485957785967657
diagnostic ESS_BULK 70.61247760811287
diagnostic ESS_TAIL 51.394768720375154
diagnostic DIVERGENCES 0.0
diagnostic PARETO_K_HIGH 3.0
metric coverage_90 0.9583333333333334
metric elpd_loo 89.0564020582058
metric mape 2.53957229942099
state  uploading outputs
complete  None
state  complete manifest uploaded
end: succeeded
succeeded mv_01m53hmgdhfby1esy1xe341hvh

(We got this output on a laptop against make dev. The demo spec has very small sampler settings on purpose. "posterior ready" shows two times. First, the model fits on the training weeks. Then it does an evaluation on the holdout. Then it fits again on all the weeks.)

runs.stream_events follows the server-sent event stream. It yields four types of items:

Item	When	Notes
RunSnapshot	First, and after each change of state	The full Run
RunEventFrame	Persisted events: state, warning, diagnostic, metric, checkpoint, complete, error	Has an id. The stream sends it again after a reconnect.
RunLiveFrame	progress and log	Live only. The stream never sends it again.
RunTrackingFrame	tracking (v0.11): the metric series, params or tags changed	Live only. A TrackingFrame with the changed series summaries. From v0.14, its field media is True when the run logged new images or tables.
RunEnd	One time, when the run is terminal	Then the stream closes.

If the connection stops, the client continues it with Last-Event-ID. Thus, no persisted event is lost or repeated. runs.wait polls (with backoff) and does not use the stream. With raise_on_failure=True, it raises the error of the run. runs.logs(run_id) gives the captured output and the structured user lines page by page (include_user=False gives only the captured output).

Inspect the model version

model = next(m for m in api.models.list() if m.name == "eu-geo-mmm")
mv = api.models.version(model.id, model.latest_version)
print(mv.ref, mv.scorecard.overall if mv.scorecard else None)  # pass | warn | fail
print(mv.metrics)
for item in mv.scorecard.items if mv.scorecard else []:
    print(f"{item.status:4} {item.code}: {item.message_plain}")

model://default/eu-geo-mmm@v1 fail
{'mape': 2.53957229942099, 'elpd_loo': 89.0564020582058, 'coverage_90': 0.9583333333333334}
warn RHAT: The chains mostly agree, but not fully (R-hat 1.05).
fail ESS: Only 71 independent draws: the numbers are not stable.
pass DIVERGENCES: The sampler explored the model without problems.
pass HOLDOUT_MAPE: The forecast error on the holdout periods was 3%.
pass HOLDOUT_COVERAGE: 96% of held-out values fell inside the 90% range, as expected.
pass NEGATIVE_ROI: No channel is likely to have a negative return.
pass ROI_ABSURD: The highest channel return (4.34 per unit spent) is believable.
pass SHARE_OUTLIER: No single channel explains most of the KPI.
pass BASELINE_SHARE: The baseline (KPI without media) is 84%, a plausible share.
pass PRIOR_OVERLAP: The data moved every parameter away from its prior.
pass DATA_MISSING: Only 0% of the values are missing.
fail DATA_HISTORY: Only 0.6 years of history: too short to separate media from seasonality.
pass SPEND_VARIATION: Every channel's spend varies enough to measure its effect.

The scorecard works correctly here. The demo spec samples 100 draws on 2 chains from about half a year of weekly data. This keeps the tour fast. The scorecard tells this in plain words: too few independent draws and too little history. If a scorecard fails, you cannot promote the model version to an endpoint (PROMOTION_BLOCKED). If a scorecard only warns, an admin must give an audited reason (the promotion gate).

A ModelVersion also has the signed manifest, the diagnostics, the capabilities, the lineage and the list of artifacts. You can download each of them:

metrics = api.models.download(model.id, mv.version, "metrics.json")  # bytes, CRC32C-checked
api.models.download(model.id, mv.version, "bundle.zarr.zip", dest="eu-geo-mmm.zarr.zip")

The API checks the signature of the manifest before it makes the short-lived signed URL. The client compares the download with the CRC32C and the size that the object store calculated. mapi_sdk.bundle.Bundle opens the downloaded bundle.zarr.zip in a notebook.

Score

result = api.models.score(
    model.id,
    mv.version,
    {
        "model": mv.ref,
        "mode": "forecast",
        "horizon": {"dim": "time", "periods": 8},
        "data": "dataset://default/eu_media_future@latest",
        "side_tables": {"fx": "dataset://default/fx_daily@latest"},  # score_policy: supplied
        "output": {"kind": "expected", "summary": ["mean", "hdi_90"]},
    },
)
print(result.run.state.value, [link.name for link in result.outputs])
summary = result.tree["summary"]  # an xr.DataTree (needs the `xarray` extra)
print(result.tree.attrs["output_space"])  # {"y_hat": "model-owned", ...}: weekly EUR
print(summary["y_hat"].sel(stat="mean", market="NL").round(0).values)

succeeded ['plots.json', 'result.json', 'score.zarr.zip', 'summary.json', 'ui/guided/forecast.json', 'ui/summary.arrow']
{"contributions": "model-owned", "y_hat": "model-owned"}
[272645. 254447. 259143. 258366. 265242. 269497. 268112. 266377.]

model-owned means that the model returned its answers in the units of its own stage. The units attribute of each variable records these units. Here, the units are weekly EUR. The pipeline converted and resampled the data before the model.

models.score sends the request with Prefer: wait=60. If the run is not complete at that time, the client sends the same request again with the same idempotency key. The server answers with the run that it made before. Thus, a slow score never becomes two runs. With the xarray extra, the client downloads the result tree, does a check of it and puts it in result.tree.

The request is a ScoreRequest as a dict, a YAML string or the Pydantic model. The client fills in apiVersion and kind.

Errors

from mapi_spec.errors import MapiError, ModeNotSupported

request = {
    "model": mv.ref,
    "mode": "forecast",
    "horizon": {"dim": "time", "periods": 8},
    "data": "dataset://default/eu_media_future@latest",
}
try:  # the fx side table has `score_policy: supplied`, and this request does not supply it
    api.models.score(model.id, mv.version, request)
except MapiError as err:
    print(err.code, err)  # SIDE_TABLE_COVERAGE, with a JSON pointer: /side_tables/fx

try:  # this model cannot filter (that is for state-space models)
    api.models.score(model.id, mv.version, {**request, "mode": "filter"})
except ModeNotSupported as err:
    print(err.code)  # MODE_NOT_SUPPORTED, raised before any run is created

SIDE_TABLE_COVERAGE validation failed with 1 error; first at /side_tables/fx: side table 'fx' has score_policy 'supplied' and must be provided for stage 'raw'
MODE_NOT_SUPPORTED

The client changes each RFC 7807 problem from the server into a MapiError. The error can also be a subclass, for example ModeNotSupported, ValidationFailed, CoordPlanError, QuotaExceeded or IntegrityFailure. It has the stable code, the HTTP status, the JSON pointer and the individual issues. The two requests above failed in the static checks, before the server made a run. Transport failures raise mapi_client.TransportError. If the client cannot parse a response, it raises UnexpectedResponse.

The client tries idempotent calls again on 502, 503, 504 and connection errors. Idempotent calls are GETs, and POSTs with an idempotency key. The client obeys Retry-After (ClientConfig.max_retries, default 2). For 429 answers, see Quotas and rate limits.

Projects, tokens and usage v0.3

On a multi-tenant API, the client needs credentials and a project (Tenancy and authentication). The credentials come from an API token (Client(token=…), MAPI_TOKEN or the profile). If there is no token, they come from the sign-in that mapi login stored for the profile. The client renews that sign-in automatically. It sends the sign-in only to the API that made it.

The project comes from Client(project=…), MAPI_PROJECT or the profile (mapi use). The client sends the project as X-Mapi-Project, only on the requests for a project:

me = api.me()
for tenant in me.tenants:
    for project in tenant.projects:
        print(f"{project.tenant_id}/{project.project_id}: {project.role.value}")
api.use("acme/growth")  # X-Mapi-Project on every project-scoped request from now on

Make a token for a program. The API returns the secret only one time. The secret does not show in repr:

created = api.tokens.create(
    "budget-planner",
    ["score:geo-mmm-prod", "models:read"],
    expires_in_days=30,
    rate_limit_per_minute=120,
)
print(created.token.display, created.token.expires_at.date())
secret = created.secret  # the only copy: put it in your secret store now

Usage reports and runs have exact costs in integer micro-units of the currency:

report = api.usage.get(period="month", group_by="resource_class")
print(report.totals.amount_micros / 1e6, report.totals.currency)
for quota in report.quotas:
    print(quota.limit_name.value, f"{quota.fraction_used:.0%}", quota.resets_at)
run = api.runs.get("run_01jaf3q2x8m9v7k5h4g3f2e1d0")
print(run.cost.amount_micros, run.cost.currency, run.cost.estimated)

api.tenants contains the calls for tenants, projects and members. api.admin contains the operator endpoints: tenants, quotas, kill switches, executors and the audit log. See the reference.

Quotas and rate limits

A 429 answer has one of two codes:

• RATE_LIMITED: The server refused the request before it did work on it. Thus, the client waits and sends the request again, for all methods. It waits for Retry-After, then for an exponential back-off with a small jitter. It raises RateLimited only after RetryPolicy.rate_limit_retries attempts. If the server asks for a wait longer than max_rate_limit_wait, the client raises RateLimited immediately.
• QUOTA_EXCEEDED: The client raises QuotaExceeded immediately. retry_after_of(exc) tells when the quota window starts again. It is None when capacity becomes free only after the work that runs now is complete.

try:
    api.runs.fit("spc_01jaf3q2x8m9v7k5h4g3f2e1d0", 1)
except QuotaExceeded as exc:
    wait = retry_after_of(exc)  # seconds until the quota window resets (None: no reset)
    print(exc.code.value, exc.detail, wait)

Use a RetryPolicy to change this behavior. With max_quota_wait, the client can also wait until a short quota window ends:

patient = Client(
    config=config,
    retry=RetryPolicy(rate_limit_retries=8, max_rate_limit_wait=300, max_quota_wait=0),
)

Endpoints, webhooks and sharing v0.4

The examples in this section come from docs/snippets/client_integrations.py. This script runs offline against a substitute API. For the concepts, see Integrations.

Promote a model version to an endpoint

endpoint = api.endpoints.promote(
    "geo-mmm-prod",
    "model://growth/eu-geo-mmm@v3",
    expected_revision=3,  # 409 CONFLICT if someone else promoted in between
)
print(endpoint.ref, "->", endpoint.model_ref, endpoint.gate.outcome)

endpoint://growth/geo-mmm-prod -> model://growth/eu-geo-mmm@v3 allowed

A promotion goes through the promotion gate. If the scorecard of a version warns, the call raises MapiError with code OVERRIDE_REQUIRED. Then a signed-in project admin must do the call again with override=True, reason="…". Nobody can override PROMOTION_BLOCKED. api.endpoints also has list, get, history, rollback(alias, reason, to_revision=…) and delete.

Score an endpoint into xarray

api.endpoint(alias) returns a RemoteEndpoint:

• schema() tells what to send.
• validate(ds) does a check of a payload, but does not run it.
• score(ds) sends an xr.Dataset inline (as XrJSON, up to 5 MB) and returns the scored xr.Dataset.

geo_mmm = api.endpoint("geo-mmm-prod")
schema = geo_mmm.schema()
print(sorted(schema.snapshot.input_schema.vars))

plan = xr.Dataset(
    {"spend_local": (("market", "channel", "date"), np.array([[[1200.0], [5300.0]]]))},
    coords={"market": ["de"], "channel": ["search", "tv"], "date": ["2025-12-30"]},
)
result = geo_mmm.score(plan, wait=60)  # inline XrJSON in, xr.Dataset out
print(result["revenue_local"].sel(stat="mean").values)

['revenue_local', 'spend_local']
[[13260.]]

score returns the summary group. This group has a stat dim (mean, hdi_90_lower, hdi_90_upper and more). For posterior draws along sample, pass group="draws" with output={"draws": 100}. For a counterfactual, pass group="delta/summary". Each other EndpointScoreCall field is a keyword, for example mode="forecast", horizon={"dim": "time", "periods": 8}, scenario=plan_ds and future_covariates="carry_last".

Like models.score, it sends Prefer: wait=60. It sends the call again with the same idempotency key until the run is complete. timeout= sets the maximum wait. Thus, a slow score never makes two runs. api.endpoints.score(alias, call) is the lower-level call. It returns a ScoreResult with the run, the output links and the downloaded tree. ScoreResult.summary holds the summary as tables. Each table gives dims (its coordinate columns) and time_dim (the dim that is time, or None). Inline payloads need the xarray extra. If the data is larger, upload it as a dataset and send it as data="dataset://growth/plan@v2".

mapi_client.xrjson.encode(ds) changes an xr.Dataset into the XrJSON payload. decode(payload) changes the payload back into an xr.Dataset. XrJSON is the wire format of the SDK. It has dates as ISO strings and NaN as null. The client does a check of the size before it sends data.

Forecast with your latest actuals

api.endpoints.forecast(alias, rows) sends actuals and the dates to forecast in one call. It does not fit the model again. rows is a long table in the input layout of the endpoint (schema().input_table). This layout has the time column, the target and the drivers. rows can be a pandas DataFrame, a list of row mappings or a mapping of columns. Rows that have a target value are actuals. The model forecasts the rows after them that have no target value:

import pandas as pd

rows = pd.DataFrame({
    "t": ["1961-01-01", "1961-02-01", "1961-03-01", "1961-04-01", "1961-05-01"],
    "y": [417, 391, 419, None, None],
})
result = api.endpoints.forecast("ts-forecast", rows)
print(result.conditioning.message)
# Forecast updated with 3 actuals (1961-01-01 to 1961-03-01) after the fitted history, which
# ends 1960-12-01; the model's parameters were not re-estimated. 2 periods forecast from
# 1961-04-01 to 1961-05-01.

You can also give the two parts separately:

• actuals=: the rows with the target.
• future=: the dates to forecast, rows with their driver values, or a number of periods to add after the last actual (future=6).

The client carries empty driver values on future rows forward (future_covariates="carry_last", the default here). If a model cannot use actuals, it refuses them with ActualsNotSupported (422 ACTUALS_NOT_SUPPORTED). api.endpoint(alias).forecast(rows) does the same and gets the schema one time. mapi_client.forecast_call(schema, rows, ...) only makes the call. ScoreResult.conditioning is the record for each forecast. If you sent no actual, it has conditioned=False and the end of the fitted history.

Snippets

python = api.endpoints.snippets("geo-mmm-prod", ["python"]).items[0]
print(python.code.splitlines()[2])

api.endpoints.snippets(alias, languages) returns the code that the Integrations hub shows. The languages are Python, TypeScript, curl, R, Apps Script and a notebook cell. The Python snippet is the same as the api.endpoints.score(...) call above. The notebook cell uses the api.endpoint(alias).score(...) form. The two snippets run without changes. The tests of the client run them.

Webhooks

created = api.webhooks.create(
    "https://hooks.example.com/mapi", ["model.version.ready", "endpoint.promoted"]
)
signing_secret = created.secret  # the only copy: store it with the receiver
print(created.webhook.id, created.webhook.secret_hint)

create and rotate_secret return the signing secret only one time. The secret does not show in repr. api.webhooks also has list, get, update, delete, deliveries(id, status=…) and replay(id, delivery_id). For the checks of deliveries on the receiver side, see Receiving webhooks.

Render a plot

    render = api.plots.render(
        model_id="mdl_01jaf3c1d2e3f4g5h6j7k8m9n0",
        version=3,
        plot_id="mmm.response_curves",
        format="png",
        width=1200,
    )
    path = api.plots.download_render(render, Path(tmp) / "response_curves.png")

plots.render accepts a PlotSpec, a saved plot (saved_plot_id=) or a plot of a model version. Like a score, it waits for the render in the sandbox. download_render gets the file and does a CRC32C check. plots.save, saved, get_saved and delete_saved manage saved plots. create_embed returns a read-only embed link one time. Also see embeds, revoke_embed and the public embed(token) view. See Sharing plots.

Sign-up, verification, deletion and export

    export = api.tenants.export("acme")
    archive = api.tenants.download_export("acme", export.id, Path(tmp) / "acme-export.zip")

download_export waits until the export is ready. Then it downloads the export through the storage client to a temporary file. It never uses your API credentials for this download. When the size is correct, it renames the file to its final name.

api.tenants also has verification, request_deletion(tenant_id, confirm=tenant_id), deletion and cancel_deletion. api.signup(...) and api.verify_email(token) are the public sign-up calls. The client sends them without credentials. Operators have api.admin.signups, review_signup, set_verification and retry_deletion. See Deleting and exporting data.

Guided mode

api.guided controls Guided mode from code. It has these calls:

• templates() and template(id).
• detect_roles(template_id, columns=[…] | dataset="dataset://…").
• draft(template_id=…, name=…, dataset=…, answers={…}). This returns the ingest mapping, the ModelSpec and its YAML. It does not store data.
• compile({...}). This returns the ScoreRequest for a what-if, a forecast or a preset.
• preview_priors(business_priors, spec_id=…, revision=…). This starts a translate_priors run in the sandbox and waits for it.

Workflows, governance and explorers v0.5

All the v0.5 calls are on the same client. The mapi_spec DTOs give their types. The documentation tests run the tour below, without changes, against a real server.

Help me choose

from mapi_client import Client

api = Client()  # MAPI_URL, the `mapi login` sign-in (or MAPI_TOKEN) and MAPI_PROJECT

result = api.catalog.recommend(
    goal="forecast", data_shape="one_series", drivers="yes", experience="beginner", limit=3
)
for item in result.items:  # best first, with the reasons in plain words
    print(f"{item.score:.0%}  {item.title}  ({item.kind.value} {item.id})")
    for reason in item.reasons_plain:
        print("     +", reason)
    for caution in item.cautions_plain:
        print("     !", caution)

card = api.catalog.about("ts.sarimax", kind="model")  # any item's About card
print(card.summary_plain)
print("use it when:", "; ".join(card.use_when))
print("pick something else when:", "; ".join(card.avoid_when))

Run a workflow and follow it live

from pathlib import Path

check = api.workflows.validate(Path("weekly-readiness.yaml"))  # on the server; saves nothing
for issue in check.issues:  # every problem with a JSON pointer into the document
    print(issue.severity.value, issue.pointer, issue.message)
print(" -> ".join(check.order))  # the order the nodes will run in

from mapi_client import WorkflowEnd, WorkflowEventFrame, WorkflowSnapshot

# create the workflow, or save the file as its next revision when it changed
workflow, revision = api.workflows.upsert(Path("weekly-readiness.yaml"))
run = api.workflows.run(
    workflow.id,
    inputs={"data": "dataset://growth/weekly_sales@latest"},  # pinned to a version at start
    params={"drivers": ["price"], "window": 80, "max_p": 2, "max_q": 2},
)
for item in api.workflows.stream_events(run.run_id):  # server-sent events, resumable
    match item:
        case WorkflowSnapshot(run=now):
            print(f"{len(now.nodes)} nodes, {now.status.value}")
        case WorkflowEventFrame(event=e) if e.type.value == "workflow.branch.decided":
            print(f"  {e.node_id}: took the {e.data['arm']} arm")
        case WorkflowEventFrame(event=e) if e.node_id is not None:
            print(f"  {e.node_id}: {e.type.value.rsplit('.', 1)[-1]}")
        case WorkflowEnd(status=status):
            print("run", status.value if status else "ended")

stream_events yields these items:

• WorkflowSnapshot: the run with all its nodes. It comes first on each connection.
• WorkflowEventFrame: a node is queued, started, succeeded, cached, skipped, failed or retrying, or a branch is decided.
• WorkflowEnd.

If the connection stops, the client continues it with Last-Event-ID. api.workflows.wait(run_id) polls and does not use the stream. It raises the error of the node that failed.

run = api.workflows.wait(run.run_id)  # raises the run's error if it failed
nodes = {n.node_id: n for n in run.nodes}
print(nodes["adf"].result["verdict_plain"])  # e.g. "The series wanders: consider differencing it."
for d in run.decisions:
    print(d.node_id, d.arm, d.inputs)  # every branch with the values its condition read
order, d = nodes["order"].result, 1 if nodes["wanders"].result["value"] else 0
print(f"ARIMA({order['best_p']}, {d}, {order['best_q']})")  # chosen by information criterion
print(nodes["dm"].result["verdict_plain"])  # is ARIMAX's edge over naive more than luck?
print(nodes["select"].result["reason_plain"])  # kept only if it beats the benchmarks
model_ref = run.outputs["model"].ref  # model://growth/wf-weekly-readiness-fit@v1
report = run.outputs["report"].inline  # mapi.report/v1: every test, decision and table

Since v0.9, a node can give a plot output (plot.render, model.forecast and evaluate.residuals). api.workflows.node(run_id, node_id).outputs["plot"].inline is the PlotSpec document of the plot. The server puts it through the plot sanitizer each time it sends it. If the plot is not safe, inline is None. The gallery also has recipes for Bayesian and science models, for example sci_spatial_ladder and bayes_workflow_audit. The workflow cookbook shows how to run each one with the client.

Governance

gov = nodes["governance"].result  # the policy in force, measured on the selected model
print("passed" if gov["passed"] else f"failed: {gov['failed_checks']}")
validation = api.validations.get(gov["validation_id"])  # recorded because submit: true
print(validation.status.value)  # validated, or awaiting_approval when a reviewer must sign
for r in validation.results:
    print(f"{r.check:28} {r.status.value:7} {r.message_plain}")
api.validations.download_evidence_pack(validation.id, "evidence.html")  # for the auditor

import yaml

with open("forecast-policy.yaml", encoding="utf-8") as fh:
    policy = api.governance.create_policy(yaml.safe_load(fh))
print(policy.id, policy.version, [c.check for c in policy.checks])

# edit and save the next version (refused with 409 if someone saved one meanwhile)
stricter = api.governance.update_policy(
    policy.id, policy.model_copy(update={"approvals_required": 2}), expected_version=1
)
for p in api.governance.policies():  # in force in the project: tenant-wide first
    print(p.scope, p.name, f"v{p.version}")

validation = api.models.validate(model_id, version, note="Q4 refresh", wait=True)
print(validation.status.value)  # validated, awaiting_approval or checks_failed
for r in validation.results:
    print(r.check, r.status.value, r.message_plain, r.suggested_action or "")
if validation.status.value == "checks_failed":
    validation = api.validations.resubmit(validation.id, note="refitted on clean data")

# a reviewer, signed in with `mapi login` (never the submitter, never an API token)
reviewer = Client(project=os.environ.get("MAPI_PROJECT"))
for pending in reviewer.validations.list(status="awaiting_approval"):
    evidence = list(reviewer.validations.evidence(pending.id))
    if all(r.status.value in ("pass", "skipped") for r in evidence):
        reviewer.validations.approve(pending.id, comment="Evidence reviewed")
    else:
        reviewer.validations.reject(pending.id, "The null test is not convincing yet")

Explorers and budget optimisation

ds = api.datasets.find("weekly_sales")
assert ds is not None and ds.latest_version is not None
api.datasets.describe(ds.id, "Weekly sales and price by region, from the finance warehouse")
print(api.datasets.about(ds.id).summary_generated)  # "Weekly data from ... for 5 regions ..."

explored = api.datasets.explore(  # computed in the sandbox; the same request is cached
    ds.id,
    ds.latest_version,
    sections=["overview", "stationarity", "ccf", "stl", "quality"],
    kpi="sales",
    max_lag=12,
)
for section in explored.index.sections if explored.index else []:
    print(section.section, section.status, section.message_plain or "")
stationarity = api.explorations.view(explored, "stationarity")  # the view, as JSON

model_id, version = "mdl_01jaf3c1d2e3f4g5h6j7k8m9n0", 3
explored = api.models.explore(
    model_id, version, sections=["params", "residuals", "forecast"], horizon=13
)
residuals = api.explorations.view(explored, "residuals")

compared = api.models.compare(
    ["model://growth/arimax@v3", "model://growth/ets@v1", "model://growth/theta@v2"], horizon=4
)
table = api.explorations.view(compared, "comparison")  # metrics, ranking, Diebold-Mariano

plan = api.models.optimize(  # mmm-framework's planner on the posterior; waits for the plan
    model_id,
    version,
    budget_change_pct=0,  # reallocate today's total
    channels={"tv": {"min_multiplier": 0.5, "max_multiplier": 1.5}},
    objective="p10",  # plan for a cautious outcome, not the average one
    frontier={"n_points": 8},
)
assert plan.result is not None
for a in plan.result.allocation:
    print(f"{a.channel:10} {a.current_spend:>12,.0f} -> {a.optimal_spend:>12,.0f}")
uplift = plan.result.uplift
print("uplift", uplift.mean if uplift else None, "P(better)", plan.result.prob_positive_uplift)

Schedules and triggers

weekly = api.workflows.schedule(
    workflow.id,
    "0 6 * * 1",  # Mondays at 06:00
    timezone="Europe/Berlin",
    inputs={"data": "dataset://growth/weekly_sales@latest"},
    params={"drivers": ["price"]},
)
on_upload = api.workflows.trigger(
    workflow.id, "dataset://growth/weekly_sales", "data", debounce_seconds=600
)
print(weekly.next_run_at, on_upload.id)

Track metrics, params and tags v0.11

You can add metrics, params and tags to a run, also after the run finished. For example, log a business KPI that you measured after you deployed the model. The writes need the runner role. An API token needs the runs:write scope.

run_id = "run_01jaf3q2x8m9v7k5h4g3f2e1d0"

api.runs.log_metric(run_id, "revenue_lift", 0.042, labels={"region": "north"})
api.runs.log_metrics(run_id, {"mae": 3.1, "rmse": 4.2}, labels={"split": "post_launch"})
api.runs.log_params(run_id, {"analyst": "kim", "holdout_weeks": 8})
api.runs.set_tags(run_id, {"stage": "production", "old_tag": None})  # None deletes the tag
api.runs.update(run_id, name="weekly sales, adstock 4", note="Holiday weeks removed.")

with api.runs.track(run_id) as t:
    for week, value in weekly_kpi:
        t.log_metric("weekly_lift", value, step=week)

Call	What it does
runs.update(run_id, name=…, note=…, tags=…, experiment_id=…)	PATCH /v1/runs/{run_id}. Only the arguments that you give change. None clears the name, the note or (v0.14) the experiment.
runs.log_metric(run_id, key, value, step=None, labels=None, ts=None)	Log one point
runs.log_metrics(run_id, values, step=None, labels=None)	Log a dict of values, or a list of MetricPointIn. The client sends a maximum of 1000 points in each request.
runs.log_params(run_id, params)	Log params (500 in each request). A different value for a stored key raises a CONFLICT error.
runs.set_tags(run_id, tags)	Set tags. None deletes a tag.
runs.track(run_id, flush_every=500)	A context manager with log_metric, log_metrics, log_param, log_params, set_tag and set_tags. It sends the points in batches and when the block ends.
runs.metrics(run_id, key=None, attempt=None)	The series of the run with their summaries (MetricSeries)
runs.metric_history(run_id, key, max_points=1000, attempt=None, labels=None)	The points of one key (MetricHistory), downsampled
runs.params(run_id, prefix=None)	The params (RunParam)

runs.list can search by the tracking data. The client does a check of each filters, order_by and columns value before it sends the request:

for run in api.runs.list(
    filters=["metric.rmse < 12", "param.fit.engine = nutpie"],
    tags=["stage:candidate"],
    order_by="metric.rmse",
    columns=["metric.rmse", "param.settings.adstock.l_max"],
):
    print(run.id, run.name, run.columns)

The other filters of runs.list are kind, state, spec_id, parent_run_id, name and (v0.14) experiment_id. For the grammar, see Search runs. For the logging API inside a model, see Track metrics, params and tags.

Media, experiments and comparison v0.14

# images and tables of a run
api.runs.log_image(run_id, "holdout_map", "map.png", caption="Holdout error by region")
api.runs.log_table(run_id, "lift", {"region": ["north", "south"], "lift": [0.04, 0.02]})
for item in api.runs.media(run_id, key="samples"):
    png = api.runs.download_media(run_id, item)   # checks the size, CRC32C and SHA-256

# experiments
exp = api.experiments.create("adstock length study", description="Adstock 2, 4 and 8.")
api.experiments.add_runs(exp.id, [run_a, run_b])
run = api.runs.fit(spec_id, revision, experiment_id=exp.id)
api.runs.update(run_c, experiment_id=exp.id)        # move one run (None: no experiment)

# compare and save a search
cmp = api.runs.compare([run_a, run_b], keys=["loss"], media_keys=["samples"])
view = api.run_views.create(
    "best fits", {"filters": ["metric.rmse < 12"], "order_by": "metric.rmse"},
    experiment_id=exp.id,
)
runs = api.runs.list(experiment_id=exp.id, columns=["metric.rmse"])

Call	What it does
runs.media(run_id, key=None, kind=None, attempt=None, links=True)	The images and tables of a run (RunMedia). With links, each item has a signed url.
runs.media_keys(run_id, attempt=None)	The media series with their counts and steps (MediaKeySummary)
runs.get_media(run_id, media_id)	One item with a new signed url
runs.download_media(run_id, media)	The bytes. The client does a check of the size, the CRC32C and the SHA-256.
runs.log_image(run_id, key, image, step=None, labels=None, caption=None)	Log PNG or JPEG bytes or a file path. With mapi-sdk, also a numpy array, a Pillow image or a DataArray. A maximum of 4 MiB.
runs.log_table(run_id, key, table, step=None, labels=None, caption=None)	Log a DataFrame, a dict of columns or a list of records
runs.track(run_id)	Also has log_image and log_table (sent immediately)
runs.keys(experiment_id=None, prefix=None)	The metric series, param keys and tag keys of the runs (RunKeys)
runs.compare(run_ids, keys=(), media_keys=(), max_points=300, params=True)	Compare 2 to 10 runs (RunCompare)
runs.create(run, experiment_id=None), runs.fit(spec_id, revision, experiment_id=None)	Start a run in an experiment
experiments.list(name=None, archived=None), get, find(name_or_id)	Find experiments
experiments.create(name, description=None), update(id, name=…, description=…, archived=…), delete(id)	Manage experiments
experiments.add_runs(id, run_ids), remove_runs(id, run_ids)	Move runs (a maximum of 200 in each request)
run_views.list(experiment_id=None), find, create(name, query, experiment_id=None), update(id, name=…, query=…), delete(id)	Saved views of the runs table

The async client has the same calls. For the web app and the CLI, see Compare runs and use experiments.

Image collections and weights v0.15

from pathlib import Path

from mapi_spec import CollectionOptions, IngestMapping

# how many images are new? (the client calculates the SHA-256 of each image)
plan = api.datasets.missing_images(dataset_id, ["9f86d0…", "60303a…"])  # a ds_… id
print(plan.present, len(plan.missing))

# upload a folder: only the missing images, in parts, with a state file to continue later
version = api.datasets.upload_collection(
    "product_photos",
    Path("photos"),
    label_from_folder=True,
    options=CollectionOptions(size=(64, 64)),
    state_file=Path("photos.upload.json"),
    wait=True,
)
print(version.layout, version.collection.n_images, version.collection.new_images)

# pretrained weights as a dataset
weights = api.datasets.upload(
    "resnet.safetensors", dataset="tiny-resnet", mapping=IngestMapping(format="safetensors")
)

Call	What it does
datasets.upload_collection(dataset, root, *, manifest=None, label_from_folder=False, options=None, base_version=None, part_bytes=DEFAULT_PART_BYTES, state_file=None, message=None, wait=False, timeout=None, progress=None)	Upload a folder of images as a collection version (DatasetVersion). dataset is a name (the client makes the dataset when it does not exist) or a ds_… id. It uploads only the images that the base version does not have. With state_file, a second call continues the upload: it skips the parts that arrived and renews expired URLs.
datasets.missing_images(dataset_id, shas, *, base_version=None)	BlobMissing. The client sends the digests in groups of 5000 and merges the answers.
datasets.get_upload(dataset_id, upload_id)	An UploadSession with received
datasets.renew_upload(dataset_id, upload_id)	A fresh signed PUT for an upload whose file did not arrive
datasets.upload_weights(source, *, dataset=None, dataset_id=None, arch=None, message=None, wait=True, timeout=None)	Upload a .safetensors file of network weights as a weights dataset. arch is a dict or a JSON file, for a file without mapi_arch metadata. See Deep learning.

mapi_client.collections has the parts of upload_collection that do not need a server: scan_images(root, manifest, path_column=..., label_from_folder=...), write_parts(plan, missing, out_dir, part_bytes) and client_manifest_csv(plan). The parts are zip files with no compression. The same images always give the same bytes.

Async

AsyncClient has the same resources and arguments. Use await for the calls. Use async for for the lists:

import asyncio

from mapi_client import AsyncClient


async def recent_fits(n: int = 3) -> list[str]:
    async with AsyncClient() as aapi:  # same resources and arguments, awaited
        found: list[str] = []
        async for run in aapi.runs.list(kind="fit", limit=n):  # follows next_cursor lazily
            found.append(f"{run.id} {run.state.value}")
            if len(found) == n:
                break
        return found


print(asyncio.run(recent_fits()))

The v0.5 resources work in the same way:

import asyncio

from mapi_client import AsyncClient


async def main() -> None:
    async with AsyncClient() as aapi:
        wf = await aapi.workflows.find("weekly-readiness")
        assert wf is not None
        run = await aapi.workflows.run(
            wf.id, inputs={"data": "dataset://growth/weekly_sales@latest"}, wait=True
        )
        async for node in aapi.workflows.nodes(run.run_id):
            print(node.node_id, node.status.value)


asyncio.run(main())

Reference

Resource	Calls
projects	list, get
catalog	get, models, model, steps, plots, schema
datasets	create, list, get, find, ensure, create_upload, create_version, versions, version, wait_version, upload, artifact, download; v0.8: upload_files, inspect, inspect_file; v0.15: upload_collection, missing_images, get_upload, renew_upload, upload_weights
specs	validate, create, list, get, find, revise, upsert, revisions, revision, dry_run
runs	create, fit, get, list, wait, stream_events, events, logs, plots, cancel, resume, retry; v0.11: update, log_metric, log_metrics, log_params, set_tags, track, metrics, metric_history, params; v0.14: media, media_keys, get_media, download_media, log_image, log_table, keys, compare
experiments (v0.14)	list, get, find, create, update, delete, add_runs, remove_runs
run_views (v0.14)	list, find, create, update, delete
models	list, get, versions, version, artifact, download, score
plots	list, get, validate; v0.4: save, saved, get_saved, delete_saved, render, render_status, render_output, download_render, create_embed, embeds, revoke_embed, embed
guided (v0.4)	templates, template, detect_roles, draft, compile, translate_priors, prior_translation, wait_prior_translation, preview_priors
endpoints (v0.4)	list, get, promote, delete, history, rollback, schema, validate, submit, score, snippets; api.endpoint(alias): info, schema, history, snippets, validate, score, score_result
webhooks (v0.4)	list, create, get, update, delete, rotate_secret, deliveries, replay
tenants (v0.3)	list, get, rename, projects, create_project, update_project, delete_project, members, add_member, set_role, remove_member, usage, audit; v0.4: verification, request_deletion, deletion, cancel_deletion, export, export_status, wait_export, download_export
tokens (v0.3)	list, create, get, revoke
usage (v0.3)	get (the project), tenant
admin (v0.3, operators)	tenants, tenant, create_tenant, update_tenant, quotas, set_quota, delete_quota, kill_switches, set_kill_switches, executors, executor, audit; v0.4: signups, review_signup, set_verification, retry_deletion

List calls return lazy paginators that follow next_cursor. api.health() and api.ready() call /healthz and /readyz. api.me() calls /v1/me. api.use(project) changes the selected project. api.signup(), api.verify_email() and api.endpoint(alias) are v0.4. mapi_client.webhooks verifies deliveries. mapi_client.xrjson encodes inline payloads.

The platform optimizes budgets (see Optimise a budget). For the DTOs, see mapi_spec. For the endpoints, see the HTTP API reference.