Python SDK

The geneflow Python package is the official client. It's a drop-in for mlflow at the surface level (same function shapes), and exposes additional GeneFlow-native modules.

pip install geneflow
# or from this repo:
pip install -e packages/geneflow-python

Setup

import geneflow

geneflow.set_tracking_uri("https://api.genedata.io")
geneflow.set_tenant("ACME")

Or via env vars (auto-picked up):

export GENEFLOW_TRACKING_URI=https://api.genedata.io
export GENEDATA_PAT=$(genedata auth token)   # or MLFLOW_TRACKING_TOKEN
export GENEFLOW_TENANT_ID=ACME

Tracking

with geneflow.start_run(experiment_name="fraud_v3", run_name="run-7") as run:
    geneflow.log_param("max_depth", 7)
    geneflow.log_params({"lr": 0.01, "subsample": 0.8})
    for step in range(1, 6):
        geneflow.log_metric("auc", 0.7 + step * 0.04, step=step)
    geneflow.log_metric("cost_usd", 2.34)        # GeneFlow tracks $ per run
    geneflow.set_tag("env", "training")

Retrieve later:

from geneflow.tracking import get_run_cost, get_metric_history

history = get_metric_history(run.run_id, "auc")
cost = get_run_cost(run.run_id)
print(cost)  # {'cost_usd': 2.34, 'tokens_used': 0, 'duration_ms': 12831}

Artifacts

from geneflow import artifacts

artifacts.upload(run_id, "/tmp/model.pkl", dst_path="model/model.pkl")
files = artifacts.list_artifacts(run_id)
local = artifacts.download(run_id, "model/model.pkl", dst_path="/tmp/dl.pkl")

Model registry

from geneflow import models

models.register("fraud-detector", description="XGBoost fraud model")
v1 = models.create_version(
    "fraud-detector",
    source=f"runs:/{run.run_id}/model",
    run_id=run.run_id,
)
models.transition("fraud-detector", v1["model_version"]["version"], to_stage="Staging")

# Compare two versions: $ + metric deltas
diff = models.compare("fraud-detector", 1, 2)

Model serving (geneflow.serving)

from geneflow import serving

ep = serving.create_endpoint(
    name="fraud-prod",
    model_name="fraud-detector",
    model_version=3,
    instance_type="cpu-large",
    min_replicas=2,
    max_replicas=10,
    drift_check_enabled=True,
)

# Live metrics
m = serving.get_metrics(ep["id"], window_minutes=60)
print(f"QPS={m['qps']:.2f} p95={m['p95LatencyMs']}ms err={m['errorRatePct']:.2f}%")

# Rolling update to v4
serving.update_endpoint(ep["id"], model_version=4, reason="precision win")

# Drift baseline (run once per registered version)
serving.save_drift_baselines(
    model_name="fraud-detector",
    model_version=3,
    features=[{
        "feature_name": "transaction_amount",
        "feature_type": "numeric",
        "histogram": {"buckets": [12, 34, 88, 150, 90, 40, 21, 8, 3, 1], "edges": [0,50,100,150,200,250,300,400,600,1000,5000]},
        "mean": 142.0, "stddev": 87.2, "sample_size": 100_000,
    }],
)

# Run drift check
serving.check_drift(ep["id"], window_minutes=60)

# Tear down
serving.delete_endpoint(ep["id"])

Prompts (geneflow.prompts)

from geneflow import prompts

p = prompts.register(
    name="customer_support",
    content="You are a {{tone}} support agent. {{question}}",
)
p2 = prompts.register("customer_support", content="...")   # bumps to v2
prompts.transition("customer_support", 1, to_stage="Staging", reason="initial deploy")

# Semantic search across the registry
hits = prompts.search_similar("polite responses to angry users", k=3)

# Eval
es = prompts.create_eval_set("support-v1", examples=[...])
prompts.run_eval(es["id"], prompt_version_id=p.id, judge_method="llm_as_judge")

Lineage (geneflow.lineage)

from geneflow import lineage

lineage.upstream(run.run_id, kind="feature_group", id="user_features_v2")
lineage.upstream(run.run_id, kind="dataset", id="s3://bucket/train/2026-05.parquet")

# Downstream: what models came from this feature group?
ds = lineage.get_downstream(kind="feature_group", id="user_features_v2")

Projects (geneflow.projects)

from geneflow import projects

projects.run(
    uri="git@github.com:org/repo.git",
    entry_point="train",
    parameters={"alpha": 0.1},
    backend="kubernetes",          # or "local" / "docker"
    experiment_name="fraud_v3",
)

MLflow import (geneflow.import_mlflow)

from geneflow.import_mlflow import import_from

job = import_from(
    source_uri="https://mlflow.old-company.com",
    experiment_prefix="mlflow:",
    dry_run=True,
)
print(job["counts"])

Or use the CLI: gfctl import-mlflow ....

Common gotchas

• MLFLOW_TRACKING_URI works too — we honor that env var so existing scripts run unchanged.
• X-Tenant-Id precedence — set_tenant() wins, else env, else JWT claim.
• PAT scoping — your token must have geneflow:write for tracking and geneflow:serve for serving operations.
• Artifact uploads — for files >100 MiB, use artifacts.sign_upload() + your own multipart S3 client instead of upload().