Chat Commands

StarStreamer makes it easy to create powerful chat commands using Python. Commands can range from simple responses to complex integrations with multiple services.

Implementation Status: ✅ Core Functionality Implemented - Basic command handling, access control, and cooldowns are available. Advanced integrations (OBS, external APIs) require additional implementation.

Basic Commands

Simple Response Command

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("!discord"))
async def discord_command(event: Event, twitch: TwitchClient):
    """Share Discord link"""
    await twitch.send_message("Join our Discord: discord.gg/example")

Personalized Response

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!hello"))
async def hello_command(event: Event, twitch: TwitchClient):
    """Greet the user by name"""
    username = event.data['user']['username']
    await twitch.send_message(f"Hello @{username}! Welcome to the stream! 👋")

Commands with Arguments

Single Argument

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!shoutout"))
async def shoutout_command(event: Event, twitch: TwitchClient):
    """!shoutout @username - Give a shoutout"""
    parts = event.data['message'].split()

    if len(parts) < 2:
        await twitch.send_message("Usage: !shoutout @username")
        return

    target = parts[1].lstrip('@')
    await twitch.send_message(
        f"Check out @{target}! They're an awesome streamer! "
        f"https://twitch.tv/{target}"
    )

Multiple Arguments

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!reminder"))
async def reminder_command(event: Event, twitch: TwitchClient):
    """!reminder <minutes> <message> - Set a reminder"""
    parts = event.data['message'].split(maxsplit=2)

    if len(parts) < 3:
        await twitch.send_message("Usage: !reminder <minutes> <message>")
        return

    try:
        minutes = int(parts[1])
        message = parts[2]

        await twitch.send_message(f"Reminder set for {minutes} minutes: {message}")

        # Schedule the reminder
        await asyncio.sleep(minutes * 60)
        await twitch.send_message(f"⏰ Reminder: {message}")

    except ValueError:
        await twitch.send_message("Minutes must be a number!")

Access Control

Moderator Commands

from starstreamer.triggers import ModOnlyTrigger

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!settitle"))
@trigger(ModOnlyTrigger())
async def set_title_command(event: Event, twitch: TwitchClient):
    """Mod-only command to change stream title"""
    parts = event.data['message'].split(maxsplit=1)

    if len(parts) < 2:
        await twitch.send_message("Usage: !settitle <new title>")
        return

    new_title = parts[1]
    # Note: Stream title update method depends on Twitch API implementation
    await twitch.send_message(f"Stream title update requested: {new_title}")

Subscriber Commands

from starstreamer import filter
from starstreamer.events.filters import TwitchFilters

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!songrequest"))
@filter(TwitchFilters.subscriber_only())
async def song_request(event: Event, twitch: TwitchClient):
    """Subscriber-only song requests"""
    parts = event.data['message'].split(maxsplit=1)

    if len(parts) < 2:
        await twitch.send_message("Usage: !songrequest <song name or URL>")
        return

    song = parts[1]
    # Add to song queue logic here
    await twitch.send_message(f"Added to queue: {song}")

VIP Commands

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!priority"))
@filter(lambda e: 'vip' in e.data.get('badges', {}))
async def vip_priority(event: Event, twitch: TwitchClient):
    """VIP-only priority queue"""
    username = event.data['user']['username']
    await twitch.send_message(f"@{username} added to priority queue!")

Cooldown Management

Per-User Cooldown

from starstreamer.triggers import CooldownTrigger

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!clip"))
@trigger(CooldownTrigger(cooldown_seconds=30, per_user=True))
async def clip_command(event: Event, twitch: TwitchClient):
    """Create a clip - 30 second per-user cooldown"""
    username = event.data['user']['username']

    # Note: create_clip method depends on Twitch API implementation
    await twitch.send_message(f"@{username} clip creation requested!")

Global Cooldown

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!hype"))
@trigger(CooldownTrigger(cooldown_seconds=300, per_user=False))
async def hype_command(event: Event, twitch: TwitchClient):
    """Global hype train - 5 minute cooldown"""
    await twitch.send_message("🚂 HYPE TRAIN ACTIVATED! GET HYPED! 🚂")
    # Trigger hype animations, sounds, etc.

Cooldown Status

from starstreamer.core.cooldowns import CooldownTracker

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!cooldowns"))
async def cooldown_status(
    event: Event, 
    twitch: TwitchClient,
    cooldowns: CooldownTracker
):
    """Check cooldown status"""
    username = event.data['user']['username']
    stats = cooldowns.get_stats()

    await twitch.send_message(
        f"@{username} - Active cooldowns: {stats['active_cooldowns']}"
    )

Command Aliases

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!so"))
async def shoutout_short(event: Event, twitch: TwitchClient):
    """Short shoutout command"""
    # Implementation here
    pass

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!shoutout"))
async def shoutout_long(event: Event, twitch: TwitchClient):
    """Full shoutout command"""
    # Same implementation, currently requires separate handlers
    pass

# Note: Command aliases are planned for future implementation

Dynamic Commands

Command Registry

# Command registry pattern
COMMANDS = {}

def register_command(name: str, description: str):
    """Decorator to register commands"""
    def decorator(func):
        COMMANDS[name] = {
            'handler': func,
            'description': description
        }
        return func
    return decorator

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!help"))
async def help_command(event: Event, twitch: TwitchClient):
    """List all available commands"""
    command_list = ", ".join(f"!{cmd}" for cmd in COMMANDS.keys())
    await twitch.send_message(f"Available commands: {command_list}")

# Register commands
@register_command("game", "Set the stream game")
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!game"))
async def game_command(event: Event, twitch: TwitchClient):
    # Implementation
    pass

Integration Commands

OBS Integration

Note: OBS integration is planned for future implementation. Currently, you would need to implement your own OBS WebSocket client:

# Note: OBSClient is not yet implemented
# This example shows the planned API

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!scene"))
@trigger(ModOnlyTrigger())
async def scene_command(
    event: Event,
    twitch: TwitchClient
):
    """Switch OBS scene (planned functionality)"""
    parts = event.data['message'].split(maxsplit=1)

    if len(parts) < 2:
        await twitch.send_message("Usage: !scene <scene_name>")
    else:
        scene_name = parts[1]
        # OBS integration would go here when implemented
        await twitch.send_message(f"Scene switch to {scene_name} requested")

API Integration

import httpx

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!weather"))
async def weather_command(
    event: Event,
    twitch: TwitchClient,
    http: httpx.AsyncClient
):
    """Get weather for a location"""
    parts = event.data['message'].split(maxsplit=1)

    if len(parts) < 2:
        await twitch.send_message("Usage: !weather <location>")
        return

    location = parts[1]

    # Call weather API
    response = await http.get(
        f"https://api.weather.com/v1/location/{location}"
    )

    if response.status_code == 200:
        data = response.json()
        temp = data['temperature']
        desc = data['description']
        await twitch.send_message(f"Weather in {location}: {temp}°F, {desc}")
    else:
        await twitch.send_message(f"Couldn't get weather for {location}")

Built-in Commands

StarStreamer includes these commands by default:

Command	Description	Access
!hello	Greet the user	Everyone
!ping	Bot health check	Everyone
!uptime	Stream uptime	Everyone
!commands	List commands	Everyone
!clip	Create a clip	Everyone (cooldown)
!shoutout	Shoutout a user	Moderator
!cooldown	Manage cooldowns	Moderator

Command Patterns

Stateful Commands

# Track command state
command_states = {}

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!poll"))
async def poll_command(event: Event, twitch: TwitchClient):
    """Multi-step poll creation"""
    user = event.data['user']['username']
    message = event.data['message']

    if user not in command_states:
        # Start poll creation
        command_states[user] = {'step': 'question'}
        await twitch.send_message(f"@{user} What's your poll question?")
    elif command_states[user]['step'] == 'question':
        # Save question, ask for options
        command_states[user]['question'] = message
        command_states[user]['step'] = 'options'
        await twitch.send_message(f"@{user} Enter options (comma separated)")
    # Continue with poll creation...

Queued Commands

from collections import deque

song_queue = deque(maxlen=50)

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!sr"))
async def song_request_queue(event: Event, twitch: TwitchClient):
    """Add song to queue"""
    parts = event.data['message'].split(maxsplit=1)
    if len(parts) > 1:
        song = parts[1]
        username = event.data['user']['username']
        song_queue.append({'song': song, 'user': username})
        position = len(song_queue)
        await twitch.send_message(f"Added '{song}' to queue (position {position})")

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!nowplaying"))
async def now_playing(event: Event, twitch: TwitchClient):
    """Show current song"""
    if song_queue:
        current = song_queue[0]
        await twitch.send_message(
            f"Now playing: {current['song']} (requested by {current['user']})"
        )

Error Handling

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!dice"))
async def dice_command(event: Event, twitch: TwitchClient, logger):
    """Roll dice with error handling"""
    parts = event.data['message'].split()

    try:
        sides = int(parts[1]) if len(parts) > 1 else 6

        if sides < 2 or sides > 1000:
            await twitch.send_message("Dice must have 2-1000 sides!")
            return

        import random
        result = random.randint(1, sides)
        username = event.data['user']['username']
        await twitch.send_message(f"@{username} rolled a {result} (d{sides})")

    except ValueError:
        await twitch.send_message("Usage: !dice [number_of_sides]")
    except Exception as e:
        logger.error(f"Error in dice command: {e}")
        await twitch.send_message("Something went wrong with the dice roll!")

Testing Commands

# test_commands.py
import pytest
from unittest.mock import AsyncMock, MagicMock

@pytest.mark.asyncio
async def test_hello_command():
    # Create mock event
    mock_event = MagicMock()
    mock_event.data = {
        'user': {'username': 'testuser'},
        'message': '!hello'
    }

    # Create mock Twitch client
    mock_twitch = AsyncMock()

    # Run command
    await hello_command(mock_event, mock_twitch)

    # Verify response
    mock_twitch.send_message.assert_called_once_with(
        "Hello @testuser! Welcome to the stream! 👋"
    )

Next Steps

• Learn about Cooldown Management
• Explore OBS Integration
• See Examples for more patterns