Python Server SDK

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

kraty-server-sdk 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

The package isn't on PyPI yet, so install directly from the public GitHub repo against a tagged release.

# With uv (recommended):
uv add 'kraty-server-sdk @ git+https://github.com/PedroTrincheiras/kraty-sdk-server-python.git@v0.12.0'

# Or pip:
pip install 'git+https://github.com/PedroTrincheiras/kraty-sdk-server-python.git@v0.12.0'

Browse releases at github.com/PedroTrincheiras/kraty-sdk-server-python/releases. Once we publish to PyPI (planned for v1.0) you'll be able to swap to uv add kraty-server-sdk.

Quickstart

import os
from kraty_server_sdk import KratyServer

kraty = KratyServer(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, with 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 KratyServer(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.leaderboards  # submit_score (server-authoritative, ungated)
kraty.events        # report_progress (server-authoritative) / finish
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, since the SDK does the translation). Reward entries inside grants.create keep the wire shape (currencyKey, crateItemKey) because they are nested under the entries array.

Idempotency

Every POST is auto-stamped with a UUID idempotency_key if you do not 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_server_sdk 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, combining 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",
    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)

Server-authoritative scoring

The server key is trusted; it bypasses the acceptClientScores gate that can lock client SDKs out of scoring. Use these methods when scoring lives on your backend (anti-cheat verification, simulation results, server match outcomes), or when a board / event has client scoring turned off.

Submit a leaderboard score

result = kraty.leaderboards.submit_score(
    "player_42",        # external player id
    "daily_steps",      # board key
    8421,               # value
    segment="EU",
    idempotency_key="steps_2026_06_29",
)
print(result["leaderboardId"], result["score"], result["rank"])

Only score-ranked boards accept this; a progression-ranked board raises KratyServerError with code='score_not_supported' (400); adjust the progression item via a grant instead. Segmentation: pass segment for context boards, omit it for progression boards (the server derives the bucket from the player's progression state). Unknown player or board raises KratyServerError with is_not_found = True.

Report event progress

Drives an in-flight attempt the player started client-side, by attempt_id:

result = kraty.events.report_progress(
    "player_42",     # external player id
    "bounty_hunt",   # event key
    attempt_id,
    mode="increment",   # "set" writes the value; "increment" adds
    metric_value=1,
)
for fired in result["milestonesFired"]:
    ...  # same milestone payload the client progress path returns

This is the server counterpart to the client SDK's progress endpoint, required when the event has Accept client scores off. See Server-authoritative scoring on the events page. report_progress is also available under the shorter progress alias, kraty.events.progress(...), with an identical signature.

Finish an attempt

Finalizes an in-flight attempt at its current score, the server counterpart to the client SDK's finish endpoint. Use it to end an untimed score-attack run from your backend (that kind has no completion target, so nothing else ends it before the window closes):

res = kraty.events.finish(
    "player_42",     # external player id
    "score_attack",  # event key
    attempt_id,      # the attempt the player started client-side
)
# res -> {"attempt": {...}, "outcome": "completed" | "expired"}
# "completed": no target, or target met, so completion rewards rolled.
# "expired":   had a target the player did not meet, so participation
#              only, same as a timeout.

Rejected on session / lobby events (those end via their session rules) and on an already-finished attempt. See How an attempt ends on the events page.

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

Friends (read-only)

Read any player's social graph for support / moderation tooling. These are read-only: the player's own add / accept / block actions happen through the client SDKs' friends client, not the server key. Each method takes an external_player_id and raises KratyServerError with is_not_found = True for an unknown player.

# Accepted friends, enriched with display identity + live presence:
friends = kraty.players.friends("player_42")
for f in friends:
    print(f["externalPlayerId"], f["online"], f["status"], f["lastActiveAt"])
    print(f["displayIdentity"])  # {"name", "avatar", "country"} | None

# Pending requests, both directions:
reqs = kraty.players.friend_requests("player_42")
print(len(reqs["incoming"]), len(reqs["outgoing"]))
# each request: {"requestId", "direction", "player": {...}, "createdAt"}

# Players this player has blocked:
blocked = kraty.players.blocks("player_42")
for b in blocked:
    print(b["externalPlayerId"], b["blockedAt"])

Like the rest of the SDK these return plain dicts with the wire's camelCase keys (friends / blocks unwrap data.friends / data.blocked into a list; friend_requests returns the {"incoming": [...], "outgoing": [...]} dict).

Override a player's synthetic identity

Synthetic identities are auto-generated from the game's identity pool on first contact. Call kraty.players.set_identity to override with a custom display name + optional avatar (e.g. the player's chosen handle). The value shows on leaderboards and in the client SDK's snapshot. The player must already be registered — 404 otherwise.

kraty.players.set_identity(
    "player_42",
    name="CaptainAlice",
    avatar="https://cdn.example.com/avatars/alice.png",
)

Pass avatar=None (or omit) to clear a previously-set avatar and fall back to the pool default.

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_server_sdk import KratyServer, RetryConfig

kraty = KratyServer(
    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_server_sdk 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 does not 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 does not have to hand-roll the HMAC verification, and does not accidentally introduce a timing leak or replay window bug in the process.

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

app = FastAPI()

@app.post("/kraty/webhook")
async def kraty_webhook(
    request: Request,
    x_signature: str = Header(...),
):
    # CRITICAL: read the raw bytes, since 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 = KratyServer(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 (KratyServerAsync) 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.leaderboards`	`submit_score(external_player_id, key, value, segment=..., idempotency_key=...)`, server-authoritative, ungated
`kraty.events`	`report_progress(external_player_id, event_key, attempt_id, mode=..., ...)` (alias: `progress`), server-authoritative progress; `finish(external_player_id, event_key, attempt_id)` finalizes an attempt
`kraty.lobbies`	`push(game_id, event_key, key, external_player_ids, ...)`, `read(game_id, lobby_id)`
`kraty.players`	`get(external_player_id)`, `set_identity(external_player_id, name=..., ...)`, `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)`, `friends(external_player_id)`, `friend_requests(external_player_id)`, `blocks(external_player_id)` (read-only social graph)
`kraty.migrate`	`players(rows)`, `wallet(rows)`, `inventory(rows)`, bulk import, 1,000 rows max
`kraty.health`	`ping()`