Twitch Channel Points in Minecraft: Key-Drops und Stream-Effekte für deinen Server
Wenn Spieler deinen Minecraft-Server streamen, kannst du ihre Zuschauer direkt ins Spiel holen. In diesem Guide baust du eine Twitch-Integration mit zwei Teilen: Ein Chat-Bot verlost während des Streams regelmäßig Keys an Zuschauer, die selbst auf dem Server spielen, und Kanalpunkte, Bits, Subs und Raids lösen Effekte beim Streamer im Spiel aus. Das Beispiel ist mein Cobblemon-Server CobbleBox, auf dem genau dieses Setup läuft. Der Code ist Python und kommt fast nur mit der Standard-Library aus. Einzige Ausnahme ist EventSub für die Kanalpunkte, dafür brauchst du das Paket websocket-client. Die Befehle landen über die Konsole des Game Panels (der Weboberfläche zur Server-Verwaltung) oder per RCON auf dem Server.
Kurz erklärt
- Zuschauer, die auf deinem Server spielen, gewinnen im Twitch-Chat mit
!claimSchlüssel, die automatisch im Inventar landen. - Streamer bekommen nach 30 Minuten eine Streamer-Kiste, und ihre Zuschauer lösen mit Kanalpunkten, Bits, Subs und Raids Feuerwerk und harmlose Effekte im Spiel aus.
- Gedacht für Betreiber eines Minecraft-Servers, auf dem Streamer spielen. Kanalpunkte gehen nur bei Affiliate- oder Partner-Kanälen.
- Zum Nachbauen brauchst du etwas Python und Server-Wissen. Dazu kommen ein Linux-Server für den Bot, ein Game Panel mit API (hier Pelican) und RCON am Minecraft-Server. Eine Website zum Verbinden der Twitch-Konten ist optional, für den Anfang reicht eine JSON-Datei. Ein fertiges Plugin gibt es nicht.
Was Zuschauer und Streamer davon haben
Auf CobbleBox läuft das so: Streamt ein verbundener Streamer Minecraft, droppt der Bot alle 45 bis 60 Minuten einen Schlüssel in den Chat. Zuschauer schreiben innerhalb von zwei Minuten !claim, danach zieht der Bot einen Gewinner. Meistens ist es ein Poké-Schlüssel, mit 10 Prozent Chance ein Pokémon-Schlüssel. Der Gewinn landet im Inventar, sobald der Spieler das nächste Mal online ist. Dazu beantwortet der Bot Befehle wie !stats, !potd (Pokémon des Tages) und !missionen aus den Daten, die die Server-Website ohnehin erzeugt.
Der Streamer selbst bekommt nach 30 Minuten Stream eine Streamer-Kiste. Bits, Subs und Raids zünden ein Partikel-Feuerwerk über seinem Charakter, Kanalpunkte lösen kurze Effekte wie Übelkeit, Blindheit oder Schweben aus. Das ist alles kosmetisch oder harmlos, Items gibt es über Bits oder Kanalpunkte für niemanden.
Die Architektur im Überblick
Das Setup besteht aus vier Bausteinen, die über JSON-Dateien (einfache Datendateien) miteinander reden:
- Die Website, auf der Spieler ihr Twitch-Konto mit ihrem Minecraft-Account verbinden. Das Ergebnis ist eine Datei, die Twitch-Login auf Spielername und UUID abbildet.
- Der Chat-Bot als systemd-Service (ein dauerhaft laufender Linux-Dienst). Er hängt per IRC (dem Protokoll hinter dem Twitch-Chat) verschlüsselt über TLS in den Chats. Dort prüft er, wer live ist, macht die Drops und beantwortet Befehle.
- Pro Streamer ein EventSub-Thread im Bot-Prozess. EventSub ist der Benachrichtigungsdienst von Twitch, ein Thread ein parallel laufender Nebenstrang im Programm. Er hört per WebSocket (einer dauerhaft offenen Verbindung) auf Kanalpunkt-Einlösungen.
- Die Panel-API, über die der Bot Konsolenbefehle an den Server schickt. Eine API ist eine Schnittstelle für Programme. Dazu kommt eine Reward-Queue (Warteschlange) für Gewinne an Spieler, die gerade offline sind.
Welche Kanäle der Bot betritt, stellst du auf einer Admin-Seite ein. Der Bot liest die Liste jede Minute neu und betritt neue Kanäle mit JOIN, entfernte verlässt er mit PART, alles ohne Neustart.
Website, Admin-Seite und Reward-Queue sind auf CobbleBox eigener Code. Zum Nachbauen brauchst du sie nicht. In diesem Guide ersetzt du sie durch einfache Bausteine: eine Datei viewers.json statt der Website, eine Kanalliste in der Config statt der Admin-Seite und eine kleine Zustell-Schleife per RCON statt der Reward-Queue.
Vorbereitung: Twitch-App registrieren
Bevor du eine App anlegen kannst, muss auf deinem Twitch-Account die Zwei-Faktor-Authentifizierung (2FA) aktiv sein. Dann gehst du so vor:
- Öffne die Twitch Developer Console, wechsle zu „Applications“ und klick auf „Register Your Application“.
- Gib einen eindeutigen Namen ein. Als OAuth Redirect URL trägst du
http://localhostein und klickst auf „Add“. Der Device Code Flow nutzt die Adresse nicht, das Formular verlangt aber eine. - Wähle eine Kategorie, etwa „Chat Bot“, und als Client Type „Public“.
- Nach „Create“ öffnest du die App mit „Manage“ und kopierst die Client ID.
Ein Public Client hat kein Client Secret und darf nur den Device Code Flow nutzen. Das reicht für den Bot und auch für die Streamer-Logins in diesem Guide. Willst du später eine echte Website mit „Mit Twitch verbinden“-Button bauen, brauchst du dafür den Authorization Code Flow, und der geht nur mit einer Confidential App samt Client Secret. Dann braucht auch der Bot das Secret beim Erneuern der Tokens.
Für die Kanalpunkte gilt außerdem: Bot, Streamer-Login und eine eventuelle Website müssen dieselbe Client ID benutzen. Twitch lässt den Status einer Einlösung nur von der App ändern, die die Belohnung angelegt hat. Mit einer zweiten App kannst du die Belohnungen der ersten weder auf FULFILLED noch auf CANCELED setzen.
Die Helper für HTTP, Tokens und JSON
Die Code-Beispiele in diesem Guide stützen sich auf ein paar kleine Funktionen. Sie kommen an den Anfang deines Bot-Scripts, die Client ID liest der Bot aus einer Umgebungsvariable. access_token erneuert den Token automatisch, wenn er in weniger als fünf Minuten abläuft, und speichert dabei jedes Mal den neuen Refresh Token:
import json, os, random, re, socket, ssl, struct, sys, time, queue, threading
import urllib.error, urllib.parse, urllib.request
CLIENT_ID = os.environ["TWITCH_CLIENT_ID"]
VIEWERS_FILE, DROPS_FILE, DELIVERED_FILE = "viewers.json", "drops.json", "delivered.json"
def post_form(url, fields): # Formular-POST, gibt (Status, JSON) zurück
data = urllib.parse.urlencode(fields).encode()
try:
with urllib.request.urlopen(urllib.request.Request(url, data=data), timeout=15) as r:
return r.status, json.load(r)
except urllib.error.HTTPError as e:
return e.code, json.load(e)
def http(method, url, token, client_id, body=None): # Helix-Aufruf mit JSON
req = urllib.request.Request(url, method=method,
data=None if body is None else json.dumps(body).encode(),
headers={"Authorization": f"Bearer {token}", "Client-Id": client_id,
"Content-Type": "application/json"})
with urllib.request.urlopen(req, timeout=15) as r:
raw = r.read()
return json.loads(raw) if raw else {}
def read_json(path, default):
try:
with open(path, encoding="utf-8") as f:
return json.load(f)
except FileNotFoundError:
return default
def write_json(path, data): # atomar schreiben
with open(path + ".tmp", "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False)
os.replace(path + ".tmp", path)
def save_token(t, path="bot_token.json"):
t["expires_at"] = time.time() + t.get("expires_in", 0)
write_json(path, t)
os.chmod(path, 0o600) # nur für deinen User lesbar
def access_token(path="bot_token.json"):
t = read_json(path, None)
if t["expires_at"] - time.time() < 300:
status, new = post_form("https://id.twitch.tv/oauth2/token", {
"client_id": CLIENT_ID, "grant_type": "refresh_token",
"refresh_token": t["refresh_token"]}) # Public Client: ohne Secret
if status != 200:
raise RuntimeError(f"Token-Refresh fehlgeschlagen, bitte neu einloggen: {new}")
save_token(new, path) # neuen Refresh Token immer speichern
t = new
return t["access_token"]Bei einem Public Client läuft jeder Refresh Token 30 Tage nach seiner Ausstellung ab, und ein benutzter Refresh Token gilt danach nicht mehr. Deshalb speichert access_token nach jedem Refresh den neuen. Läuft der Bot mehr als 30 Tage gar nicht, musst du den Login aus Schritt 2 wiederholen.
Schritt 1: Twitch-Konto auf der Website verbinden
Ohne Verknüpfung weiß der Bot nicht, welcher Zuschauer welcher Spieler ist. Auf CobbleBox hat jeder Spieler eine Trainerseite mit dem Button „Mit Twitch verbinden“. Danach steht in einer Datei, welcher Twitch-Login zu welchem Minecraft-Namen und welcher UUID gehört. Wer beim Drop !claim schreibt, ohne verbunden zu sein, bekommt vom Bot eine kurze Antwort mit dem Link zur Trainerseite.
Streamer verbinden sich zusätzlich für die Kanalpunkte. Dafür meldet sich der Streamer mit den Scopes channel:read:redemptions und channel:manage:redemptions an. Scopes sind Berechtigungen, die die App anfragt. Der Webserver speichert die Tokens (Zugangsschlüssel) lokal. Die gleiche Verknüpfung nutzt der Bot auch, um den Streamer im Spiel zu finden: Effekte gehen an den Minecraft-Namen, der zu diesem Twitch-Kanal gehört.
So geht es ohne eigene Website
Für den Start pflegst du die Zuordnung von Hand in viewers.json. Schlüssel ist der Twitch-Login in Kleinbuchstaben, dazu kommen Minecraft-Name und UUID:
{
"zuschauer_login": {"name": "Spielername", "uuid": "0f1e2d3c-4b5a-6978-8796-a5b4c3d2e1f0"}
}Die UUID eines Spielers, der schon einmal auf deinem Server war, steht in der Datei usercache.json im Server-Verzeichnis, als Eintrag mit name und uuid. Zuschauer schicken dir ihren Twitch-Namen und Minecraft-Namen, etwa per Discord, und du trägst sie ein. Der Bot liest die Datei bei jedem Drop neu, ein Neustart ist also nicht nötig.
Streamer verbinden sich für die Kanalpunkte mit dem gleichen Device Code Flow wie der Bot in Schritt 2, nur mit den Scopes channel:read:redemptions channel:manage:redemptions und einer eigenen Token-Datei. Du startest den Login, schickst dem Streamer Link und Code, er bestätigt mit seinem Twitch-Account, und der Token liegt danach auf deinem Server. Den Minecraft-Namen des Streamers trägst du genauso in viewers.json ein.
Schritt 2: Der Bot und der Device Code Flow
Du legst auf dev.twitch.tv eine App an und erstellst einen eigenen Twitch-Account für den Bot. Als Public Client braucht die App kein Client Secret (App-Passwort). Deshalb passt der Device Code Flow (Login per Code-Eingabe auf einer Webseite) gut zu einem Bot auf einem Server ohne Browser. Du startest das Script einmal mit --login, öffnest den angezeigten Link, meldest dich als Bot-Account an und gibst den Code ein. Das Script unten holt den Login-Code, zeigt ihn an und fragt regelmäßig nach, bis Twitch den Token herausgibt.
SCOPES = "chat:read chat:edit moderator:read:followers"
def login(client_id, scopes=SCOPES, path="bot_token.json"):
status, d = post_form("https://id.twitch.tv/oauth2/device",
{"client_id": client_id, "scopes": scopes})
if status != 200:
sys.exit(f"Device-Login fehlgeschlagen: {d}")
print(f"Öffne {d['verification_uri']} und gib {d['user_code']} ein")
deadline = time.time() + d.get("expires_in", 1800)
while time.time() < deadline:
time.sleep(d.get("interval", 5))
status, t = post_form("https://id.twitch.tv/oauth2/token", {
"client_id": client_id, "scopes": scopes,
"device_code": d["device_code"],
"grant_type": "urn:ietf:params:oauth:grant-type:device_code"})
if status == 200:
save_token(t, path) # access, refresh, expires (Datei mit chmod 600)
return
if t.get("message") not in ("authorization_pending", "slow_down"):
sys.exit(f"Login fehlgeschlagen: {t}")
sys.exit("Code abgelaufen")
if "--login" in sys.argv: # python3 bot.py --login
login(CLIENT_ID); sys.exit()
if "--login-streamer" in sys.argv: # python3 bot.py --login-streamer kanalname
name = sys.argv[-1].lower()
login(CLIENT_ID, "channel:read:redemptions channel:manage:redemptions",
f"streamer_{name}.json")
sys.exit()Läuft der Access Token in weniger als fünf Minuten ab, holt der Bot mit dem Refresh Token (einem Schlüssel nur zum Erneuern) einen neuen, auch das ohne Secret. Danach verbindet sich der Bot per TLS mit dem Twitch-IRC und schickt PASS oauth:… und NICK. Außerdem fordert er die Capabilities twitch.tv/tags und twitch.tv/commands an. Die Tags brauchst du, weil Bits in den Tags einer Chat-Nachricht stehen. Subs und Raids kommen als USERNOTICE mit einer msg-id. Bricht die Verbindung ab oder schickt Twitch ein RECONNECT, verbindet sich der Bot neu. Die Wartezeit dafür verdoppelt sich jedes Mal, von 5 bis maximal 300 Sekunden.
Die Verbindung läuft zu irc.chat.twitch.tv auf Port 6697 mit TLS. Zuerst kommt PASS oauth:<token>, dann NICK mit dem Login des Bot-Accounts in Kleinbuchstaben, dann CAP REQ und pro Kanal ein JOIN #kanal. Regelmäßig schickt Twitch ein PING :tmi.twitch.tv, darauf antwortest du mit PONG :tmi.twitch.tv, sonst trennt Twitch die Verbindung. Die Kanalliste steht in diesem Grundgerüst in einer config.json als {"channels": ["streamer1", "streamer2"]}. Damit der Bot nebenbei Drops und Live-Checks erledigen kann, wartet er höchstens eine Sekunde auf neue Daten und ruft unabhängig vom Chat-Verkehr etwa einmal pro Sekunde tick aus Schritt 3 auf:
BOT_LOGIN = os.environ["TWITCH_BOT_LOGIN"]
CHANNELS = [c.lower() for c in read_json("config.json", {}).get("channels", [])]
PRIVMSG = re.compile(r"^(?:@(\S+) )?:(\w+)!\S+ PRIVMSG #(\w+) :(.*)$")
USERSTATE = re.compile(r"^@(\S+) :tmi\.twitch\.tv USERSTATE #(\w+)")
class Chat:
def __init__(self, sock):
self.sock = sock
def send(self, line):
self.sock.sendall((line + "\r\n").encode())
def say(self, channel, text):
self.send(f"PRIVMSG #{channel} :{text}")
def irc_connect():
raw = socket.create_connection(("irc.chat.twitch.tv", 6697), timeout=15)
chat = Chat(ssl.create_default_context().wrap_socket(raw, server_hostname="irc.chat.twitch.tv"))
chat.send(f"PASS oauth:{access_token()}")
chat.send(f"NICK {BOT_LOGIN.lower()}")
chat.send("CAP REQ :twitch.tv/tags twitch.tv/commands")
for c in CHANNELS:
chat.send(f"JOIN #{c}")
chat.sock.settimeout(1)
return chat
def run():
delay = 5
while True:
try:
chat, buf, last_tick = irc_connect(), b"", 0
while True:
try:
data = chat.sock.recv(4096)
if not data:
raise ConnectionError("Verbindung geschlossen")
except socket.timeout:
data = None # keine neuen Daten
if data:
delay, buf = 5, buf + data
*lines, buf = buf.split(b"\r\n")
for line in lines:
line = line.decode("utf-8", "replace")
if line.startswith("PING"):
chat.send("PONG" + line[4:])
elif line.endswith(" RECONNECT"):
raise ConnectionError("Twitch verlangt RECONNECT")
else:
handle_line(chat, line)
if time.time() - last_tick >= 1: # tick läuft auch bei vollem Chat
last_tick = time.time()
try:
tick(chat) # Schritt 3
except Exception as e:
print("tick-Fehler:", e)
except OSError as e: # auch ConnectionError
print(f"IRC getrennt ({e}), neuer Versuch in {delay} s")
time.sleep(delay)
delay = min(delay * 2, 300)
def handle_line(chat, line):
m = USERSTATE.match(line)
if m: # ist der Bot in diesem Kanal Mod?
state["mod"][m[2]] = "mod=1" in m[1].split(";")
m = PRIVMSG.match(line)
if not m:
return
tags = dict(kv.split("=", 1) for kv in (m[1] or "").split(";") if "=" in kv)
login, channel, text = m[2], m[3], m[4].strip()
if text.lower() == "!claim" and channel in state["claims"]:
state["claims"][channel].append(login)
# tags.get("bits") enthält Bits; USERNOTICE (Subs, Raids) wertest du hier ebenfalls ausSelbst zu schreiben sind in diesem Grundgerüst die Auswertung von Bits und USERNOTICE für die Effekte aus Schritt 6 und die weiteren Chat-Befehle wie !stats. Twitch empfiehlt für neue Chat-Bots inzwischen EventSub und die Helix-API. IRC funktioniert weiterhin und ist für den Einstieg einfacher.
Schritt 3: Live-Check und Key-Drops
Drops gibt es nur, solange ein Kanal live ist. Deshalb fragt der Bot alle zwei Minuten mit dem Bot-Token den Stream-Status ab. Dafür nutzt er den Endpunkt (eine Abfrage-Adresse) Get Streams der offiziellen Twitch-API Helix. Fällt der aus, springt der Dienst decapi als Fallback ein. Helix ist auf 800 Punkte pro Minute begrenzt (Rate Limit). Das reicht für ein paar Kanäle alle zwei Minuten locker. Live zählt ein Kanal nur, wenn als Spiel Minecraft eingetragen ist.
Für jeden Live-Kanal legt der Bot einen zufälligen Zeitpunkt in 45 bis 60 Minuten fest. Ist er erreicht, kündigt der Bot den Drop an und sammelt zwei Minuten lang alle !claim. Gewinnen kann nur, wer
- sein Twitch-Konto verbunden hat,
- dem Kanal folgt (das prüft der Bot über die Helix-Followers-API, dafür muss er Mod (Moderator) im Kanal sein),
- nicht der Streamer selbst ist
- und heute noch keine fünf Drops gewonnen hat.
Damit der Chat nicht zugespammt wird, bestätigt der Bot neue Claims gesammelt, höchstens eine Nachricht alle zehn Sekunden pro Kanal. Dabei sagt er auch, wer warum nicht gewinnen kann. Der laufende Drop wird in eine State-Datei (Zwischenspeicher) geschrieben. Ein Neustart des Bots verliert also keine Claims. Admins können mit !drop zusätzlich manuell einen bestimmten Schlüssel droppen, höchstens alle zehn Minuten pro Kanal.
Ein einfaches Grundgerüst dafür sieht so aus. check_live fragt Get Streams für bis zu 100 Kanäle auf einmal ab und zählt nur Streams mit dem Spiel Minecraft. live_channels mit dem Debounce findest du weiter unten bei den typischen Problemen. Die gesammelten Bestätigungen, die State-Datei, !drop und der decapi-Fallback fehlen hier, die kannst du später ergänzen:
state = {"next_check": 0, "live": set(), "misses": {}, "drop_at": {},
"claims": {}, "until": {}, "mod": {}, "next_delivery": 0}
def check_live(chans):
try:
q = "&".join("user_login=" + c for c in chans)
data = http("GET", f"https://api.twitch.tv/helix/streams?{q}",
access_token(), CLIENT_ID)["data"]
except Exception:
return {c: None for c in chans} # keine Antwort zählt nicht als offline
playing = {s["user_login"]: s.get("game_name") == "Minecraft" for s in data}
return {c: playing.get(c, False) for c in chans}
def tick(chat):
now = time.time()
if now >= state["next_check"]: # alle zwei Minuten
state["next_check"] = now + 120
state["live"] = live_channels(CHANNELS, state["live"], state["misses"])
for c in list(state["drop_at"]):
if c not in state["live"]:
del state["drop_at"][c] # offline: Timer verwerfen
for c in state["live"]:
state["drop_at"].setdefault(c, now + random.randint(45 * 60, 60 * 60))
for c, t in list(state["drop_at"].items()):
if now >= t and c not in state["until"]:
chat.say(c, "Key-Drop! Schreib jetzt !claim, du hast zwei Minuten.")
state["claims"][c], state["until"][c] = [], now + 120
del state["drop_at"][c]
for c, t in list(state["until"].items()):
if now >= t:
del state["until"][c]
resolve_drop(chat, c, state["claims"].pop(c), now) # Schritt 4
if now >= state["next_delivery"]: # Gewinne zustellen, Schritt 4
state["next_delivery"] = now + 120
deliver()
process_redemptions() # Kanalpunkte, Schritt 6claim_problem gibt einen kurzen Grund zurück, warum jemand nicht gewinnen kann, oder None. Für den Follower-Check braucht der Bot die Twitch-User-IDs, die er über Get Users holt. Ist der Bot im Kanal kein Mod (das meldet Twitch per USERSTATE beim JOIN), liefert die Followers-API keine Einzeldaten, deshalb überspringt der Code den Check dann und blockiert niemanden:
def wins_today():
midnight = time.mktime(time.localtime()[:3] + (0, 0, 0, 0, 0, -1))
won = {}
for w in read_json(DROPS_FILE, []):
if w["t"] >= midnight:
won[w["login"]] = won.get(w["login"], 0) + 1
return won
def follows(login, channel):
if not state["mod"].get(channel):
return True
try:
users = http("GET", f"https://api.twitch.tv/helix/users?login={login}&login={channel}",
access_token(), CLIENT_ID)["data"]
ids = {u["login"]: u["id"] for u in users}
r = http("GET", "https://api.twitch.tv/helix/channels/followers"
f"?broadcaster_id={ids[channel]}&user_id={ids[login]}",
access_token(), CLIENT_ID)
return bool(r["data"])
except Exception:
return True # Abfrage kaputt: niemanden blockieren
def claim_problem(login, channel, viewers, won):
if login == channel:
return "Streamer können nicht gewinnen"
if login not in viewers:
return "Twitch-Konto nicht verbunden"
if won.get(login, 0) >= 5:
return "heute schon fünf Drops gewonnen"
if not follows(login, channel):
return "folgt dem Kanal nicht"
return NoneSchritt 4: Gewinne über die Reward-Queue zustellen
Der Bot gibt Items nie selbst aus. Er hängt den Gewinn mit Twitch-Login, UUID, Spielername, Schlüssel-Typ und Zeitstempel an eine Datei an (die letzten 500 Einträge bleiben). Ein Timer (ein Linux-Wecker für Scripts) ruft alle zwei Minuten das Website-Script auf. Das liest diese Datei und übernimmt neue Gewinne in die Reward-Queue des Servers. Die Queue liefert direkt ins Inventar, sobald der Spieler online ist, egal ob der Gewinner gerade spielt oder nur zuschaut.
Der Code lost den Gewinner unter allen gültigen Claims aus, würfelt den Schlüssel-Typ und schreibt den Gewinn in die Datei.
def resolve_drop(chat, channel, claims, now):
viewers, won = read_json(VIEWERS_FILE, {}), wins_today()
claims = list(dict.fromkeys(claims)) # doppelte Claims raus
eligible = [u for u in claims
if not claim_problem(u, channel, viewers, won)]
if not eligible:
chat.say(channel, "Diesmal konnte niemand gewinnen.")
return
login = random.choice(eligible)
v = viewers[login] # {"uuid": ..., "name": ...}
key = "pokemon_key" if random.random() < 0.1 else "poke_key"
wins = read_json(DROPS_FILE, [])
wins.append({"id": f"{int(now)}-{channel}", "login": login,
"uuid": v["uuid"], "name": v["name"],
"key": key, "t": int(now)})
write_json(DROPS_FILE, wins[-500:]) # Website-Script -> Reward-Queue
chat.say(channel, f"@{login} ({v['name']}) gewinnt! Kommt ins Inventar, "
"sobald du online bist.")Einfache Zustellung per RCON
Ohne eigene Reward-Queue übernimmt eine kleine Schleife im Bot die Zustellung. Sie fragt per RCON, wer online ist, gibt diesen Spielern ihre offenen Gewinne und merkt sich in delivered.json, was schon zugestellt ist. Wer offline ist, bekommt seinen Gewinn beim nächsten Durchlauf, sobald er wieder online ist. RCON schaltest du in der server.properties ein und startest den Server danach neu:
enable-rcon=true
rcon.port=25575
rcon.password=ein-langes-zufaelliges-passwortDen RCON-Port gibst du nicht ins Internet frei. Läuft der Bot auf einem anderen Rechner als der Server, erlaubst du den Port in der Firewall nur für die IP des Bots. Bei Pelican musst du den Port dem Server zusätzlich als Allocation zuweisen. Der RCON-Client braucht nur die Standard-Library:
RCON_HOST = os.environ.get("RCON_HOST", "127.0.0.1")
RCON_PORT = int(os.environ.get("RCON_PORT", "25575"))
RCON_PASSWORD = os.environ["RCON_PASSWORD"]
# Beispiel: anpassen an den Befehl deines Crate-Plugins oder Mods
GIVE = {"poke_key": "give {name} minecraft:diamond 1",
"pokemon_key": "give {name} minecraft:diamond 3"}
def _recv(s, n):
buf = b""
while len(buf) < n:
chunk = s.recv(n - len(buf))
if not chunk:
raise ConnectionError("RCON getrennt")
buf += chunk
return buf
def rcon(cmd):
with socket.create_connection((RCON_HOST, RCON_PORT), timeout=10) as s:
for req_id, kind, body in ((1, 3, RCON_PASSWORD), (2, 2, cmd)): # 3 = Login, 2 = Befehl
data = struct.pack("<ii", req_id, kind) + body.encode() + b"\x00\x00"
s.sendall(struct.pack("<i", len(data)) + data)
size = struct.unpack("<i", _recv(s, 4))[0]
resp_id, _ = struct.unpack("<ii", _recv(s, 8))
text = _recv(s, size - 8)[:-2].decode("utf-8", "replace")
if resp_id == -1:
raise PermissionError("RCON-Passwort falsch")
return text
def online_players():
text = rcon("list") # "There are 2 of a max of 20 players online: Alex, Steve"
names = text.split(":", 1)[1] if ":" in text else ""
return {n.strip() for n in names.split(",") if n.strip()}
def deliver():
done = set(read_json(DELIVERED_FILE, []))
online = online_players()
for w in read_json(DROPS_FILE, []):
if w["id"] in done or w["name"] not in online:
continue # schon zugestellt oder offline
rcon(GIVE[w["key"]].format(name=w["name"]))
done.add(w["id"])
write_json(DELIVERED_FILE, sorted(done)) # sofort merken, nie doppelt gebenServerspezifisch ist nur die Tabelle GIVE. Mit dem Vanilla-Befehl give kannst du alles testen. Für echte Schlüssel trägst du den Befehl deines Crate-Plugins oder Mods ein, so wie er auch in der Server-Konsole funktioniert. tick aus Schritt 3 ruft deliver alle zwei Minuten auf.
Schritt 5: Kanalpunkte per EventSub
Kanalpunkt-Einlösungen stehen nicht im IRC-Chat. Dafür nutzt du EventSub über WebSocket. Der Bot startet für jeden verbundenen Streamer einen eigenen Thread. Beim ersten Start legt der Thread elf Belohnungen im Kanal an, vom Feuerwerk für 500 Punkte bis zu Blindheit und Schweben für 5000 Punkte, jeweils mit einem globalen Cooldown (Sperrzeit nach einer Einlösung) von 60 bis 300 Sekunden. Das klappt nur bei Affiliate- oder Partner-Kanälen. Ist der Kanal kein Affiliate oder Partner, schreibt der Bot nur eine Meldung ins Log und läuft ohne Kanalpunkte weiter. Belohnungen, die der Streamer selbst angelegt hat, ordnet der Bot über Stichwörter im Titel zu, etwa „blind“ oder „levit“.
Die Belohnungen legt der Bot mit dem Token des Streamers über Create Custom Rewards an. Pro Kanal sind maximal 50 Belohnungen erlaubt, und der Titel darf höchstens 45 Zeichen haben. Die IDs der angelegten Belohnungen speicherst du, denn nur für diese darf deine App später den Status setzen. Preise und Cooldowns hier sind Beispielwerte:
REWARDS = [("Feuerwerk", 500, 60, "feuerwerk"), ("Übelkeit", 2000, 120, "uebelkeit"),
("Blindheit", 5000, 300, "blind"), ("Schweben", 5000, 300, "schweben")]
REDEMPTIONS = queue.Queue()
def create_rewards(broadcaster_id, token_file):
mine = {}
for title, cost, cooldown, effect in REWARDS:
r = http("POST", "https://api.twitch.tv/helix/channel_points/custom_rewards"
f"?broadcaster_id={broadcaster_id}", access_token(token_file), CLIENT_ID,
body={"title": title, "cost": cost, "is_global_cooldown_enabled": True,
"global_cooldown_seconds": cooldown})
mine[r["data"][0]["id"]] = effect
return mine # {reward_id: effekt}, als JSON speichern
def set_status(broadcaster_id, token_file, reward_id, redemption_id, status):
http("PATCH", "https://api.twitch.tv/helix/channel_points/custom_rewards/redemptions"
f"?id={redemption_id}&broadcaster_id={broadcaster_id}&reward_id={reward_id}",
access_token(token_file), CLIENT_ID, body={"status": status}) # FULFILLED oder CANCELED
KEYWORDS = {"blind": "blind", "levit": "schweben", "schweb": "schweben",
"übel": "uebelkeit", "feuerwerk": "feuerwerk"}
def effect_for_title(title): # selbst angelegte Belohnungen per Stichwort
t = title.lower()
return next((e for k, e in KEYWORDS.items() if k in t), None)Für EventSub brauchst du das Paket websocket-client (pip install websocket-client), denn die Standard-Library kann kein WebSocket. Der Bot verbindet sich mit wss://eventsub.wss.twitch.tv/ws?keepalive_timeout_seconds=30. Nach dem Verbinden schickt Twitch eine session_welcome mit einer Session-ID. Mit der abonnierst du innerhalb von 10 Sekunden das Event für Einlösungen, sonst schließt Twitch die Verbindung. Das Abo braucht einen User Access Token des Streamers, ein App-Token funktioniert bei WebSockets nicht. Danach kommen Einlösungen als notification.
Bei session_reconnect verbindest du dich sofort mit der mitgeschickten reconnect_url, so wie sie ist. Die Abos ziehen automatisch mit um, du abonnierst also nicht neu. Die alte Verbindung schließt du erst, wenn auf der neuen die session_welcome angekommen ist. Reißt die Verbindung dagegen einfach ab (kein Keepalive innerhalb der 30 Sekunden), baust du eine frische Verbindung auf und abonnierst neu. So sieht der Thread pro Streamer aus:
import websocket # pip install websocket-client
FRESH_URL = "wss://eventsub.wss.twitch.tv/ws?keepalive_timeout_seconds=30"
def eventsub(login, broadcaster_id, token_file, my_rewards):
url, subscribed, old = FRESH_URL, False, None
while True:
try:
ws = websocket.create_connection(url, timeout=40) # keepalive 30 s, also 40 s Timeout
while True:
msg = json.loads(ws.recv())
kind = msg["metadata"]["message_type"]
if kind == "session_welcome":
if old:
old.close()
old = None
if not subscribed: # nur bei frischer Verbindung, nie nach Reconnect
http("POST", "https://api.twitch.tv/helix/eventsub/subscriptions",
access_token(token_file), CLIENT_ID, body={
"type": "channel.channel_points_custom_reward_redemption.add",
"version": "1",
"condition": {"broadcaster_user_id": broadcaster_id},
"transport": {"method": "websocket",
"session_id": msg["payload"]["session"]["id"]}})
subscribed = True
elif kind == "session_reconnect":
url, old = msg["payload"]["session"]["reconnect_url"], ws
break # neue Verbindung, alte bleibt bis zum Welcome offen
elif kind == "notification":
ev = msg["payload"]["event"]
rid = ev["reward"]["id"]
effect = my_rewards.get(rid) or effect_for_title(ev["reward"]["title"])
if effect:
REDEMPTIONS.put((login, effect, ev["user_name"], ev["id"], rid))
except Exception as e: # Timeout, Abbruch, HTTP-Fehler
print(f"EventSub {login}: {e}")
url, subscribed = FRESH_URL, False
time.sleep(5)Beim Start holt der Bot für jeden Streamer die Broadcaster-ID (GET https://api.twitch.tv/helix/users ohne Parameter liefert den Besitzer des Tokens), legt beim ersten Mal die Belohnungen an und startet den Thread. Danach läuft die IRC-Schleife aus Schritt 2. Dieser Start-Block gehört ganz ans Ende von bot.py, hinter alle Funktionen aus diesem Guide, auch die aus Schritt 6 und den typischen Problemen:
BROADCASTERS = {} # kanal -> broadcaster_id
for name in CHANNELS:
token_file = f"streamer_{name}.json"
if not os.path.exists(token_file):
continue # Streamer hat sich nicht verbunden
try:
me = http("GET", "https://api.twitch.tv/helix/users", access_token(token_file), CLIENT_ID)
bid = me["data"][0]["id"]
rewards_file = f"rewards_{name}.json"
mine = read_json(rewards_file, None)
if mine is None:
mine = create_rewards(bid, token_file) # scheitert ohne Affiliate/Partner
write_json(rewards_file, mine)
BROADCASTERS[name] = bid
threading.Thread(target=eventsub, args=(name, bid, token_file, mine), daemon=True).start()
except Exception as e:
print(f"Kanalpunkte für {name} deaktiviert: {e}")
continue
run()Der Thread hängt die Einlösung in eine Queue, die Hauptschleife des Bots arbeitet sie ab. Danach setzt der Bot die Einlösung per Helix auf FULFILLED (erledigt). Manchmal ist der Streamer gerade nicht im Spiel, oder derselbe Trank-Effekt kam vor weniger als 20 Sekunden schon. Dann setzt der Bot sie auf CANCELED und der Zuschauer bekommt seine Punkte zurück. Das funktioniert nur bei Belohnungen, die die eigene App angelegt hat.
Schritt 6: Effekte über die Panel-Konsole auslösen
Bits, Subs, Raids und Kanalpunkte landen alle in derselben Effekt-Warteschlange. Pro Streamer feuert der Bot höchstens alle vier Sekunden einen Effekt, der Rest wartet. Die Stufen hängen am Anlass: unter 100 Bits ein kleines Feuerwerk, ab 100 Bits und bei Subs ein größeres, ab 500 Bits und bei Raids das große. Ab 1000 Bits, fünf verschenkten Subs oder einem Raid mit 20 Zuschauern bekommt der Streamer zusätzlich einen Titel auf den Bildschirm und alle Spieler eine Chatzeile.
Die Kanalpunkt-Effekte sind normale Trank-Effekte mit mindestens 30 Sekunden Dauer, nie tödlich. Schweben ist der einzige Sonderfall: sechs Sekunden Levitation, danach 30 Sekunden Federfall für eine weiche Landung. Das Feuerwerk ist eine Funktion aus einem Datapack (ein Paket eigener Befehle, das ohne Mods läuft) mit einem Makro-Parameter (Platzhalter) für den Spielernamen. Der Code ordnet den Belohnungen Trank-Effekte zu und schickt alles als Konsolenbefehl über die Client-API des Pelican Panels (des Game Panels auf CobbleBox):
POTIONS = {
"uebelkeit": [("nausea", 30, 0)],
"blind": [("blindness", 30, 0)],
"schweben": [("levitation", 6, 1), ("slow_falling", 30, 0)],
}
def console(cmd):
req = urllib.request.Request(
"https://panel.example.de/api/client/servers/<server-id>/command",
data=json.dumps({"command": cmd}).encode(), method="POST",
headers={"Authorization": "Bearer <panel-api-key>",
"Content-Type": "application/json", "Accept": "application/json"})
urllib.request.urlopen(req, timeout=10)
def fire_effect(player, effect, text):
if effect in POTIONS:
for eff, sec, lvl in POTIONS[effect]:
console(f"effect give {player} minecraft:{eff} {sec} {lvl} true")
console(f'function example:feuerwerk_klein {{player:"{player}"}}')
console(f'title {player} actionbar {{"text":"{text}","color":"light_purple"}}')Den Panel-API-Key erstellst du in Pelican über dein Profil (oben rechts) im Reiter „API Keys“ mit „Create API Key“. Das ist ein Client-Key, er darf genau das, was dein Panel-User darf. Unter „Allowed IPs“ kannst du ihn auf die IP des Bots beschränken. Die Server-ID steht in der Adresse, wenn du den Server im Panel öffnest: Bei https://panel.example.de/server/1a2b3c4d ist es 1a2b3c4d. Der Konsolenbefehl kommt nur an, solange der Server läuft.
Die Hauptschleife holt die Einlösungen aus der Queue, löst den Effekt aus und setzt danach den Status. Eine Minimalversion ohne den Vier-Sekunden-Abstand und die 20-Sekunden-Sperre für gleiche Tränke sieht so aus. tick aus Schritt 3 ruft sie schon auf, die Sperren ergänzt du selbst:
def process_redemptions():
viewers = read_json(VIEWERS_FILE, {})
while not REDEMPTIONS.empty():
login, effect, viewer, redemption_id, reward_id = REDEMPTIONS.get()
player = viewers.get(login, {}).get("name") # Minecraft-Name des Streamers
try:
online = player in online_players()
except Exception: # RCON nicht erreichbar
online = False
if online:
fire_effect(player, effect, f"{viewer} hat {effect} eingelöst")
if reward_id in read_json(f"rewards_{login}.json", {}): # nur eigene Belohnungen
set_status(BROADCASTERS[login], f"streamer_{login}.json", reward_id,
redemption_id, "FULFILLED" if online else "CANCELED")Das Feuerwerk selbst ist eine Funktion in einem Datapack. Für Minecraft 1.21 und 1.21.1 hat ein Datapack das Format 48, und Funktionen liegen im Ordner function (Einzahl). Leg im Weltordner (Standard: world, siehe level-name in der server.properties) diese zwei Dateien an:
world/datapacks/twitch_effekte/pack.mcmeta
world/datapacks/twitch_effekte/data/example/function/feuerwerk_klein.mcfunctionIn die pack.mcmeta kommt:
{"pack": {"pack_format": 48, "description": "Twitch-Effekte"}}Und in feuerwerk_klein.mcfunction zwei Makro-Zeilen. Zeilen mit $ am Anfang sind Makros, $(player) wird durch den Namen aus dem Aufruf function example:feuerwerk_klein {player:"Name"} ersetzt:
$execute at $(player) run particle minecraft:firework ~ ~2 ~ 0.5 0.5 0.5 0.1 80 force
$execute at $(player) run playsound minecraft:entity.firework_rocket.blast master @a ~ ~ ~ 1 1Danach lädst du das Pack mit reload in der Server-Konsole. Ob es aktiv ist, zeigt datapack list. Für die größeren Stufen legst du weitere Funktionen nach dem gleichen Muster an, etwa mit mehr Partikeln.
Über die Konsole bekommst du keine Antwort zurück. Für Effekte reicht das. Für Abfragen mit Rückgabewert nutzt du RCON (eine Fernsteuerung für die Server-Konsole). Denk außerdem an das Rate Limit des Panels: Auf CobbleBox liegt es bei rund 60 Schreibzugriffen pro Minute, deshalb der Abstand zwischen zwei Effekten.
📸 [Screenshot: Kanalpunkt-Belohnungen im Twitch-Kanal und der Effekt im Spiel, Lion fügt ein]
So prüfst du, ob es läuft
Für den ersten Test brauchst du einen Twitch-Kanal, den du selbst streamen kannst, einen zweiten Twitch-Account als Zuschauer und einen Minecraft-Account, der in viewers.json zu diesem Zuschauer gehört. Setz die Drop-Wartezeit vorübergehend herunter, zum Beispiel random.randint(60, 120) statt 45 bis 60 Minuten.
pip install websocket-clientausführen und die UmgebungsvariablenTWITCH_CLIENT_ID,TWITCH_BOT_LOGINundRCON_PASSWORDsetzen (export NAME="wert").python3 bot.py --loginausführen, Link öffnen, als Bot-Account anmelden und Code eingeben. Danach liegtbot_token.jsonneben dem Script.- Den Bot im Testkanal zum Mod machen (
/mod botnameim Chat) und den Kanal inconfig.jsoneintragen. Bot mitpython3 bot.pystarten. - Einen Teststream mit der Kategorie Minecraft starten, etwa aus OBS. Nach spätestens zwei Minuten erkennt der Live-Check den Stream.
- Wenn der Bot den Drop ankündigt, schreibt der zweite Account
!claim. Nach zwei Minuten nennt der Bot den Gewinner, und indrops.jsonsteht ein neuer Eintrag. - Mit dem Minecraft-Account auf den Server gehen. Spätestens nach zwei Minuten kommt das Item, und die ID steht in
delivered.json. - Für die Kanalpunkte (nur Affiliate oder Partner):
python3 bot.py --login-streamer kanalnameausführen, Bot neu starten und prüfen, ob die Belohnungen im Kanal auftauchen. Dann mit dem zweiten Account eine billige Belohnung einlösen, während der Streamer im Spiel ist.
Passiert beim Drop nichts, prüfe zuerst die Ausgabe des Bots. Ein HTTP Error 401 bei Helix heißt meistens, dass der Token zu einer anderen Client ID gehört als TWITCH_CLIENT_ID.
Typische Probleme
Drops kamen im Stream nie an
In einem Stream auf CobbleBox blieben die Drops aus, obwohl der Streamer Minecraft streamte. Damals lief der Live-Check nur über decapi, und decapi antwortet ab und zu nicht oder kurz falsch. Ein einziger negativer Check reichte, damit der Kanal als offline galt. Sobald ein Kanal offline ist, löscht der Bot den geplanten Drop-Zeitpunkt und würfelt beim nächsten positiven Check neu 45 bis 60 Minuten aus. Kommt so ein Aussetzer öfter als einmal pro Stunde, setzt sich der Timer immer wieder zurück und der Drop passiert nie.
Behoben habe ich das an zwei Stellen. Der Live-Check läuft jetzt primär über Helix Get Streams, decapi ist nur noch Fallback. Außerdem hat der Bot ein Debounce (einen Puffer gegen kurze Aussetzer). Ein Kanal, der live war, gilt erst nach zwei negativen Checks in Folge als offline. Gibt der Check gar keine Antwort, zählt das nicht als negativ, der Kanal bleibt einfach live. Der Code zählt dafür die negativen Checks pro Kanal mit:
OFFLINE_AFTER = 2
def live_channels(chans, was, misses):
status = check_live(chans) # {kanal: True | False | None (keine Antwort)}
live = set()
for c in chans:
if status.get(c):
misses.pop(c, None)
live.add(c)
elif c in was:
if status.get(c) is not None:
misses[c] = misses.get(c, 0) + 1
if misses.get(c, 0) < OFFLINE_AFTER:
live.add(c) # noch nicht offline, Drop-Timer bleibt
else:
misses.pop(c, None)
return liveDrops und Meldungen in Streams ohne Minecraft
Ein verbundener Streamer streamt nicht immer Minecraft. Zählt jeder Stream als live, droppt der Bot Keys für deinen Server auch während eines ganz anderen Spiels und postet Server-Meldungen in einen Chat, den das nicht interessiert. Deshalb prüft der Live-Check auf CobbleBox das eingetragene Spiel, und nur Minecraft-Streams zählen als live. Das gilt für die Drops genauso wie für die Server-Meldungen im Chat und die Live-Karte auf der Website.
Kleinere Stolperfallen
Der Follower-Check braucht den Scope moderator:read:followers und einen Bot, der im Kanal Mod ist. Klappt die Abfrage nicht, blockiert der Bot niemanden, damit ein fehlender Mod-Status nicht jeden Drop kaputtmacht. Bei einer Gift-Bomb (viele verschenkte Abos auf einmal) schickt Twitch eine Sammelmeldung und zusätzlich eine Meldung pro Abo. Die einzelnen subgift-Meldungen mit einer Community-Gift-ID überspringst du, sonst gibt es zehn Feuerwerke statt einem. Bei den Kanalpunkten lohnt sich ein Blick auf die Preise: Zuschauer sammeln grob 300 Punkte pro Stunde, deshalb habe ich die Preise auf CobbleBox am 01.10.2026 halbiert. Ein kleines Script schiebt die neuen Preise per Helix PATCH in alle Kanäle, in denen die Belohnungen schon existieren.
Fazit
Live in Aktion siehst du das Ganze auf CobbleBox unter 2mc.mapplebox.de: Twitch-Konto auf der Trainerseite verbinden, einem Streamer zuschauen und beim nächsten Drop !claim schreiben. Falls du deinen eigenen Server erst noch aufsetzen musst, findest du im Guide Cobblemon Server erstellen das komplette Setup mit dem Pelican Panel, über dessen Konsole auch der Bot seine Befehle schickt.
Hat dir der Guide geholfen? Dann spendier mir gern einen Kaffee auf Ko-fi. Damit bleibt MappleBox werbefrei.
