Metadata-Version: 2.4
Name: gertlabs
Version: 0.2.0
Summary: Python SDK for the Gert Labs AI evaluation platform
Project-URL: Homepage, https://gertlabs.com
Project-URL: Repository, https://github.com/Gert-Labs-Inc/gert_labs
Project-URL: Documentation, https://gertlabs.com/docs
Author: Gert Labs
License-Expression: MIT
License-File: LICENSE
Requires-Python: >=3.10
Requires-Dist: httpx>=0.27
Requires-Dist: websocket-client>=1.6
Provides-Extra: data
Requires-Dist: pandas>=2.0; extra == 'data'
Requires-Dist: pyarrow>=15.0; extra == 'data'
Provides-Extra: test
Requires-Dist: pytest>=8.0; extra == 'test'
Provides-Extra: verifiers
Requires-Dist: verifiers<0.2,>=0.1.14; extra == 'verifiers'
Description-Content-Type: text/markdown

# 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

```bash
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:

```bash
export GERT_API_KEY=sk_gert_...
```

```python
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.

```python
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:

```python
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`).

```python
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 +
a skill-banded slice of the rated opponent league + the scoreboard, and your
trainer calls your own model. Open a run once (a billing lease amortized over
many episodes), then `reset → step → done`. Reward is the policy seat's dense
per-tick score delta; the opponent skill band (`min_percentile`/`max_percentile`)
is your curriculum knob.

```python
from gertlabs import GertClient, rollouts
client = GertClient()

run = client.rollout_runs.open(max_episodes=1000)

def act(observation):
    # observation includes `_action_space` (legal actions) when the game enumerates them.
    # Call YOUR model here, parse its output into an action:
    return rollouts.parse_action(my_model(rollouts.format_observation(observation)))

result = rollouts.run_episode(
    client, run["run_id"], game="pot_limit_omaha_hilo",
    act_fn=act, min_percentile=0.4, max_percentile=0.6,   # hard-but-winnable band
)
print(result["total_reward"], result["outcome"])          # ... train on result["steps"] ...

client.rollout_runs.close(run["run_id"])                  # settles billing at actual ticks
```

Low-level access is `client.rollout_runs.reset/step/close` if you want to manage
the loop yourself. For [Prime Intellect `verifiers`](https://github.com/PrimeIntellect-ai/verifiers)
users, `pip install 'gertlabs[verifiers]'` and:

```python
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:`)
```

**Self-play & curriculum.** Pass `seats=N` to `reset` to control N seats with your
own policy (read per-seat `observations`/`rewards` from the response and pass
`actions={seat: action}` to `step`). Open with `auto_curriculum=True` to let the
opponent band track your win-rate, and grow a long run with
`client.rollout_runs.extend(run_id, additional_episodes=...)`.

**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 well-stocked game matching all the tags each episode — 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. A game needs enough distinct
rated opponents in the band to be playable; if a reset fails because a game is too thin,
widen the band, use `opponent_source="both"`, or pick a richer tag. 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()`.

## 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.

```python
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
for export_id in result["export_job_ids"]:
    client.jobs.wait(export_id)
    df = client.exports.load(export_id)
    print(len(df), "rows")
```

## Counterfactual data (branch exploration)

```python
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"])
```

## Export only top performers (filter by evaluation score)

```python
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.

```python
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:

```python
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`, `load`, `reset_tracking` |
| `client.jobs` | `get`, `wait` |
| `client.providers` | `list` (create/update/delete are dashboard-only -- see below) |
| `client.submissions` | `list`, `get`, `create`, `delete`, `evaluate`, `batch_evaluate` |
| `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`, `reset`, `step`, `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`.

## 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:

```python
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"])
```
