Custom Commands Examples¶
This guide provides comprehensive examples of custom chat commands in StarStreamer, from simple responses to complex integrations. These examples demonstrate best practices and common patterns for building engaging stream interactions.
Getting Started¶
All commands follow the same basic pattern:
from starstreamer import on_event
from starstreamer.triggers import trigger, CommandTrigger
from starstreamer.plugins.twitch import TwitchClient
from starstreamer.runtime.types import Event
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!command"))
async def my_command(event: Event, twitch: TwitchClient) -> None:
"""Command description"""
# Command logic here
await twitch.send_message("Response message")
Basic Commands¶
Simple Response Commands¶
Discord Invite¶
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!discord"))
async def discord_command(event: Event, twitch: TwitchClient) -> None:
"""Share Discord invite link"""
await twitch.send_message("🎮 Join our Discord community: https://discord.gg/your-invite-code")
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!socials"))
async def socials_command(event: Event, twitch: TwitchClient) -> None:
"""Share all social media links"""
socials = [
"📱 Twitter: https://twitter.com/yourusername",
"📸 Instagram: https://instagram.com/yourusername",
"🎮 YouTube: https://youtube.com/@yourchannel",
"💬 Discord: https://discord.gg/your-invite-code"
]
await twitch.send_message("Find me on social media:")
for social in socials:
await twitch.send_message(social)
Stream Information¶
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!schedule"))
async def schedule_command(event: Event, twitch: TwitchClient) -> None:
"""Show streaming schedule"""
schedule = [
"📅 **Streaming Schedule**",
"Monday: 7-10 PM EST - Coding Stream",
"Wednesday: 7-10 PM EST - Gaming",
"Friday: 8-11 PM EST - Community Games",
"Saturday: 2-6 PM EST - Variety Stream"
]
for line in schedule:
await twitch.send_message(line)
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!setup"))
async def setup_command(event: Event, twitch: TwitchClient) -> None:
"""Show streaming setup"""
await twitch.send_message(
"🖥️ Setup: RTX 4080 | Ryzen 7900X | 32GB RAM | "
"🎤 Audio-Technica AT2020 | 📷 Sony A7 III"
)
Personalized Commands¶
Greeting with Username¶
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!hello"))
async def hello_command(event: Event, twitch: TwitchClient) -> None:
"""Personalized greeting"""
user = event.data.get("user", {})
username = user.get("display_name", user.get("username", "friend"))
greetings = [
f"Hello {username}! 👋 Welcome to the stream!",
f"Hey there {username}! Glad you're here! 😊",
f"What's up {username}? Thanks for stopping by! 🎉",
f"Welcome {username}! Hope you enjoy the stream! ✨"
]
import random
greeting = random.choice(greetings)
await twitch.send_message(greeting)
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!lurk"))
async def lurk_command(event: Event, twitch: TwitchClient) -> None:
"""Acknowledge lurkers"""
user = event.data.get("user", {})
username = user.get("display_name", user.get("username", "friend"))
await twitch.send_message(
f"Thanks for lurking, {username}! Enjoy the stream in the background 💜"
)
Social Interaction Commands¶
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!hug"))
async def hug_command(event: Event, twitch: TwitchClient) -> None:
"""Send virtual hugs: !hug @username or !hug for everyone"""
message = event.data.get("message", "")
parts = message.split(maxsplit=1)
user = event.data.get("user", {})
sender = user.get("display_name", user.get("username", "Someone"))
if len(parts) > 1 and parts[1]:
# Hug specific user
target = parts[1].lstrip("@")
await twitch.send_message(f"{sender} sends a warm hug to {target}! 🤗")
else:
# Hug everyone
await twitch.send_message(f"{sender} sends hugs to everyone in chat! 🤗💕")
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!highfive"))
async def highfive_command(event: Event, twitch: TwitchClient) -> None:
"""Give high fives: !highfive @username"""
message = event.data.get("message", "")
parts = message.split(maxsplit=1)
user = event.data.get("user", {})
sender = user.get("display_name", user.get("username", "Someone"))
if len(parts) > 1:
target = parts[1].lstrip("@")
await twitch.send_message(f"{sender} gives {target} an epic high five! ✋")
else:
await twitch.send_message(f"High five, {sender}! ✋")
Commands with Arguments¶
Single Argument Commands¶
Shoutout Command¶
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!so")) # Short alias
@trigger(CommandTrigger("!shoutout"))
async def shoutout_command(event: Event, twitch: TwitchClient) -> None:
"""Give a shoutout to another streamer: !so @username"""
message = event.data.get("message", "")
parts = message.split()
if len(parts) < 2:
await twitch.send_message("Usage: !so @username")
return
target = parts[1].lstrip("@")
# Multiple shoutout messages for variety
shoutout_messages = [
f"🌟 Check out @{target}! They're an amazing streamer! https://twitch.tv/{target}",
f"📺 Go show some love to @{target}! They create awesome content! https://twitch.tv/{target}",
f"💜 Shoutout to @{target}! Definitely worth a follow! https://twitch.tv/{target}"
]
import random
message = random.choice(shoutout_messages)
await twitch.send_message(message)
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!weather"))
async def weather_command(event: Event, twitch: TwitchClient) -> None:
"""Get weather for a city: !weather City"""
message = event.data.get("message", "")
parts = message.split(maxsplit=1)
if len(parts) < 2:
await twitch.send_message("Usage: !weather <city>")
return
city = parts[1]
# This would integrate with a weather API in a real implementation
# For demo purposes, we'll simulate it
await twitch.send_message(f"🌤️ Weather for {city}: Partly cloudy, 72°F (22°C)")
Multiple Argument Commands¶
Timer/Reminder System¶
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!timer"))
async def timer_command(event: Event, twitch: TwitchClient) -> None:
"""Set a timer: !timer <minutes> <message>"""
import asyncio
message = event.data.get("message", "")
parts = message.split(maxsplit=2)
if len(parts) < 3:
await twitch.send_message("Usage: !timer <minutes> <message>")
return
try:
minutes = int(parts[1])
timer_message = parts[2]
if minutes <= 0 or minutes > 60:
await twitch.send_message("Timer must be between 1-60 minutes!")
return
user = event.data.get("user", {})
username = user.get("display_name", "Someone")
await twitch.send_message(f"⏰ Timer set by {username} for {minutes} minutes: {timer_message}")
# Run timer in background
asyncio.create_task(run_timer(twitch, minutes, timer_message, username))
except ValueError:
await twitch.send_message("Minutes must be a number!")
async def run_timer(twitch: TwitchClient, minutes: int, message: str, username: str) -> None:
"""Background task to handle timer"""
import asyncio
await asyncio.sleep(minutes * 60)
await twitch.send_message(f"⏰ Timer by {username}: {message}")
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!quote"))
async def quote_command(event: Event, twitch: TwitchClient) -> None:
"""Add or get quotes: !quote add <text> OR !quote <number>"""
message = event.data.get("message", "")
parts = message.split(maxsplit=2)
if len(parts) == 1:
# Random quote
quotes = [
"The only way to do great work is to love what you do. - Steve Jobs",
"Code is like humor. When you have to explain it, it's bad. - Cory House",
"Programming isn't about what you know; it's about what you can figure out. - Chris Pine"
]
import random
quote = random.choice(quotes)
await twitch.send_message(f"📝 Quote: {quote}")
elif len(parts) >= 3 and parts[1].lower() == "add":
# Add new quote (in a real implementation, this would save to database)
quote_text = parts[2]
await twitch.send_message(f"✅ Quote added: \"{quote_text}\"")
else:
await twitch.send_message("Usage: !quote [add <text>] OR !quote [number]")
Access Control¶
Moderator Commands¶
from starstreamer.triggers import ModOnlyTrigger
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!title"))
@trigger(ModOnlyTrigger())
async def set_title_command(event: Event, twitch: TwitchClient) -> None:
"""Mod-only: Change stream title"""
message = event.data.get("message", "")
parts = message.split(maxsplit=1)
if len(parts) < 2:
await twitch.send_message("Usage: !title <new title>")
return
new_title = parts[1]
# In a real implementation, this would update the stream title via Twitch API
await twitch.send_message(f"✅ Stream title updated to: {new_title}")
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!clear"))
@trigger(ModOnlyTrigger())
async def clear_chat_command(event: Event, twitch: TwitchClient) -> None:
"""Mod-only: Clear chat"""
# In a real implementation, this would clear chat via Twitch API
await twitch.send_message("🧹 Chat has been cleared by a moderator")
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!timeout"))
@trigger(ModOnlyTrigger())
async def timeout_command(event: Event, twitch: TwitchClient) -> None:
"""Mod-only: Timeout user"""
message = event.data.get("message", "")
parts = message.split()
if len(parts) < 3:
await twitch.send_message("Usage: !timeout <username> <seconds>")
return
username = parts[1].lstrip("@")
try:
seconds = int(parts[2])
# In a real implementation, this would timeout via Twitch API
await twitch.send_message(f"⏳ {username} timed out for {seconds} seconds")
except ValueError:
await twitch.send_message("Timeout duration must be a number!")
Subscriber Commands¶
from starstreamer.triggers import SubscriberOnlyTrigger
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!subhype"))
@trigger(SubscriberOnlyTrigger())
async def sub_hype_command(event: Event, twitch: TwitchClient) -> None:
"""Subscriber-only hype command"""
user = event.data.get("user", {})
username = user.get("display_name", "Subscriber")
hype_messages = [
f"🎉 {username} is spreading the sub hype! Thank you for your support! 💜",
f"🔥 Sub squad represent! {username} is awesome! 🔥",
f"✨ {username} keeping the energy high! Love our sub community! ✨"
]
import random
await twitch.send_message(random.choice(hype_messages))
Cooldown Commands¶
Individual User Cooldowns¶
from starstreamer.triggers import CooldownTrigger
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!daily"))
@trigger(CooldownTrigger(86400, per_user=True)) # 24 hours per user
async def daily_bonus_command(event: Event, twitch: TwitchClient) -> None:
"""Daily bonus command with 24-hour cooldown"""
user = event.data.get("user", {})
username = user.get("display_name", "User")
# Simulate giving daily bonus
bonus_amount = 100
await twitch.send_message(f"🎁 {username} claimed their daily bonus of {bonus_amount} points!")
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!work"))
@trigger(CooldownTrigger(300, per_user=True)) # 5 minutes per user
async def work_command(event: Event, twitch: TwitchClient) -> None:
"""Work command with 5-minute cooldown"""
user = event.data.get("user", {})
username = user.get("display_name", "Worker")
jobs = [
("delivered pizzas", "🍕", 50),
("fixed computers", "💻", 75),
("streamed on Twitch", "📺", 100),
("created memes", "😂", 25),
("coded an app", "⌨️", 80)
]
import random
job, emoji, earnings = random.choice(jobs)
await twitch.send_message(f"{emoji} {username} {job} and earned {earnings} points!")
Global Cooldowns¶
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!trivia"))
@trigger(CooldownTrigger(60, per_user=False)) # Global 1-minute cooldown
async def trivia_command(event: Event, twitch: TwitchClient) -> None:
"""Trivia question with global cooldown"""
trivia_questions = [
{
"question": "What year was Python first released?",
"answer": "1991",
"options": ["A) 1989", "B) 1991", "C) 1993", "D) 1995"]
},
{
"question": "Which planet is known as the Red Planet?",
"answer": "Mars",
"options": ["A) Venus", "B) Jupiter", "C) Mars", "D) Saturn"]
},
{
"question": "What does CPU stand for?",
"answer": "Central Processing Unit",
"options": ["A) Computer Processing Unit", "B) Central Processing Unit", "C) Core Processing Unit", "D) Central Program Unit"]
}
]
import random
trivia = random.choice(trivia_questions)
await twitch.send_message(f"🧠 Trivia: {trivia['question']}")
for option in trivia['options']:
await twitch.send_message(option)
# In a real implementation, you'd wait for answers and track correct responses
Database Integration¶
User Economy System¶
from starstreamer.services.economy import EconomyService
from starstreamer.services.users import UserService
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!balance"))
async def balance_command(
event: Event,
twitch: TwitchClient,
economy: EconomyService,
users: UserService
) -> None:
"""Check your balance"""
user_data = event.data.get("user", {})
username = user_data.get("display_name", "Unknown")
user_id = user_data.get("id", "")
if not user_id:
await twitch.send_message(f"@{username} Could not identify your account!")
return
# Ensure user exists
await users.get_or_create_user(user_id, username)
# Get balance
balance = await economy.get_balance(user_id)
await twitch.send_message(f"@{username} Your balance is ${balance}")
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!give"))
async def give_money_command(
event: Event,
twitch: TwitchClient,
economy: EconomyService
) -> None:
"""Give money to another user: !give @user amount"""
message = event.data.get("message", "")
parts = message.split()
if len(parts) < 3:
await twitch.send_message("Usage: !give @username amount")
return
user_data = event.data.get("user", {})
sender_name = user_data.get("display_name", "Unknown")
sender_id = user_data.get("id", "")
target_name = parts[1].lstrip("@")
try:
amount = int(parts[2])
if amount <= 0:
await twitch.send_message("Amount must be positive!")
return
except ValueError:
await twitch.send_message("Amount must be a number!")
return
# In a real implementation, you'd lookup the target user's ID
target_id = target_name.lower() # Simplified
success = await economy.transfer_money(sender_id, target_id, amount, f"Gift from {sender_name}")
if success:
await twitch.send_message(f"💰 {sender_name} gave ${amount} to @{target_name}!")
else:
await twitch.send_message(f"@{sender_name} You don't have enough money!")
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!leaderboard"))
@trigger(CooldownTrigger(30, per_user=False)) # Global cooldown to prevent spam
async def leaderboard_command(event: Event, twitch: TwitchClient, economy: EconomyService) -> None:
"""Show top 5 richest users"""
leaderboard = await economy.get_leaderboard(limit=5)
if not leaderboard:
await twitch.send_message("No users have earned money yet!")
return
await twitch.send_message("💰 **Top 5 Richest Users:**")
for i, (username, balance) in enumerate(leaderboard, 1):
medal = "🥇" if i == 1 else "🥈" if i == 2 else "🥉" if i == 3 else "🏅"
await twitch.send_message(f"{medal} {i}. {username}: ${balance}")
Variable Storage¶
from starstreamer.db.repositories.variable_repository import VariableRepository
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!counter"))
async def counter_command(event: Event, twitch: TwitchClient, variables: VariableRepository) -> None:
"""Stream counter that persists between restarts"""
count = await variables.increment("stream_counter", default=0)
await twitch.send_message(f"🔢 Stream counter: {count}")
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!setgoal"))
@trigger(ModOnlyTrigger())
async def set_goal_command(event: Event, twitch: TwitchClient, variables: VariableRepository) -> None:
"""Mod-only: Set donation goal"""
message = event.data.get("message", "")
parts = message.split()
if len(parts) < 2:
await twitch.send_message("Usage: !setgoal <amount>")
return
try:
goal = float(parts[1])
await variables.set("donation_goal", goal, description="Current stream donation goal")
await twitch.send_message(f"🎯 Donation goal set to ${goal}")
except ValueError:
await twitch.send_message("Goal amount must be a number!")
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!goal"))
async def check_goal_command(event: Event, twitch: TwitchClient, variables: VariableRepository) -> None:
"""Check current donation goal"""
goal = await variables.get("donation_goal", default=0)
current = await variables.get("current_donations", default=0)
if goal == 0:
await twitch.send_message("No donation goal is currently set!")
return
remaining = max(0, goal - current)
percentage = min(100, (current / goal) * 100)
await twitch.send_message(
f"🎯 Goal: ${current:.2f} / ${goal:.2f} ({percentage:.1f}%) - "
f"${remaining:.2f} remaining!"
)
Interactive Commands¶
Poll System¶
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!poll"))
@trigger(ModOnlyTrigger())
async def create_poll_command(event: Event, twitch: TwitchClient, variables: VariableRepository) -> None:
"""Mod-only: Create a poll"""
message = event.data.get("message", "")
parts = message.split(maxsplit=1)
if len(parts) < 2:
await twitch.send_message("Usage: !poll <question>")
return
question = parts[1]
# Store poll data
poll_data = {
"question": question,
"options": {"A": 0, "B": 0, "C": 0, "D": 0},
"active": True,
"voters": []
}
await variables.set("current_poll", poll_data, persistent=False)
await twitch.send_message(f"📊 **Poll Started:** {question}")
await twitch.send_message("Vote with: !vote A, !vote B, !vote C, or !vote D")
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!vote"))
async def vote_command(event: Event, twitch: TwitchClient, variables: VariableRepository) -> None:
"""Vote in active poll"""
message = event.data.get("message", "")
parts = message.split()
if len(parts) < 2:
return
vote = parts[1].upper()
if vote not in ["A", "B", "C", "D"]:
return
user_data = event.data.get("user", {})
user_id = user_data.get("id", "")
username = user_data.get("display_name", "Unknown")
# Get current poll
poll_data = await variables.get("current_poll", default=None, persistent=False)
if not poll_data or not poll_data.get("active"):
await twitch.send_message("No active poll!")
return
# Check if user already voted
if user_id in poll_data.get("voters", []):
await twitch.send_message(f"@{username} You already voted!")
return
# Record vote
poll_data["options"][vote] += 1
poll_data["voters"].append(user_id)
await variables.set("current_poll", poll_data, persistent=False)
await twitch.send_message(f"✅ {username} voted for option {vote}!")
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!pollresults"))
async def poll_results_command(event: Event, twitch: TwitchClient, variables: VariableRepository) -> None:
"""Show poll results"""
poll_data = await variables.get("current_poll", default=None, persistent=False)
if not poll_data:
await twitch.send_message("No poll data available!")
return
total_votes = sum(poll_data["options"].values())
if total_votes == 0:
await twitch.send_message("No votes yet!")
return
await twitch.send_message(f"📊 **Poll Results:** {poll_data['question']}")
for option, votes in poll_data["options"].items():
if votes > 0:
percentage = (votes / total_votes) * 100
await twitch.send_message(f"Option {option}: {votes} votes ({percentage:.1f}%)")
Game Commands¶
Dice Rolling¶
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!roll"))
async def dice_roll_command(event: Event, twitch: TwitchClient) -> None:
"""Roll dice: !roll [sides] [count]"""
message = event.data.get("message", "")
parts = message.split()
user = event.data.get("user", {})
username = user.get("display_name", "Player")
# Default to 1d6
sides = 6
count = 1
try:
if len(parts) >= 2:
sides = int(parts[1])
if sides < 2 or sides > 100:
await twitch.send_message("Dice must have 2-100 sides!")
return
if len(parts) >= 3:
count = int(parts[2])
if count < 1 or count > 10:
await twitch.send_message("Can roll 1-10 dice at once!")
return
except ValueError:
await twitch.send_message("Usage: !roll [sides] [count]")
return
import random
rolls = [random.randint(1, sides) for _ in range(count)]
total = sum(rolls)
if count == 1:
await twitch.send_message(f"🎲 {username} rolled a {rolls[0]} (d{sides})")
else:
rolls_str = ", ".join(map(str, rolls))
await twitch.send_message(f"🎲 {username} rolled: {rolls_str} = {total} ({count}d{sides})")
#### Coin Flip
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!flip"))
async def coin_flip_command(event: Event, twitch: TwitchClient) -> None:
"""Flip a coin"""
user = event.data.get("user", {})
username = user.get("display_name", "Player")
import random
result = random.choice(["Heads", "Tails"])
emoji = "🪙" if result == "Heads" else "🪙"
await twitch.send_message(f"{emoji} {username} flipped: **{result}**!")
#### 8-Ball
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!8ball"))
async def eightball_command(event: Event, twitch: TwitchClient) -> None:
"""Magic 8-ball: !8ball <question>"""
message = event.data.get("message", "")
parts = message.split(maxsplit=1)
if len(parts) < 2:
await twitch.send_message("Usage: !8ball <question>")
return
user = event.data.get("user", {})
username = user.get("display_name", "Seeker")
responses = [
"It is certain", "Without a doubt", "Yes definitely", "You may rely on it",
"As I see it, yes", "Most likely", "Outlook good", "Yes",
"Signs point to yes", "Reply hazy, try again", "Ask again later",
"Better not tell you now", "Cannot predict now", "Concentrate and ask again",
"Don't count on it", "My reply is no", "My sources say no",
"Outlook not so good", "Very doubtful"
]
import random
response = random.choice(responses)
await twitch.send_message(f"🎱 {username} asks: {parts[1]}")
await twitch.send_message(f"🎱 Magic 8-Ball says: **{response}**")
API Integration Examples¶
Weather Command (Mock)¶
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!weather"))
@trigger(CooldownTrigger(30, per_user=False)) # Rate limit API calls
async def weather_command(event: Event, twitch: TwitchClient, logger: logging.Logger) -> None:
"""Get weather for a location: !weather <city>"""
message = event.data.get("message", "")
parts = message.split(maxsplit=1)
if len(parts) < 2:
await twitch.send_message("Usage: !weather <city>")
return
city = parts[1]
try:
# Mock weather API call (replace with real API)
import random
temps = [15, 20, 25, 30, 22, 18, 28]
conditions = ["Sunny", "Cloudy", "Rainy", "Partly Cloudy", "Overcast"]
temp = random.choice(temps)
condition = random.choice(conditions)
await twitch.send_message(
f"🌤️ Weather in {city}: {condition}, {temp}°C ({int(temp * 9/5 + 32)}°F)"
)
except Exception as e:
logger.error(f"Weather API error: {e}")
await twitch.send_message("Weather service temporarily unavailable!")
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!time"))
async def time_command(event: Event, twitch: TwitchClient) -> None:
"""Get current time in different timezones"""
from datetime import datetime
import pytz
timezones = {
"EST": "America/New_York",
"PST": "America/Los_Angeles",
"GMT": "GMT",
"JST": "Asia/Tokyo"
}
await twitch.send_message("🕐 **Current Times:**")
for tz_name, tz_string in timezones.items():
tz = pytz.timezone(tz_string)
current_time = datetime.now(tz).strftime("%H:%M")
await twitch.send_message(f"{tz_name}: {current_time}")
Quote Database¶
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!addquote"))
@trigger(ModOnlyTrigger())
async def add_quote_command(event: Event, twitch: TwitchClient, variables: VariableRepository) -> None:
"""Mod-only: Add a quote to the database"""
message = event.data.get("message", "")
parts = message.split(maxsplit=1)
if len(parts) < 2:
await twitch.send_message("Usage: !addquote <quote text>")
return
quote_text = parts[1]
# Get existing quotes
quotes = await variables.get("stream_quotes", default=[], persistent=True)
# Add new quote with ID
quote_id = len(quotes) + 1
new_quote = {
"id": quote_id,
"text": quote_text,
"added_by": event.data.get("user", {}).get("display_name", "Moderator"),
"date": datetime.now().isoformat()
}
quotes.append(new_quote)
await variables.set("stream_quotes", quotes, description="Stream quotes database")
await twitch.send_message(f"✅ Quote #{quote_id} added: \"{quote_text}\"")
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!quote"))
async def get_quote_command(event: Event, twitch: TwitchClient, variables: VariableRepository) -> None:
"""Get a random quote or specific quote by ID"""
message = event.data.get("message", "")
parts = message.split()
quotes = await variables.get("stream_quotes", default=[], persistent=True)
if not quotes:
await twitch.send_message("No quotes available! Mods can add them with !addquote")
return
if len(parts) >= 2:
try:
quote_id = int(parts[1])
quote = next((q for q in quotes if q["id"] == quote_id), None)
if quote:
await twitch.send_message(f"📝 Quote #{quote['id']}: \"{quote['text']}\"")
else:
await twitch.send_message(f"Quote #{quote_id} not found!")
return
except ValueError:
pass
# Random quote
import random
quote = random.choice(quotes)
await twitch.send_message(f"📝 Quote #{quote['id']}: \"{quote['text']}\"")
Advanced Patterns¶
Command Aliases¶
# Multiple triggers for the same command
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!so"))
@trigger(CommandTrigger("!shoutout"))
@trigger(CommandTrigger("!s"))
async def shoutout_aliases(event: Event, twitch: TwitchClient) -> None:
"""Shoutout command with multiple aliases"""
# Implementation here
pass
Dynamic Command Registration¶
class CustomCommandManager:
def __init__(self, variables: VariableRepository):
self.variables = variables
async def add_command(self, name: str, response: str) -> bool:
"""Add a custom command"""
commands = await self.variables.get("custom_commands", default={})
commands[name] = response
await self.variables.set("custom_commands", commands)
return True
async def get_response(self, command: str) -> str | None:
"""Get response for custom command"""
commands = await self.variables.get("custom_commands", default={})
return commands.get(command)
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!addcmd"))
@trigger(ModOnlyTrigger())
async def add_custom_command(
event: Event,
twitch: TwitchClient,
variables: VariableRepository
) -> None:
"""Mod-only: Add custom command"""
message = event.data.get("message", "")
parts = message.split(maxsplit=2)
if len(parts) < 3:
await twitch.send_message("Usage: !addcmd <command> <response>")
return
cmd_name = parts[1].lstrip("!")
response = parts[2]
manager = CustomCommandManager(variables)
await manager.add_command(cmd_name, response)
await twitch.send_message(f"✅ Custom command !{cmd_name} added!")
@on_event("twitch.chat.message")
async def handle_custom_commands(
event: Event,
twitch: TwitchClient,
variables: VariableRepository
) -> None:
"""Handle dynamic custom commands"""
message = event.data.get("message", "")
if not message.startswith("!"):
return
cmd_name = message.split()[0][1:] # Remove !
manager = CustomCommandManager(variables)
response = await manager.get_response(cmd_name)
if response:
await twitch.send_message(response)
Error Handling and Best Practices¶
Robust Command Structure¶
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!robust"))
async def robust_command(
event: Event,
twitch: TwitchClient,
logger: logging.Logger
) -> None:
"""Example of robust command with proper error handling"""
try:
# Extract user data safely
user_data = event.data.get("user", {})
username = user_data.get("display_name", user_data.get("username", "Unknown"))
user_id = user_data.get("id")
if not user_id:
logger.warning(f"No user ID for command from {username}")
await twitch.send_message("Unable to identify user!")
return
# Command logic here
await twitch.send_message(f"Command executed successfully for {username}!")
except Exception as e:
logger.error(f"Error in robust_command: {e}")
await twitch.send_message("Command failed! Please try again later.")
Input Validation¶
def validate_username(username: str) -> bool:
"""Validate Twitch username format"""
import re
return bool(re.match(r'^[a-zA-Z0-9_]{3,25}$', username))
def validate_amount(amount_str: str, min_val: int = 1, max_val: int = 1000000) -> int | None:
"""Validate and parse amount input"""
try:
amount = int(amount_str)
if min_val <= amount <= max_val:
return amount
except ValueError:
pass
return None
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!pay"))
async def validated_pay_command(event: Event, twitch: TwitchClient) -> None:
"""Pay command with input validation"""
message = event.data.get("message", "")
parts = message.split()
if len(parts) < 3:
await twitch.send_message("Usage: !pay @username amount")
return
# Validate username
target_username = parts[1].lstrip("@")
if not validate_username(target_username):
await twitch.send_message("Invalid username format!")
return
# Validate amount
amount = validate_amount(parts[2], min_val=1, max_val=10000)
if amount is None:
await twitch.send_message("Amount must be between 1 and 10,000!")
return
# Process payment...
await twitch.send_message(f"Payment of {amount} to @{target_username} processed!")
Testing Commands¶
import pytest
from unittest.mock import AsyncMock, Mock
@pytest.mark.asyncio
async def test_hello_command():
"""Test hello command execution"""
# Mock dependencies
twitch = AsyncMock()
event = Mock()
event.data = {
"user": {
"display_name": "TestUser",
"username": "testuser"
}
}
# Import and execute command
from your_commands import hello_command
await hello_command(event, twitch)
# Verify message was sent
twitch.send_message.assert_called_once()
args = twitch.send_message.call_args[0]
assert "TestUser" in args[0]
assert "Welcome" in args[0]
Performance Tips¶
- Use cooldowns on expensive operations
- Batch database operations when possible
- Cache frequently accessed data in variables
- Limit response length to avoid rate limits
- Handle errors gracefully to maintain stability
- Log important events for debugging
- Validate input to prevent crashes
Command Organization¶
File Structure¶
src/modules/your_module/
├── __init__.py
├── module.py
└── actions/
├── __init__.py
├── basic_commands.py # Simple commands
├── economy_commands.py # Economy features
├── moderation_commands.py # Mod tools
├── game_commands.py # Interactive games
└── api_commands.py # External integrations
Module Registration¶
# In your module.py
from starstreamer.modules import BaseModule
class YourModule(BaseModule):
name = "your_module"
description = "Custom commands module"
async def load(self) -> None:
"""Load the module and register commands"""
# Commands are automatically registered via decorators
from . import actions # noqa: F401
self.logger.info("Custom commands loaded")
This comprehensive guide should give you everything needed to create engaging, robust chat commands for your StarStreamer bot. Start with simple commands and gradually add more complex features as needed!