Nestra-PY

Oficjalne SDK Pythona do API i Gatewaya Nestra. Klient REST, klient WebSocket, embedy, komendy prefiksowe i reconnect w jednym pakiecie.

Cechy

• REST: gildie, kanały, wiadomości, członkowie, moderacja (kick/ban/timeout/role), reakcje, zaproszenia, webhooki, DM, voice, komendy, presence i uploady.
• Gateway: połączenie asyncio, IDENTIFY z botToken, heartbeat i reconnect z backoffem.
• Asynchroniczne handlery eventów: READY, MESSAGE_CREATE, MESSAGE_UPDATE, MESSAGE_DELETE, GUILD_CREATE i dowolne inne eventy.
• EmbedBuilder z limitami zgodnymi z API Nestra.
• CommandHandler dla komend prefiksowych z aliasami, pomocą i obsługą błędów.
• Rate limit queue czytająca nagłówki X-RateLimit-* i Retry-After.
• Zero zależności zewnętrznych w runtime. Wymagany jest tylko Python 3.11+.

Instalacja

Z katalogu repozytorium:

python -m pip install -e integrations/nestra-py

Albo bez instalacji, przez PYTHONPATH:

$env:PYTHONPATH = "integrations/nestra-py"
python examples/nestra-py-bot/bot.py

Wymagania: Python 3.11 lub nowszy. SDK nie wymaga websockets, aiohttp ani requests. Testy deweloperskie można uruchomić przez pytest albo wbudowany unittest.

Szybki start (REST)

import asyncio

from nestra import EmbedBuilder, NestraREST


async def main() -> None:
    rest = NestraREST.from_env()
    me = await rest.get_current_user()
    print(f"Zalogowany bot: {me.username}")

    guilds = await rest.get_guilds()
    if not guilds:
        print("Bot nie jest członkiem żadnego serwera.")
        return

    detail = await rest.get_guild(guilds[0].id)
    channel = detail.channel_by_name("ogolny")
    if channel is None:
        print("Brak kanału ogolny.")
        return

    embed = (
        EmbedBuilder()
        .set_title("Nestra-PY")
        .set_description("Wiadomość wysłana przez SDK")
        .set_color("blurple")
        .set_footer("nestra.gg")
    )
    await rest.send_message(channel.id, embeds=embed)


asyncio.run(main())

NestraREST.from_env() czyta BOT_TOKEN oraz opcjonalny NESTRA_API_URL (domyślnie http://localhost:4000). Token bota ma prefiks nestra_bot_ i jest pokazywany w panelu Developer tylko raz. Bot musi być członkiem serwera: dodaje go właściciel lub administrator przez panel albo bot dołącza kodem zaproszenia (join_guild).

Szybki start (Gateway)

import asyncio

from nestra import GatewayClient, NestraREST


async def main() -> None:
    rest = NestraREST.from_env()
    bot = GatewayClient.from_env(rest=rest)

    @bot.event("READY")
    async def on_ready(data):
        print(f"Połączono jako {bot.user.name if bot.user else '?'}")

    @bot.event("MESSAGE_CREATE")
    async def on_message(data):
        message = data["message"]
        if message["author"] and message["content"] == "!ping":
            await rest.send_message(message["channelId"], "pong")

    await bot.start()
    await bot.wait_until_ready(timeout=30)
    await bot.wait_until_closed()


asyncio.run(main())

GatewayClient.from_env() czyta BOT_TOKEN, opcjonalny NESTRA_GATEWAY_URL (domyślnie ws://localhost:4001/gateway) oraz NESTRA_GATEWAY_ORIGIN (domyślnie http://localhost:5174). Gateway Nestra wymaga nagłówka Origin z listy CORS, dlatego bot nie jest przeglądarką, ale nadal wysyła ten nagłówek.

Eventy

Handlery rejestruje się dekoratorem @bot.event("NAZWA"). Handler może być funkcją synchroniczną albo async. SDK wywołuje je po kolei, więc wiadomości zachowują kolejność. Błąd handlera trafia do @bot.on_error i nie zamyka połączenia.

@bot.on_error
async def on_error(error, event_name, data):
    print(f"Błąd {event_name}: {error}")

GUILD_CREATE jest obsługiwany jak każdy inny event. Obecny Gateway wysyła gildie w READY i eventy GUILD_UPDATE/GUILD_DELETE; rejestracja GUILD_CREATE pozwala odebrać ten event, gdy pojawi się w protokole.

Reconnect

Klient sam wysyła heartbeat na podstawie HELLO. Po zerwaniu połączenia bot wysyła ponownie IDENTIFY, bo Gateway nie obsługuje RESUME dla tokenów botów. Sesje użytkownika (token_type="Bearer") mogą użyć RESUME. Opóźnienia rosną wykładniczo (ReconnectBackoff) i mają jitter. Zamknięcia 4001 (błędny token) oraz 4002 (błędny payload) są traktowane jako trwałe, bez reconnectu.

Komendy prefiksowe

from nestra import CommandContext, CommandHandler

commands = CommandHandler(prefix="!", rest=rest)


@commands.command("ping", description="Odpowiada pongiem")
async def ping(ctx: CommandContext) -> None:
    await ctx.reply("pong")


@bot.event("MESSAGE_CREATE")
async def on_message(data):
    from nestra import Message

    await commands.handle_message(Message.from_dict(data["message"]))

parse_command("!ping teraz", "!") zwraca ParsedCommand(name="ping", args="teraz", argv=["teraz"]). Parser obsługuje cudzysłowy przez shlex i nie wywala się na niezamkniętym cudzysłowie.

Embedy

EmbedBuilder waliduje limity API: 256 znaków tytułu, 4096 opisu, 25 pól, 6000 znaków łącznie. Kolory można podawać jako int, hex (#5865F2) albo nazwę (blurple, green, red).

embed = (
    EmbedBuilder()
    .set_title("Status")
    .set_description("Wszystkie systemy działają")
    .set_color("#57F287")
    .add_field("API", "ok", inline=True)
    .add_field("Gateway", "ok", inline=True)
    .set_timestamp()
)

Do API można przekazać pojedynczy builder, listę builderów albo zwykłe dict-y. SDK wysyła wyłącznie adresy http/https w polach URL.

API klienta

Metoda	Opis
get_guilds()	lista gildii bota
get_guild(guild_id)	szczegóły gildii, kanały i role
get_guild_channels(guild_id)	kanały gildii
get_members(guild_id, limit=50)	członkowie gildii (do 1000)
edit_guild(guild_id, ...)	edycja nazwy/opisu/koloru gildii
kick_member(guild_id, user_id, reason=...)	wyrzucenie członka
ban_member(...) / unban_member(...)	ban i unban
get_bans(guild_id)	lista banów
timeout_member(...)	timeout członka
add_role(...) / remove_role(...)	role członka
join_guild(invite_code)	dodanie bota przez zaproszenie, zwraca ID gildii
leave_guild(guild_id)	opuszczenie gildii
create_invite(...)	zaproszenie do kanału
get_invites(guild_id) / delete_invite(code)	lista i usuwanie zaproszeń
create_channel(guild_id, name, ...)	nowy kanał
update_channel(guild_id, channel_id, ...)	edycja kanału
delete_channel(guild_id, channel_id)	usunięcie kanału
get_messages(channel_id, limit=50, before=None)	historia wiadomości
send_message(channel_id, content, embeds=..., reply_to_id=...)	wysyłka wiadomości
edit_message(channel_id, message_id, content=..., embeds=...)	edycja treści i/lub embedów
edit_embed_message(channel_id, message_id, embeds)	edycja z samym embedem
delete_message(channel_id, message_id)	usunięcie wiadomości bota
add_reaction(...) / remove_reaction(...)	reakcje bota
get_current_user()	konto bota (/api/v1/bot/me)
get_user(user_id)	profil użytkownika
set_presence(status, ...)	presence bota
open_dm(user_id) / send_dm(user_id, ...)	kanał DM i wiadomość
join_voice(guild_id, channel_id) / leave_voice(guild_id)	token LiveKit i wyjście
get_commands() / register_commands(...) / delete_command(id)	komendy aplikacji
get_webhooks(guild_id)	webhooki gildii
create_webhook(...)	nowy webhook
update_webhook(...) / delete_webhook(guild_id, webhook_id)	edycja i usunięcie
execute_webhook(webhook_id, token, ...)	wysyłka przez webhook, bez auth
upload_attachment(channel_id, filename, data)	upload pliku
send_file(channel_id, filename, data, content=None)	upload i wysyłka wiadomości

Wszystkie metody bota korzystają z namespace /api/v1/bot/... (1.2.2). Kontrakt rozszerza subagent bot-api równolegle; jeśli API zwróci 404 dla nowej trasy, SDK zgłosi czytelny NestraAPIError, a pozostałe metody działają.

Wszystkie metody REST są asynchroniczne. NestraREST.request() pozwala wywołać dowolny endpoint API bez oczekiwania na dodanie metody do SDK.

Rate limity i błędy

Każde żądanie przechodzi przez RateLimitQueue. Po odpowiedzi 429 klient czeka Retry-After (albo wartość z JSON) i ponawia żądanie, maksymalnie max_retries razy. Błędy 5xx i problemy sieciowe mają krótki backoff.

Błędy API są zgłaszane jako NestraAPIError:

from nestra import NestraAPIError

try:
    await rest.send_message(channel_id, "test")
except NestraAPIError as error:
    print(error.status, error.code, error.message)

error.status == 0 oznacza błąd sieci albo niepoprawną odpowiedź JSON.

Bezpieczeństwo tokenu

• Token bota to sekret. Trzymaj go w zmiennej BOT_TOKEN, nie w kodzie ani w repozytorium.
• SDK nie loguje tokenu. repr(NestraREST(...)) nie zawiera sekretu.
• Nie wysyłaj tokenu do przeglądarki ani do kodu klienta WWW.
• W produkcji używaj https:// i wss://; połączenie z tokenem przez http/ws jest dopuszczalne tylko lokalnie.
• Przy wycieku tokenu unieważnij klucz w panelu deweloperskim i wygeneruj nowy.
• Bot powinien mieć tylko uprawnienia potrzebne do działania. Token nie zastępuje kontroli uprawnień po stronie API.

Zgodność z bot auth

Bot auth jest wdrożony w API i Gatewayu:

• REST: Authorization: Bot <token> i trasy /api/v1/bot/...,
• Gateway: IDENTIFY przyjmuje dokładnie jedno pole, botToken albo token; SDK wysyła botToken dla tokenu bota i token dla sesji użytkownika,
• READY bota zawiera bot: true, pustą obecność i listę gildii bota.

Nie mieszaj tokenu bota i JWT użytkownika w jednym kliencie. token_type="Bearer" służy do sesji użytkownika, nie do kluczy nstr_ ani tokenów nestra_bot_.

Testy

Z katalogu pakietu:

python -m unittest discover -s tests -v
python -m compileall -q nestra

Testy jednostkowe nie wymagają uruchomionego API. Test dymny uruchamia się tylko z NESTRA_SMOKE=1 i prawidłowym BOT_TOKEN:

$env:NESTRA_SMOKE = "1"
python -m unittest tests.test_smoke -v

Struktura

integrations/nestra-py/
  nestra/
    __init__.py
    rest.py       # REST, rate limit, błędy
    gateway.py    # Gateway, heartbeat, reconnect, eventy
    embed.py      # EmbedBuilder
    commands.py   # CommandHandler
    types.py      # modele DTO
    _ws.py        # minimalny klient WebSocket

Przykładowy bot znajduje się w examples/nestra-py-bot w repozytorium Nestra.