Примеры ботов

Python

Простой бот, который отвечает на слэш-команду /ping:

import asyncio
import json
import websockets

BOT_TOKEN = "your-bot-token"
SERVER_URL = "ws://your-server:8082/ws/bot"


async def main():
    uri = f"{SERVER_URL}?token={BOT_TOKEN}"

    async with websockets.connect(uri) as ws:
        print("Bot connected")

        async for message in ws:
            event = json.loads(message)
            print(f"Event: {event['type']}")

            if event["type"] == "SlashCommand" and event["command"] == "ping":
                print(f"Ping from channel {event['channelId']}")

            elif event["type"] == "MessageCreated":
                print(f"New message: {event['content']}")

            elif event["type"] == "MemberJoined":
                print(f"User {event['userId']} joined")


asyncio.run(main())

Установка зависимостей

pip install websockets

JavaScript (Node.js)

const WebSocket = require("ws");

const BOT_TOKEN = "your-bot-token";
const SERVER_URL = "ws://your-server:8082/ws/bot";

const ws = new WebSocket(`${SERVER_URL}?token=${BOT_TOKEN}`);

ws.on("open", () => {
  console.log("Bot connected");
});

ws.on("message", (data) => {
  const event = JSON.parse(data);
  console.log(`Event: ${event.type}`);

  switch (event.type) {
    case "SlashCommand":
      console.log(`Command /${event.command} with args: ${event.args}`);
      break;
    case "MessageCreated":
      console.log(`New message: ${event.content}`);
      break;
    case "MemberJoined":
      console.log(`User ${event.userId} joined channel ${event.channelId}`);
      break;
  }
});

ws.on("close", (code, reason) => {
  console.log(`Disconnected: ${code} ${reason}`);
});

Установка зависимостей

npm install ws

Kotlin

import io.ktor.client.*
import io.ktor.client.plugins.websocket.*
import io.ktor.websocket.*
import kotlinx.serialization.json.*

val BOT_TOKEN = "your-bot-token"
val SERVER_URL = "ws://your-server:8082/ws/bot"

suspend fun main() {
    val client = HttpClient { install(WebSockets) }

    client.webSocket("$SERVER_URL?token=$BOT_TOKEN") {
        println("Bot connected")

        for (frame in incoming) {
            if (frame is Frame.Text) {
                val event = Json.parseToJsonElement(frame.readText()).jsonObject
                val type = event["type"]?.jsonPrimitive?.content

                when (type) {
                    "SlashCommand" -> println("Command /${event["command"]?.jsonPrimitive?.content}")
                    "MessageCreated" -> println("New message: ${event["content"]?.jsonPrimitive?.content}")
                    "MemberJoined" -> println("User ${event["userId"]?.jsonPrimitive?.content} joined")
                }
            }
        }
    }
}

Зависимости (build.gradle.kts)

dependencies {
    implementation("io.ktor:ktor-client-cio:3.1.1")
    implementation("io.ktor:ktor-client-websockets:3.1.1")
    implementation("org.jetbrains.kotlinx:kotlinx-serialization-json:1.8.1")
}

Автопереподключение

В продакшене рекомендуется добавить автоматическое переподключение:

import asyncio
import websockets

async def connect_with_retry(uri, max_delay=60):
    delay = 1
    while True:
        try:
            async with websockets.connect(uri) as ws:
                delay = 1  # сброс при успешном подключении
                async for message in ws:
                    handle_event(message)
        except (websockets.ConnectionClosed, ConnectionRefusedError):
            print(f"Reconnecting in {delay}s...")
            await asyncio.sleep(delay)
            delay = min(delay * 2, max_delay)