Python Server SDK

kraty-admin — server-side Python client for the /server/v1 admin surface. Manual grants, IAP fulfilment, inventory grant/revoke, wallet credit/debit, lobby push.

kraty-admin is the Python server-side SDK for the Kraty platform. Same surface as the Node server SDK, designed for backends written in Python (Django, FastAPI, Flask, Celery workers, or plain scripts). Built on httpx with auto-stamped idempotency keys, retry/backoff with jitter, and typed error helpers.

Requires Python 3.10+.

Server-side only. Authenticated with a server_integration API key that can mint currency and items. Never bundle into a mobile app, Pyodide build, or anywhere a player could reach the running code.

Install

uv add kraty-admin
# or
pip install kraty-admin

Quickstart

import os
from kraty_admin import KratyAdmin

kraty = KratyAdmin(api_key=os.environ["KRATY_SERVER_KEY"])

# IAP fulfilment, idempotent on the receipt id:
kraty.wallet.credit(
    "player_42",
    "gold",
    amount=500,
    reason="iap",
    source_ref_id="apple_receipt_abc",
    idempotency_key="apple_receipt_abc",
)

kraty.inventory.grant(
    "player_42",
    "starter_chest",
    quantity=1,
    reason="iap",
    idempotency_key="apple_receipt_abc",
)

# Or a single mixed grant — currencies + items + crates atomically:
kraty.grants.create(
    "player_42",
    idempotency_key="apple_receipt_abc",
    entries=[
        {"type": "currency", "currencyKey": "gold", "amount": 500},
        {"type": "item",     "itemKey": "starter_chest", "quantity": 1},
        {"type": "crate",    "crateItemKey": "legendary_box", "quantity": 2},
    ],
    source_kind="api",
    source_ref_id="apple_receipt_abc",
)

kraty.close()

Or as a context manager:

with KratyAdmin(api_key=os.environ["KRATY_SERVER_KEY"]) as kraty:
    kraty.wallet.credit("player_42", "gold", amount=500,
                        idempotency_key="receipt_id")

Resource clients

kraty.grants      # create (manual mint) / ack
kraty.inventory   # grant / revoke
kraty.wallet      # credit / debit
kraty.lobbies     # push (pre-matched) / read
kraty.players     # get (unified snapshot)
kraty.migrate     # bulk-import players / wallet / inventory
kraty.health      # ping

Naming convention

Method args are snake_case (Pythonic); request body keys go out camelCase (matching the API contract — the SDK does the translation). Reward entries inside grants.create keep the wire shape (currencyKey, crateItemKey) because they're nested under the entries array.

Idempotency

Every POST is auto-stamped with a UUID idempotency_key if you don't supply one — but for server-side fulfilment always pass your own (typically the IAP receipt id):

Replays of the same fulfilment return the original grant.
A misconfigured retry with a different body returns KratyServerError with .is_idempotency_conflict == True.

from kraty_admin import KratyServerError

try:
    kraty.wallet.credit("p", "gold",
                        amount=500,
                        idempotency_key=receipt_id)
except KratyServerError as err:
    if err.is_idempotency_conflict:
        alert_ops(receipt_id=receipt_id)
    else:
        raise

Cache TTL is 24 hours per key.

Manual grants

The richest endpoint — combines any of currency, items, and crates atomically:

grant = kraty.grants.create(
    "player_42",
    idempotency_key="iap_starter_pack",
    entries=[
        {"type": "currency", "currencyKey": "gold", "amount": 500},
        {"type": "currency", "currencyKey": "gems", "amount": 50},
        {"type": "item",     "itemKey": "starter_chest", "quantity": 1},
    ],
    source_kind="api",
    source_ref_id="apple_receipt_abc",
    metadata={"receipt": receipt_body, "attribution": "campaign_42"},
)

Server-side claim (no client round-trip needed):

kraty.grants.ack("player_42", grant["id"])

Records ackedBy='server_api' on the audit row.

Inventory grant / revoke

kraty.inventory.grant(
    "player_42", "health_potion",
    quantity=10,
    reason="iap_potion_pack",
    idempotency_key="apple_receipt_xyz",
)

# Refund / chargeback path:
kraty.inventory.revoke(
    "player_42", "health_potion",
    quantity=10,
    reason="chargeback",
    idempotency_key="chargeback_xyz",
)

revoke returns 409 on insufficient quantity.

Wallet credit / debit

kraty.wallet.credit("p", "gold",
                    amount=500,
                    reason="iap",
                    idempotency_key="receipt")

kraty.wallet.debit("p", "gold",
                   amount=100,
                   reason="refund",
                   idempotency_key="refund_xyz")

The client SDKs can also debit; only the server SDKs (this one and @kraty/server-sdk) can credit. Mint money server-side; let clients spend.

Push lobbies

For studios that match outside Kraty (Steam, GameLift, Photon) and just want Kraty to host the event window + scoring:

lobby = kraty.lobbies.push(
    "game_1", "quick_brawl",
    key="matchmaker_lobby_123",     # idempotency key
    external_player_ids=["alice", "bob", "carol"],
    capacity=4,
    fill_bots=False,
)

Requires the event's leaderboardMode to be 'lobby_matched'.

lobby = kraty.lobbies.read("game_1", lobby_id)

Player snapshot

Unified view for support tools:

snap = kraty.players.get("player_42")
print(snap["player"]["externalPlayerId"])
print(snap["inventory"])      # list of item holdings
print(snap["wallet"])         # list of wallet holdings
print(snap["recentGrants"])   # list of grants

The Python SDK returns plain dicts (no Pydantic models in v0 — the wire shape is fully documented in the REST API reference so you can apply your own schema layer if you want one).

kraty.players.delete honours an Article 17 right-of-erasure request. Anonymizes the player row + cascades through attempts, lobbies, and Redis leaderboard meta. The financial ledger is retained per audit requirements but points at an anonymized row whose external id is a __deleted_<uuid>__ placeholder.

out = kraty.players.delete("player_42", reason="gdpr_erasure")
if out["status"] == "erased":
    # Cascade ran; player.deleted webhook fired with the original
    # external id so your own systems can mirror the deletion.
    pass
# `no_op_never_existed` is also a success — there was no data
# for this externalId in the first place.

kraty.players.export returns the full bundle for an Article 15 right-of-access request. Each list is capped at 1,000 rows. Raises KratyServerError (with is_not_found = True) when the player is unknown.

bundle = kraty.players.export("player_42")
with open("player-42-export.json", "w") as f:
    json.dump(bundle, f, indent=2)

Full flow walkthrough: Common integration tasks → GDPR.

Soft-ban a player

kraty.players.ban flags a player as banned. Subsequent player-scoped SDK writes return 403 player_banned. Existing scores, lobby memberships, and grants stay intact (soft ban).

kraty.players.ban("player_42", reason="score anomaly: gained 5000 in 2s")
# ... later:
kraty.players.unban("player_42")

Both methods are idempotent. Portal operators can ban/unban from the Player Lookup screen.

Migrating from another platform

When you bring players in from PlayFab, Firebase, Lootlocker, or your own backend, kraty.migrate does bulk import in batches of up to 1,000 rows per call.

Each row carries its own idempotencyKey — typically your stable id for the player / wallet entry / inventory holding — so retries are safe at the row level. Bad rows land in outcome["failures"]; the rest of the batch still applies.

out = kraty.migrate.players([
    {"externalPlayerId": "p_1", "idempotencyKey": "p_1"},
    {"externalPlayerId": "p_2", "idempotencyKey": "p_2",
     "contextSnapshot": {"country": "PT"}},
])
print(f'{out["applied"]} created, {out["skipped"]} replayed, {out["failed"]} failed')

kraty.migrate.wallet([
    {"externalPlayerId": "p_1", "economyKey": "gold", "amount": 1500,
     "idempotencyKey": "p_1:gold"},
])

kraty.migrate.inventory([
    {"externalPlayerId": "p_1", "itemKey": "starter_chest", "quantity": 1,
     "parameters": {"rolled": {"atk": 4}},
     "idempotencyKey": "p_1:starter_chest"},
])

Webhooks are not emitted during migration — a 100k-player import would otherwise flood your own backend with player.registered / inventory.changed / wallet.changed deliveries. Run any onboarding side-effects yourself after the import completes.

For larger datasets, loop client-side in chunks of 1,000:

from itertools import islice

def chunked(seq, n):
    it = iter(seq)
    while batch := list(islice(it, n)):
        yield batch

for chunk in chunked(all_players, 1000):
    out = kraty.migrate.players(chunk)
    if out["failed"] > 0:
        collect_for_retry(out["failures"])

Retries

from kraty_admin import KratyAdmin, RetryConfig

kraty = KratyAdmin(
    api_key="...",
    retry=RetryConfig(
        attempts=5,
        initial_delay=0.5,
        max_delay=30.0,
        jitter=0.25,
    ),
)

Retries fire on 408 / 425 / 429 / 5xx and on httpx network errors. Retry-After is honored.

Errors

from kraty_admin import KratyServerError, KratyNetworkError

try:
    kraty.grants.create("player_42", ...)
except KratyServerError as err:
    if err.is_idempotency_conflict:
        ...  # duplicate fulfilment with different body
    elif err.is_not_found:
        ...  # player or item doesn't exist in this game
    elif err.is_forbidden:
        ...  # wrong key for this game/studio
    elif err.is_rate_limited:
        ...  # 429 — retry budget exhausted
except KratyNetworkError as err:
    ...  # backend unreachable; err.original_cause has the underlying exception

Typed properties on KratyServerError:

is_idempotency_conflict — 409 idempotency_conflict
is_not_found — 404 not_found
is_forbidden — 403 forbidden
is_rate_limited — 429 rate_limited

Full code reference: Error codes.

Verify incoming webhooks

The SDK ships a verify_webhook helper so your receiver doesn't have to hand-roll the HMAC verification — and doesn't accidentally introduce a timing leak or replay window bug in the process.

import os
from fastapi import FastAPI, Header, HTTPException, Request
from kraty_admin import verify_webhook

app = FastAPI()

@app.post("/kraty/webhook")
async def kraty_webhook(
    request: Request,
    x_signature: str = Header(...),
):
    # CRITICAL: read the raw bytes — re-serialising the parsed JSON
    # can change byte order / whitespace and break the HMAC.
    raw = await request.body()
    if not verify_webhook(
        raw_body=raw,
        signature_header=x_signature,
        secret=os.environ["KRATY_WEBHOOK_SECRET"],
    ):
        raise HTTPException(status_code=401, detail="bad signature")

    event = await request.json()
    match event["eventName"]:
        case "grant.created":     pass  # …
        case "player.registered": pass  # …
        case "event.completed":   pass  # …
        # see /docs/webhooks for the full kind catalog
    return {"ok": True}

Defaults: 5-minute replay window, 60-second forward-clock tolerance, constant-time compare. Pass tolerance_seconds to widen the replay window for delivery-queue backlog scenarios. See Webhooks for the full event catalog and signature format.

Telemetry

def on_request(info):
    metrics.timing(f"kraty_server.{info.url}", info.duration_ms or 0)
    if not info.ok:
        metrics.increment(f"kraty_server.error.{info.status}")

kraty = KratyAdmin(api_key="...", on_request=on_request)

Fires once per HTTP attempt. Use info.attempt to dedupe.

Async support

The v0 SDK is sync-only — wrap with asyncio.to_thread if you need to call from an async context (FastAPI, Litestar):

import asyncio

async def fulfill_iap(receipt):
    await asyncio.to_thread(
        kraty.wallet.credit,
        receipt.player_id, "gold",
        amount=500,
        idempotency_key=receipt.id,
    )

A native async client (KratyAdminAsync) is on the roadmap once we have a real async-heavy consumer to validate the ergonomics against.

Resource reference

Client	Methods
`kraty.grants`	`create(external_player_id, idempotency_key, entries, ...)`, `ack(external_player_id, grant_id)`
`kraty.inventory`	`grant(external_player_id, item_key, quantity, ...)`, `revoke(external_player_id, item_key, quantity, ...)`
`kraty.wallet`	`credit(external_player_id, economy_key, amount, ...)`, `debit(external_player_id, economy_key, amount, ...)`
`kraty.lobbies`	`push(game_id, event_key, key, external_player_ids, ...)`, `read(game_id, lobby_id)`
`kraty.players`	`get(external_player_id)`, `delete(external_player_id, reason=...)`, `export(external_player_id)`, `ban(external_player_id, reason=...)`, `unban(external_player_id)`, `merge(from_external_player_id, to_external_player_id)`
`kraty.migrate`	`players(rows)`, `wallet(rows)`, `inventory(rows)` — bulk import, 1,000 rows max
`kraty.health`	`ping()`