gertlabs 3.0.0

Python SDK for the Gert Labs AI evaluation platform

Gert Labs Python SDK

A thin, synchronous Python wrapper over the Gert Labs REST API, for AI researchers. Connect your own model or a public LLM endpoint to our environments:

• Play locally. Observations come to you, you run inference in your own environment, and you submit actions.
• Connect a model. Register an OpenAI-compatible endpoint once on the platform and let the platform drive it server-side for large-scale code generation, evaluation, and dataset builds.

Either way, collect training data at scale (code-submission evaluations, counterfactual branch data, session replays) and load it into pandas.

Install

pip install gertlabs           # core
pip install 'gertlabs[data]'   # + pandas/pyarrow for exports.load()

Authentication

Create an API key in the dashboard (sk_gert_...) and set it in your environment:

export GERT_API_KEY=sk_gert_...

from gertlabs import GertClient
client = GertClient()                       # reads GERT_API_KEY
# or: GertClient(api_key="sk_gert_...", base_url="http://localhost:8080/api/v1")

Every method maps to an API endpoint and returns plain dicts. Errors raise a GertError subclass (AuthenticationError, ValidationError, InsufficientCreditsError, RateLimitError, ...). Async work returns a job_id; block on it with client.jobs.wait(job_id).

Discover games

List the available environments and their tags. The slugs here (for example market_simulator) are what you pass as game= in the examples below.

for g in client.games.list():
    print(g["slug"], g["tags"])

client.games.list(tag="strategy")    # filter to one category
client.games.tags(active=True)       # tag taxonomy + per-tag game counts

You don't have to name a game. play.create, dataset_builds.*, and exports.create all accept tags instead, and the platform matches an environment for you:

s = client.play.create(tags=["strategy"], seats=1, vs_ai=True, max_ticks=600)

Play locally with your own model

Control one or more seats yourself: the platform sends observations, you run your model in your own environment, and you submit actions. No provider registration or org required. Fill the other seats with the platform's AI (vs_ai=True), or control every seat yourself for self-play (seats=player_count).

from gertlabs import GertClient
client = GertClient()

s = client.play.create(game="market_simulator", seats=1, vs_ai=True, max_ticks=600)
prompt = s["prompt"]            # game rules + observation/action schema

def decide(observation):
    # your local model/policy
    return []                   # see prompt for valid actions

with client.play.connect(s["session_id"], s["player_token"]) as ws:
    for msg in ws:
        if msg["type"] == "observation":
            ws.send_action(decide(msg["data"]))
        elif msg["type"] == "game_completed":
            print(msg["scores"]); break

To instead send REST over WebSockets, poll client.play.get(session_id, player_token) for the latest observation and submit with client.play.act(...), using the same loop.

RL rollouts (pure environment, for training)

For RL training, drive the environment from your own loop: Gert is the game + the scoreboard, and your trainer calls your own model. Open a run once (a billing lease amortized over many episodes), then reset -> step -> done. By default your policy controls every seat (full self-play; a 1-player game is simply the environment), so any game is trainable with no opponent corpus (score-actions and complete are the exception: their playouts are played by rated code submissions, so a brand-new game needs some before those two scorers work); reward is each policy seat's dense per-tick score delta. Frozen league opposition is opt-in (opponent_seats=), with the opponent skill band (min_percentile/max_percentile) as its curriculum knob; for solo/coop games that declare it, difficulty= ([0,1], explicit game + no opponent seats) is the environmental curriculum axis.

from gertlabs import GertClient, rollouts
client = GertClient()

rules = client.games.get("pot_limit_omaha_hilo").get("prompt")
run = client.rollout_runs.open(max_episodes=1000)  # billable reservation - fetch rules first

def act(observation, seat):
    # The platform decision player's (system, user) messages - the user message
    # byte-identical to platform evals (see "Train/eval prompt parity" below). Call YOUR
    # model here, parse its output into an action:
    reply = my_model(system=rollouts.system_prompt(rules),
                     user=rollouts.format_observation(observation))
    return rollouts.parse_action(reply)

try:
    result = rollouts.run_episode(
        client, run["run_id"], game="pot_limit_omaha_hilo",
        act_fn=act, opponent_seats=1,                     # 1 league seat; omit for full self-play
        min_percentile=0.4, max_percentile=0.6,           # hard-but-winnable band
    )
    print(result["total_reward"], result["outcome"])     # scores/rewards are higher-is-better platform-wide -- train on result["steps"] (1-player games: outcome is None; score is the signal)
finally:
    client.rollout_runs.close(run["run_id"])              # settles billing at actual ticks - even if your model code failed

Upgrading to 3.0.0

Three breaking changes from 2.x, all in the RL rollout surface:

• rollouts.format_observation(observation) no longer takes rules=. It now returns exactly the platform decision player's user message; the rules belong in the system message via the new rollouts.system_prompt(rules). Send both (see "Train/eval prompt parity").
• Acting callbacks are seat-labeled: run_episode's act_fn(observation, seat), and score_actions_on_policy's act_fn(observation, seat) / act_batch_fn(observations, seats). Ignore the second parameter (lambda obs, _seat: ...) to keep 2.x behavior.
• The verifiers adapter defaults to history="none" (turn-isolated context) instead of accumulating the conversation. Pass history="full" for the old behavior. Games declaring turn_isolated reject it.

Low-level access is client.rollout_runs.reset/step/close if you want to manage the loop yourself; client.rollout_runs.list(status="active") recovers a lost run_id (e.g. after a crashed training driver) so you can close the run. For Prime Intellect verifiers users, pip install 'gertlabs[verifiers]' and:

from gertlabs.verifiers_env import load_environment
env = load_environment(game="pot_limit_omaha_hilo", max_episodes=5000)
# hand `env` to prime-rl / your verifiers-compatible trainer
env.close()   # settle the run's billing when done (or use `with load_environment(...) as env:`)

By default the verifiers adapter is turn-isolated (history="none"): every model call sees the game rules plus the current observation only. That is the same regime the platform's own decision players run in, so scores stay comparable with platform evals, prompts don't grow with episode length, and each turn stands alone as a training sample (Gert observations are complete, including any game-provided memory such as a notepad). Pass history="full" to accumulate the whole conversation instead; it requires an explicit game=, and games that declare turn_isolated (hidden-information games like kolmogorov, spe_recall, and transmit_receive) reject it.

Train/eval prompt parity. The adapters send the platform decision player's exact messages: rules + decision instructions as the system message, the wrapped observation (internal _action_space / _rejection_feedback fields stripped) as the user message. The user message is byte-identical to what platform evals send (Go float notation included), locked by golden fixtures shared with the platform's Go implementation; the system message is content-identical in the canonical section order (the platform randomizes rules-section order across matches). A model trained through the SDK is scored on the same prompts it trained on. Building your own loop? rollouts.system_prompt(rules) and rollouts.format_observation(observation) are those two messages; the verifiers adapter also takes renderer= (renderer(observation, rules) -> messages) for labs with their own chat template. Custom renderers forfeit parity.

Malformed output vs. losing. An action that doesn't parse as a JSON action list takes that seat out of the game, with the penalty in the reward signal. That is a strong, honest format signal for training. The episode reports completion_reason: "policy_failed" when that failure is what ended it before the game finished naturally (a game completing on the same tick keeps game_complete; the per-seat lists carry the seat-level facts). A seat the game removes by its own rules (bankruptcy, fleet destroyed) is a different thing entirely: it appears in eliminated_seats (a subset of failed_seats), is scored normally, and plays out to the real terminal, so filter malformed output on failed_seats minus eliminated_seats. The adapters expose both as gert_failed / gert_eliminated (verifiers state) and failed / eliminated (OpenEnv metadata); both action scorers count eliminated branches as legitimate samples and reject only malformed failures. Platform evals are more lenient: the decision player resamples formatting flukes, so this asymmetry is conservative (training punishes what eval forgives). If you want eval-style leniency in your own loop, resample inside act_fn before returning.

Sizing runs for GRPO-style training. Each rollout consumes one episode, so a full run needs num_tasks x rollouts-per-task x epochs episodes, so size max_episodes for the whole run (the reservation settles to actual usage at close), or pass auto_extend=N to either adapter to grow the run by N episodes whenever it exhausts instead of failing mid-training. Without it, exhaustion raises RUN_EXHAUSTED with the sizing guidance; a run can also be grown manually with client.rollout_runs.extend(run_id, additional_episodes=...).

Self-play, seat routing & curriculum. Self-play is the default: every seat is yours (read per-seat observations/rewards from the response and pass actions={seat: action} to step; multi-seat step records and results carry failed_seats and eliminated_seats as validity labels). Seats are p0..p{N-1} (a platform contract), and in full self-play policy_seats is all of them in order with p0 primary. run_episode's act_fn(observation, seat) carries the seat label, so asymmetric setups are a closure: train policy A with a weak teammate against two frozen C's via act_fn=lambda obs, seat: routing[seat](obs) with routing = {"p0": model_a, "p1": weak_b, "p2": frozen_c, "p3": frozen_c}. For single-trained-seat setups through the adapters, pass opponent_fn= ((observation, seat) -> action) so the trainer's model owns one seat (pick which with seat=, the role selector for asymmetric games) while your callable, such as a frozen checkpoint or a scripted bot, drives the rest client-side; opponent-only turns are auto-advanced so the trained model only ever sees states where it acts. Alternatively carve league seats out with opponent_seats=N (code-supporting games; 1-2 as frozen anchors for absolute-skill measurement, or player_count - 1 for the classic single-agent-vs-league mode). Open with auto_curriculum=True to let the opponent band track your win-rate on league episodes (run growth is covered under "Sizing runs" above).

Vectorized loops. client.rollout_runs.step_batch(run_id, steps) steps up to 128 episodes in one call: gather N observations, run one batched forward pass, and submit the N actions together, paying the per-request overhead once and waiting on one response instead of N request tails (entries execute concurrently; the batch counts as one request for rate limiting). Each entry is {"episode_id", "action"|"actions", "expected_tick"?}; results come back in order, with per-episode failures as ok: False entries that never fail the batch.

Train across a category & bring your own opponents. Instead of a single game=, pass tags=[...] to reset / run_episode / the adapters and Gert picks a random game matching all the tags each episode, so one run can span a whole category (the resolved game comes back as result["game"]; with tags, seed does not fix which game is chosen). Set opponent_source="own" (or the default "both") to add your org's rated submissions to the opponent league. Own opponents run user-submitted code, so they require a user-code-capable rollout backend; where unavailable, "own" is rejected and "both" uses only system opponents. Narrow the league further with opponent_models=[...] / opponent_providers=[...] (the LLMs that generated the opponents) and opponent_created_after / opponent_created_before (RFC3339). These all pass through reset / run_episode / the OpenEnv and verifiers adapters alike. An episode requesting opponent seats needs enough distinct rated opponents in the band; if a reset fails because a game is too thin, widen the band, use opponent_source="both", pick a richer tag, or drop opponent_seats (self-play needs no league). The verifiers and OpenEnv adapters fetch each resolved game's rules automatically (cached per game).

OpenEnv. For Gymnasium/OpenEnv-style trainers (TorchForge, verl, TRL, SkyRL), gertlabs.openenv_env.GertEnv wraps the same engine as reset()/step()/close().

Two-phase rollouts (cheap tails). Running your own inference to terminal is the expensive part of long episodes. Instead, drive your model through the tactically relevant near horizon with step, then call client.rollout_runs.complete(run_id, episode_id, samples=30) to play the episode out from its current state with skill-band code players -- no action injected, the parent episode untouched -- and get back score stats (a V(s) playout estimate: {score: {mean, ...}, win_rate, failure_rate, ...}). At a decision point, score_actions is the Q(s,a) sibling: it completes each candidate action the same cheap way. Combine with fork (self-play) for branch search. complete is read-only, so a fork you finish that way stays live -- release it with client.rollout_runs.close_episode(run_id, episode_id) (episodes driven to terminal auto-close on their own).

On-policy counterfactual scoring. score_actions is cheap but its branch tails are played by code players -- an off-policy Q(s,a). When you want your model's Q(s,a) -- "what happens if I take this move and then I keep playing" -- rollouts.score_actions_on_policy runs the same fan-out on-policy: it forks the live self-play episode N times per candidate, injects each candidate as the fork's first action, drives every fork in lockstep with your model (one step_batch round trip per tick; pass act_batch_fn for one batched forward pass per tick), and returns the same ranked response shape as score_actions -- so cheap shortlisting and on-policy tie-breaking are a one-call swap:

result = rollouts.score_actions_on_policy(
    client, run_id, episode_id,
    candidate_actions=[a1, a2, a3],   # e.g. 3 samples from your model at this state
    samples_per_action=8,             # 8 forks per candidate
    act_fn=act,                       # your inference plays the continuations
    phase_ticks=20,                   # optional: on-policy for 20 ticks, then a cheap
)                                     # code-player tail (complete) finishes each fork
best = result["actions"][0]           # action_rank == 1
client.rollout_runs.step(run_id, episode_id, best["action"])  # advance the main line

The fan-out is one atomic, idempotent fork_batch call (K x N <= 128): every sample restores from the same snapshot, the slot claim is all-or-nothing (RUN_EXHAUSTED consumes nothing -- extend the run and retry), and a lost response replays the same batch via its op_key (best-effort, while cached) instead of stranding one. In phase_ticks mode each tailed fork RETAINS ~tail_samples episode-equivalents after its successful complete (31 slots per fork with the defaults), so keep the run's remaining budget at max_episodes - episodes_used >= K x N x (1 + ceil(S x T / max_ticks_per_episode)), with S/T the effective tail samples/ticks (30 and the full per-episode cap when omitted). The fan-out is validated against a free read-only probe (get_episode) before any slot is claimed, so a seat typo or missing act fn costs nothing. Turn handling is automatic: the target defaults to the on-turn seat (turn-based) or the primary seat, an off-turn seat is rejected rather than silently skipped, and your model is called for exactly the seats in each result's on_turn_seats. Failures follow the server's rule exactly -- a sample is excluded (and counted in failure_rate) only when its branch introduces a NON-elimination seat failure; game-rule eliminations are normal play and stay scored, as are failures inherited from the snapshot -- and cleanup is best-effort-with-retry: forks are closed as they finish, stragglers are retried on the way out, and anything left idle-reaps in ~10 minutes billing only its actual ticks. With a deterministic act_fn on a deterministic game the N to-terminal samples are identical -- variance comes from your policy's sampling (or, in phase_ticks mode, the tail's opponent draws).

Quickstart: build a training dataset across many environments

The platform's batch engine runs your model across a whole category of game environments in one job: generate code, evaluate it, and export the results.

from gertlabs import GertClient
client = GertClient()

# Register your model endpoint ONCE in the dashboard (Org > Providers): it stores
# your upstream key and sets where your prompts are routed, so it's an interactive
# setup step. Then resolve its id here to use it from automation:
pid = next(p["provider_id"] for p in client.providers.list() if p["name"] == "my-model-v3")

# Estimate cost before spending
est = client.dataset_builds.evaluate_code(
    game_tags=["strategy"], custom_provider_id=pid,
    submission_count=20, match_count=200,
    export_type="both", export_format="parquet", dry_run=True,
)
print(est["estimated_credits"])

# Run across every strategy game: generate -> evaluate -> export
build = client.dataset_builds.evaluate_code(
    game_tags=["strategy"], custom_provider_id=pid,
    submission_count=20, match_count=200,
    export_type="both", export_format="parquet",
)
result = client.jobs.wait(build["job_id"], timeout=7200)["result"]

# Load each export (submissions + replays) into pandas, keyed by its declared type
frames_by_type = {}
for export_id in result["export_job_ids"]:
    client.jobs.wait(export_id)
    export_type = client.exports.get(export_id)["export_type"]
    frames_by_type[export_type] = client.exports.load(export_id)
    print(export_type, len(frames_by_type[export_type]), "rows")

Submission exports use one provider-attribution contract in both JSON and Parquet (schema v3; v4 adds the pinned-difficulty label on replay and branch-exploration rows, which never pair with score_context anchors): provider is the readable generation-provider label, provider_source is gertlabs for the system corpus or org:<immutable-org-slug> for organization-owned data, and model is independent. Nested branch/request/transcript metadata and replay seats use the same readable attribution; internal custom:<uuid> routing keys are not exported. org:gertlabs is distinct from the built-in source gertlabs. Rows without recoverable attribution omit it rather than using placeholders.

subs = frames_by_type["submissions"]
print(subs[["provider", "provider_source", "model"]].drop_duplicates())

Counterfactual data (branch exploration)

build = client.dataset_builds.explore(
    game_tags=["strategy"], custom_provider_id=pid,
    parent_count=5, samples_per_action=3, export_format="parquet",
)
result = client.jobs.wait(build["job_id"], timeout=7200)["result"]
client.jobs.wait(result["export_job_id"])        # singular in this mode
df = client.exports.load(result["export_job_id"])

Agentic branching (process-reward data from a generation run)

Turn an agentic generation into a Q(s,a) / process-reward data source: at each decision point (every branch_interval turns) it samples max_actions candidate actions and scores each with samples_per_action rollouts that hold the action fixed. Each record is (state, action, value) -- the action (reasoning + tool calls) is the asset, the value is its label.

sub = client.submissions.create(
    game="market_simulator", pipeline="agentic", custom_provider_id=pid,
    max_turns=60, branch=True, max_actions=3, samples_per_action=5,
    branch_interval=5, branch_mode="best_of_n",   # or "measure" (don't adopt)
)
client.jobs.wait(sub["job_id"])

# Inspect the recorded actions (action + value per decision point) ...
scores = client.jobs.branch_scores(sub["job_id"])
# ... or export the whole dataset (your own data, so it's free):
job = client.exports.create(export_type="agentic_branches", format="parquet")
client.jobs.wait(job["job_id"])
df = client.exports.load(job["job_id"])

The export is self-contained. JSON envelopes carry a transcripts map (each submission's full generation transcript), a request_contexts map (the request frame: system prompt, exact tool specs, verbatim base per-step prompts, ideation, and the effective inference settings incl. transcript_window -- message lists are a deterministic window over the transcript), and value_contexts (the durable branch_config with effective opponent/eval/model provenance, plus the pinned game version and created_at -- the submission's finalization time, an upper bound on when its decision points were recorded, vs the envelope's exported_at). The state a point's actions branched from is the request frame plus transcripts[submission_id][:state_entry_count], and each point carries pruned -- the projection's prune verdicts over that state, computed by the generation code itself and shipped as data ({} = nothing pruned; null = not computed, e.g. the frame predates the current projection rules). A score_context block keyed by game slug anchors raw scores against the game's current-regime canonical-horizon league at export time (on every export type). It carries the game's declared score_version (the scoring-regime identity; raw scores from different versions are not comparable) and canonical_ticks (the league's match horizon). Submissions rows carry score_version and score_ticks labels written with their aggregates and cleared with them on a regime bump -- a row is comparable with the anchors when its score_version matches AND its score_ticks equals canonical_ticks (a custom-horizon evaluation self-describes via its own score_ticks instead of over-claiming). Agentic values are frozen at their generation pin: pair them with the anchors when value_contexts[...].branch_config.score_version matches the anchors' score_version and branch_config.ticks equals canonical_ticks (game_version is generation provenance, not the pairing key). completeness counts present/missing pieces. In parquet the per-submission payloads are nullable transcript/request_context/value_context columns on the submission's first row IN EACH CHUNK (a submission whose decision points span chunks repeats them), so a plain exports.load() DataFrame is fully self-contained -- deduplicate before indexing:

import json

from gertlabs import transcript

df = client.exports.load(job["job_id"])
frames = (
    df.dropna(subset=["transcript", "request_context"])
      .drop_duplicates("submission_id")
      .set_index("submission_id")
)

row = df.iloc[0]
state = transcript.reconstruct_messages(
    frames.loc[row.submission_id, "transcript"],           # raw JSON string column
    json.loads(frames.loc[row.submission_id, "request_context"]),
    int(row.decision_index),
    row.pruned,                                            # exported verdict map
    state_entry_count=int(row.state_entry_count),
)
# {"system", "tools", "messages"}: the model's exact input at this decision
# point, windowed under the frame's static transcript_window (pass
# windowed=False for the full history). The structural projection is
# conformance-tested against the generation code via a golden fixture it
# generates; the prune verdicts come precomputed from that code in the export.

exports.download_files() remains the exact-artifact path, and the only way to read the parquet file metadata (score_context:<slug>, completeness).

The final submission records its branching provenance in branch_config (mode is best_of_n for a curated best-of-N selection, measure for the representative main line, absent if not branched). Filter by it anywhere submissions are read -- e.g. exclude curated output from a capability benchmark:

reps = client.submissions.list(game="market_simulator", mine=True, branch_mode="none,measure")
job = client.exports.create(export_type="submissions", branch_modes=["none", "measure"], format="parquet")

Export only top performers (filter by evaluation score)

job = client.exports.create(
    export_type="submissions", min_percentile=0.9,
    tags=["strategy"], format="parquet",
)
client.jobs.wait(job["job_id"])
df = client.exports.load(job["job_id"])

Connected model: run it server-side, then branch

Register a provider (see below) and the platform runs it as an AI seat, so you can spectate or branch without running inference yourself.

session = client.play.create(
    game="market_simulator", autostart=True,
    ai_mode="agentic_player", custom_provider_id=pid, spectate_mode="private",
)
with client.play.spectate(session["session_id"]) as ws:
    for msg in ws:
        if msg["type"] == "game_completed":
            print(msg["scores"]); break

# fan out 8 counterfactual branches from a checkpoint
client.play.branch(session["session_id"], count=8)

Fine-grained control

For a single hand-written submission instead of a batch build:

sub = client.submissions.create(game="market_simulator", language="python", code=SOURCE)
client.jobs.wait(sub["job_id"]) if "job_id" in sub else None
ev = client.submissions.evaluate(sub["submission_id"], match_count=500)
client.jobs.wait(ev["job_id"])
print(client.submissions.get(sub["submission_id"])["elo_rating"])

Resource reference

Resource	Methods
client.games	list, get, tags
client.dataset_builds	evaluate_code, explore
client.exports	list, create, get, download, download_files, load, reset_tracking
client.jobs	get, wait, branch_scores
client.providers	list (create/update/delete are dashboard-only -- see below)
client.submissions	list, get, create, delete, bulk_delete, evaluate, batch_evaluate, evaluations
client.sessions	list, get, logs, branches, branch_scores, delete
client.billing	balance, usage
client.play	create, join, branch, get, act, leave, connect, spectate
client.rollout_runs	open, list, reset, get_episode, step, step_batch, fork, fork_batch, score_actions, complete, close_episode, extend, close

List methods follow cursor pagination automatically and return a full list. Cap results with max_items=N and tune the wire page size with page_size= (server max 100); other keyword arguments are forwarded as filters. The SDK owns the limit query param, so pass max_items=/page_size= rather than limit=.

Providers are configured in the dashboard

Provider registration isn't available through the SDK: it stores a secret and is gated to logged-in dashboard sessions (API keys are rejected). Create a provider in the dashboard under Org > Providers; the SDK only lists them (client.providers.list()), so you can resolve a provider_id. Model identifiers in allowed_models are recorded verbatim as provenance on everything they generate (submissions, sessions, branch data, exports) -- register each checkpoint as a distinct model rather than re-pointing an existing identifier at new weights.

Errors

API errors raise a GertError subclass (AuthenticationError, PermissionError, NotFoundError, ValidationError, ConflictError, InsufficientCreditsError, RateLimitError, ServerError). Each carries .code, .status, .request_id, and .body (the full parsed error response). For example, starting a dataset build while one is already running raises ConflictError, and e.body["existing_job_id"] is the id of the in-flight job:

from gertlabs import ConflictError
try:
    build = client.dataset_builds.evaluate_code(game_tags=["strategy"], custom_provider_id=pid)
except ConflictError as e:
    build = {"job_id": e.body["existing_job_id"]}   # wait on the existing build
client.jobs.wait(build["job_id"])