Python SDK reference

pip install gnosyslabs

from gnosys import GnosysClient

client = GnosysClient(api_key="gn_live_...")
# or
client = GnosysClient()  # picks up env

ds = client.datasets.upload(
    "spaceship_titanic.csv",
    name="spaceship-titanic-train",
    task="classification",
    target_column="Transported",
)
print(ds.dataset_id, ds.n_rows, ds.n_columns)

created = client.runs.create(
    domain="tabular",                       # "tabular" | "strategy"
    strategist={"kind": "agent", ...},      # fixed / hp_sweep / agent / llm
    spec_template={...},                    # required for sweep / agent / llm
    specs=[...],                            # required for fixed
    max_iterations=5,
    tier_reached="paper",                   # exit on first spec clearing the strict promotion gate
    no_progress_window=2,
    concurrency=4,
    llm={"provider": "anthropic", ...},
    llm_critic={"enabled": True},
    enable_code_exec=False,                 # opt-in sandboxed code execution
    sandbox_config={"timeout_s": 600},      # only if enable_code_exec=True
)
print(created.run_id, created.status)

preds = client.runs.predict(run.run_id, "test.csv")
for label, probs in zip(preds.class_labels, preds.probabilities):
    print(label, probs)

html = client.runs.model_card(run.run_id)
with open(f"model_card_{run.run_id[:8]}.html", "wb") as f:
    f.write(html)

client.findings.list(run_id=run.run_id, severity="blocker")
client.findings.list(validator="honest_eval.shuffled_label")
client.findings.list(spec_id="hp-0-C=1")

matches = client.findings.correlations(
    validators=["llm_critic", "honest_eval.shuffled_label"],
    severity="blocker",
    mode="and",
)
for m in matches:
    print(m.spec_id, m.matched_validators)
    for f in m.findings:
        print(f"  {f.validator}: {f.detail[:80]}")

@dataclass(frozen=True, slots=True)  # actually a pydantic model
class Run:
    run_id: str
    tenant_id: str
    domain: str
    strategist: str | None
    status: str          # "queued" | "running" | "completed" | "failed"
    error: str | None
    pipeline_run_id: str | None
    created_at: str
    started_at: str | None
    completed_at: str | None
    is_terminal: bool    # property
    is_completed: bool   # property

class IterationOutcome:
    iteration: int
    n_proposed: int
    n_records: int
    best_tier_in_iter: str | None
    tier_counts: dict[str, int]
    stop_reason: str | None
    duration_seconds: float | None
    strategist_name: str | None

class ValidationFinding:
    finding_id: str
    exec_id: str | None
    pipeline_run_id: str | None
    domain: str
    spec_id: str
    validator: str
    stage: str       # "pre_execute" | "post_execute" | "honest_eval"
    severity: str    # "info" | "low" | "medium" | "high" | "blocker"
    detail: str | None
    payload: dict
    created_at: str
    is_blocker: bool  # property

class CorrelationMatch:
    spec_id: str
    domain: str
    pipeline_run_id: str | None
    matched_validators: list[str]
    findings: list[ValidationFinding]

class ApiKey:
    key_id: str
    name: str
    key_prefix: str   # e.g. "gn_live_a3"
    last_4: str       # e.g. "X9F2"
    scopes: list[str]
    created_at: str
    revoked_at: str | None
    is_active: bool

    def to_display(self) -> str   # "gn_live_a3…X9F2"

from gnosys import RateLimitError

try:
    client.runs.create(...)
except RateLimitError as exc:
    time.sleep(exc.retry_after or 60)

from gnosys import AsyncGnosysClient

async def main():
    async with AsyncGnosysClient(api_key="gn_live_...") as client:
        run = await client.runs.create(...)
        run = await client.runs.wait(run.run_id)
        findings = await client.findings.list(run_id=run.run_id)