txapi-sdk — official Python SDK for the txApi

Typed Python client for the txApi REST and WebSocket API: manage your FiveM servers' txAdmin — players, bans, warnings, whitelist, server control, the live console, the stored event log and txAdmin admins — from Python.

• Sync (TxApi) and async (AsyncTxApi) clients with the identical surface
• Every one of the 67 endpoints of API 1.8.0, fully type-hinted (mypy --strict clean), responses as TypedDicts
• Automatic retries that never repeat a ban or a restart, rate-limit information on every call
• Auto-paging iterators, webhook signature verification, live console / server log streams
• Python 3.9+, one runtime dependency (httpx); websockets only for live streams

pip install txapi-sdk            # REST client + webhooks
pip install "txapi-sdk[live]"    # + live console / server-log streams (websockets)

The distribution is called txapi-sdk (the name txapi on PyPI belongs to someone else); the import stays import txapi. Every release is also attached to GitHub Releases and served from https://txapi.cloud/sdks/ (checksums in SHA256SUMS), for setups without PyPI access:

pip install "txapi-sdk[live] @ https://txapi.cloud/sdks/txapi_sdk-1.0.1-py3-none-any.whl"

Contents

• Quickstart
• Configuration
• Calling endpoints
• Errors
• Retries
• Rate limits and raw responses
• Pagination
• Webhooks
• Live console and server log
• Hooks, custom HTTP clients, timeouts
• Endpoint reference
• Development

Quickstart

Create an API key in the dashboard (API keys), then:

from txapi import TxApi

with TxApi(api_key="txp_...", server_id=1) as client:
    page = client.players.search(query="john")
    player = page["players"][0]
    client.players.ban(
        identifier=f"license:{player['license']}",
        reason="Cheating",
        duration_hours=168,  # omit (or 0) for a permanent ban
    )

The same with asyncio:

import asyncio
from txapi import AsyncTxApi

async def main() -> None:
    async with AsyncTxApi(server_id=1) as client:  # key from TXAPI_KEY
        status = await client.server.status()
        print(status["serverName"], "online" if status.get("online") else "offline")
        await client.server.announcement(message="Restart in 5 minutes")

asyncio.run(main())

Never log or commit your API key. It is sent as Authorization: Bearer <key>; the clients' and errors' repr()/str() never contain it. Prefer the TXAPI_KEY environment variable over hard-coding it. The request headers passed to on_request hooks do contain it — don't log them.

Configuration

client = TxApi(server_id=1, timeout=10, max_retries=3, user_agent="my-panel/2.1")
print(client.account.servers())                # account-level: no serverId is sent
client.server.status(server_id=2)              # override the default server for one call

txapi.__version__ is the SDK version (1.0.1), txapi.API_VERSION the API version it was built for (1.8.0).

Calling endpoints

Endpoints are grouped like the API: client.account, players, history, server (+ server.resources), console, serverlog, events, whitelist (+ whitelist.approvals) and admins. Path parameters are positional, everything else is keyword-only and snake_case (sent as camelCase). None arguments are omitted — never sent as null. Every method's docstring names the permission it needs.

me = client.account.me()
print(me["plan"]["code"], me["quota"].get("remaining"), me["plan"]["features"])

# Players
client.players.info(identifier="discord:272800190639898628")
client.players.warn(identifier="license:abc…", reason="Mic spam")
client.players.kick(identifier="license:abc…", reason="AFK")
client.players.dm(identifier="license:abc…", message="Please read the rules")
client.players.notes(identifier="license:abc…", note="Watch this one")
client.players.whitelist(identifier="license:abc…", whitelisted=True)

# Bans incl. Discord (API 1.8.0): also ban on every Discord server linked via txBot
result = client.players.ban(identifier="license:abc…", reason="Cheating", discord_ban=True)
for guild in result.get("discord", {}).get("guilds", []):
    print(guild["guildName"], guild["code"])     # banned, missing_permission, hierarchy, …
client.history.revoke("BKL5-3SH4", reason="Appeal accepted", discord_unban=True)

# Server control
client.server.restart()
client.server.schedule_restart(time="+30")       # in 30 minutes; or "04:00" (host clock)
client.server.skip_restart(skip=True)
client.server.kick_all(reason="Maintenance")
client.server.resources.restart(name="es_extended")

# Console (REST) — see below for the live stream
client.console.command(command="say Hello")
print(client.console.buffer()["lines"][-20:])

# Stored event log
board = client.events.leaderboard(since="7d", sort="kills", limit=10)
stats = client.events.player(player="license:abc…", since="30d")
deaths = client.events.deaths(pvp=True, since="24h")
client.events.list(type=["ban", "warn"], since="2026-10-01T00:00:00Z")

# Whitelist
client.whitelist.approve("R1234")
client.whitelist.approvals.add(identifier="discord:272800190639898628")
client.whitelist.approvals.remove("discord:272800190639898628")   # percent-encoded for you

# txAdmin admins
client.admins.add(name="Moderator Bob", permissions=["players.ban", "console.write"])
client.admins.edit("Moderator Bob", discord_id="123")   # only the fields you pass change
client.admins.delete("Moderator Bob")

Responses are parsed JSON typed as TypedDicts (txapi.types). They are plain dicts at runtime, so fields the API adds in later minor versions never break anything. txAdmin passthrough endpoints (e.g. players.info, server.diagnostics) return JsonValue (Any). Write endpoints return {"success": True, "message": "…", "discord"?: …}.

Required arguments are checked before anything is sent: blank required fields and values longer than the API allows (ban reason 512, note 2000, resource name 128, console command 1024 characters) raise txapi.ValidationError without a request.

Errors

Every exception derives from txapi.TxApiError with status, message (the API's error text, else the body, else the reason phrase), parsed body, headers and rate_limit.

from txapi import NotFoundError, RateLimitError, TxAdminRejectedError, TxApiError

try:
    client.players.kick(identifier="license:abc…", reason="AFK")
except TxAdminRejectedError as exc:
    print("txAdmin refused:", exc.message)        # e.g. the player is offline
except NotFoundError:
    print("No such player")
except RateLimitError as exc:
    if exc.quota_exhausted:
        print("Monthly quota used up — resets on the 1st, 00:00 UTC")
    else:
        print(f"Too fast, retry in {exc.retry_after} s")
except TxApiError as exc:
    print(exc.status, exc.message, exc.body)

APIConnectionError and APITimeoutError are named so to not shadow Python's built-in ConnectionError/TimeoutError (the spec's names).

Retries

• 429 with Retry-After ≤ 60 s is retried for every method after exactly that delay — the request was not executed. 429 without Retry-After means the monthly quota is used up and is never retried.
• Connection errors, timeouts, 502, 503, 504 are retried for GET only. A retried POST could ban twice or restart a server twice.
• Backoff: full jitter, random() * min(8 s, 0.5 s · 2^attempt), unless Retry-After says otherwise.
• max_retries=0 disables everything. Cancelling an AsyncTxApi call (task cancel) stops waiting immediately.

Rate limits and raw responses

Every response carries the per-minute limit; the SDK parses it into RateLimit(limit, remaining, reset_seconds):

client.server.status()
rl = client.last_rate_limit          # None before the first response
print(rl.remaining, "of", rl.limit, "left, resets in", rl.reset_seconds, "s")

Errors carry it too (exc.rate_limit). For status code and headers together with the parsed data, use with_raw_response — the same methods, returning txapi.ApiResponse:

raw = client.with_raw_response.players.ban(identifier="license:abc…", reason="x")
raw.data            # {"success": True, "message": "…"}
raw.status_code     # 200
raw.headers         # httpx.Headers
raw.rate_limit      # RateLimit(...)
raw.http_response   # the httpx.Response

Pagination

List endpoints return one page; the iter_* variants page through everything lazily (the next page is fetched only when the current one is used up):

for player in client.players.iter_search(sort_by="tsLastConnection"):
    print(player["displayName"])

for ban in client.history.iter_bans():
    if ban.get("isRevoked"):
        continue
    print(ban["id"], ban["playerName"], ban["reason"])

for death in client.events.iter_deaths(since="7d", pvp=True):
    print(death["killer"], "→", death["victim"])

async for row in client.events.iter_admin_actions(since="24h"):   # AsyncTxApi
    print(row)

Iteration stops on hasReachedEnd, a missing nextCursor or an empty page. To page by hand, pass offset=/cursor= to the non-iterating method.

Webhooks

Webhooks (dashboard → Webhooks) are signed with your endpoint secret (whsec_…): X-Txapi-Signature: t=<unix seconds>,v1=<hex HMAC-SHA256>. txapi.verify_webhook checks the signature (constant time, several v1 entries allowed during secret rotation) and the timestamp (default tolerance 300 s), and returns the event {id, event, createdAt, serverId, data}:

from txapi import WebhookVerificationError, verify_webhook

event = verify_webhook(raw_body, signature_header, secret)

Verify the raw request body — the exact bytes you received. Parsing the JSON and serialising it again (json.dumps(request.json())) changes whitespace and key order and the signature will never match.

On failure it raises WebhookVerificationError with a reason: missing_timestamp, missing_signature, malformed_header, timestamp_out_of_tolerance, signature_mismatch, invalid_body or missing_secret. All event names are in txapi.WEBHOOK_EVENTS (and the txapi.WebhookEventName literal type); header names in txapi.webhooks.

Flask

from flask import Flask, abort, request
from txapi import WebhookVerificationError, verify_webhook

app = Flask(__name__)

@app.post("/txapi/webhook")
def txapi_webhook():
    try:
        event = verify_webhook(
            request.get_data(),                          # raw bytes
            request.headers.get("X-Txapi-Signature"),
            WEBHOOK_SECRET,
        )
    except WebhookVerificationError as exc:
        abort(400, exc.reason)
    if event["event"] == "player.banned":
        notify_staff(event["data"])
    return {"ok": True}

FastAPI — see examples/fastapi_webhook.py:

@app.post("/txapi/webhook")
async def txapi_webhook(request: Request):
    raw = await request.body()                               # raw bytes
    try:
        event = verify_webhook(raw, request.headers.get("X-Txapi-Signature"), WEBHOOK_SECRET)
    except WebhookVerificationError as exc:
        raise HTTPException(400, exc.reason)
    ...
    return {"ok": True}

Answer with a 2xx quickly; failed deliveries are retried by txApi.

Live console and server log

Needs pip install "txapi-sdk[live]". client.console.connect() (permission console.write) and client.serverlog.connect() (logs.view) return asyncio streams:

import asyncio
from txapi import AsyncTxApi
from txapi.live import CloseEvent, InitEvent, LogEvent

async def main() -> None:
    async with AsyncTxApi(server_id=1) as client:
        async with client.console.connect() as console:
            print(await console.send_command("status"))   # resolves with txAdmin's ack
            async for event in console:
                if isinstance(event, InitEvent):           # buffer after every (re)connect
                    print(len(event.items), "buffered lines")
                elif isinstance(event, LogEvent):          # one event per console line
                    print(event.item)
                elif isinstance(event, CloseEvent) and not event.will_reconnect:
                    print("closed:", event.code, event.reason)

asyncio.run(main())

Events (txapi.live): OpenEvent(reconnected), InitEvent(items), LogEvent(item), ErrorEvent(message), CloseEvent(code, reason, will_reconnect, by_user), ReconnectingEvent(attempt, delay, code, reason) and, with raw_frames=True, FrameEvent(frame). Instead of iterating you can pass on_event= (sync or async callback).

• send_command(cmd) validates the length (≤ 1024), sends {"type": "command"} and returns the next ack — or raises CommandError for the next error frame. Commands are correlated first-in-first-out; timeout 10 s (command_timeout= / timeout=) raises APITimeoutError. The command's output arrives later as LogEvents. An error frame that answers no pending command is emitted as ErrorEvent.
• Reconnects automatically after close codes 1001, 1006, 1011, 1012 and network loss (backoff 1 s → 30 s with jitter; reconnect=, max_reconnect_attempts=, initial_reconnect_delay=, max_reconnect_delay=). After a reconnect the server sends the full buffer again as InitEvent — de-duplicate or ignore it.
• No reconnect after 1008 (authentication, permission, quota, too slow — event.reason and stream.close_reason tell you what to fix), 1009, a refused handshake, or close(). Iteration then ends.
• Server-log LogEvent.items are txAdmin's structured entries (dict) or plain strings.
• The key travels in the WebSocket URL's query string (that is how the server authenticates streams); it is never included in events or error messages.

The synchronous TxApi hands out the same asyncio streams — use them inside asyncio.run(...); there is no blocking stream API. See examples/live_console.py.

Hooks, custom HTTP clients, timeouts

import httpx, logging

def log_request(request: httpx.Request) -> None:
    logging.info("→ %s %s", request.method, request.url.path)   # never log request.headers

def log_response(response: httpx.Response) -> None:
    logging.info("← %s", response.status_code)

client = TxApi(
    http_client=httpx.Client(limits=httpx.Limits(max_connections=20)),
    on_request=log_request,       # called before every attempt
    on_response=log_response,     # called for every response, also errors and retries
    timeout=15,
)
client.server.diagnostics(timeout=60)   # per call

Hooks can never break a request — exceptions inside them are logged and ignored. With AsyncTxApi they may be coroutines. A client you pass in is not closed by close().

Endpoint reference

Plus client.console.connect() (WS /ws/console, console.write) and client.serverlog.connect() (WS /ws/serverlog, logs.view).

Development

python -m venv .venv && .venv/bin/pip install -e ".[live,dev]"
.venv/bin/pytest                    # offline: mock HTTP transport + local WebSocket server
.venv/bin/mypy --strict src
.venv/bin/ruff check src scripts tests examples && .venv/bin/ruff format --check src scripts tests examples
python scripts/generate.py          # regenerate src/txapi/_endpoints.py + _generated.py
python scripts/generate.py --check  # CI: fails if the generated code is stale

The endpoint methods are generated from ../spec/endpoints.json (itself generated from the API catalog). When the API changes, regenerate; the coverage test fails until every endpoint has a method.

License

MIT — see LICENSE.