Your First Action¶

Let's create your first StarStreamer action! We'll build a welcome bot that greets new viewers and responds to commands using the module system.

Understanding Actions¶

Actions in StarStreamer are Python functions that: - Respond to events (chat messages, follows, subs, raids) - Use dependency injection for clean, testable code - Can use triggers for filtering (commands, cooldowns, conditions) - Live inside modules for organization

Understanding Modules¶

StarStreamer uses a module system where related actions are grouped together. Each module: - Extends the BaseModule class - Contains action handlers in an actions/ subdirectory - Can be dynamically loaded and unloaded - Has access to all framework services via dependency injection

Option 1: Add to an Existing Module (Quick Start)¶

The easiest way to start is by adding actions to the existing chat module.

Edit the Chat Module¶

Open src/modules/chat/actions/basic_commands.py and add your 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("!welcome"))
async def welcome_command(event: Event, twitch: TwitchClient) -> None:
    """Welcome command that greets the user"""
    user = event.data.get("user", {})
    username = user.get("display_name", "friend")

    await twitch.send_message(
        f"Welcome to the stream @{username}! "
        f"Enjoy your stay! 🎉"
    )

That's it! Restart StarStreamer and your command is ready to use.

Option 2: Create Your Own Module (Recommended)¶

For more complex features, create your own module.

Step 1: Create Module Structure¶

# Create module directories
mkdir -p src/modules/welcome/actions
touch src/modules/welcome/__init__.py
touch src/modules/welcome/module.py
touch src/modules/welcome/actions/__init__.py
touch src/modules/welcome/actions/commands.py

Step 2: Define Your Module¶

Create src/modules/welcome/module.py:

"""Welcome module for greeting viewers"""

from modules.base import BaseModule

class WelcomeModule(BaseModule):
    """Module for welcoming viewers and providing information"""

    @property
    def module_name(self) -> str:
        """Unique identifier for this module"""
        return "welcome"

    @property
    def description(self) -> str:
        """Human-readable description"""
        return "Welcomes viewers and provides stream information"

    @property
    def version(self) -> str:
        """Module version"""
        return "1.0.0"

    @property
    def author(self) -> str:
        """Module author"""
        return "Your Name"

    async def initialize(self) -> None:
        """Initialize module resources"""
        # Any setup code goes here
        pass

    async def shutdown(self) -> None:
        """Clean up module resources"""
        # Any cleanup code goes here
        pass

    async def register_actions(self) -> None:
        """Register action handlers"""
        # Import your actions to register them with the event bus
        from modules.welcome.actions import commands  # noqa: F401

Step 3: Create Module Init¶

Create src/modules/welcome/__init__.py:

"""Welcome module for StarStreamer"""

from .module import WelcomeModule

__all__ = ["WelcomeModule"]

Step 4: Add Your Actions¶

Create src/modules/welcome/actions/commands.py:

"""Welcome bot commands and auto-greeting"""

import logging
from starstreamer import on_event
from starstreamer.triggers import trigger, CommandTrigger, CooldownTrigger
from starstreamer.plugins.twitch import TwitchClient
from starstreamer.runtime.types import Event

# Track users we've seen (in production, use a database)
seen_users = set()

# ========== Auto-Welcome ==========

@on_event("twitch.chat.message")
async def welcome_new_chatters(
    event: Event, 
    twitch: TwitchClient, 
    logger: logging.Logger
) -> None:
    """Automatically welcome first-time chatters"""
    user = event.data.get("user", {})
    username = user.get("display_name", "viewer")
    user_id = user.get("id")

    # Skip if we've seen this user before
    if user_id in seen_users:
        return

    # Add to seen users
    seen_users.add(user_id)
    logger.info(f"First message from {username} (ID: {user_id})")

    # Welcome them
    await twitch.send_message(
        f"Hey @{username}, welcome to the stream! "
        f"Thanks for chatting! Type !help for commands."
    )

# ========== Welcome Commands ==========

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!welcome"))
async def welcome_command(event: Event, twitch: TwitchClient) -> None:
    """Welcome command that greets the user"""
    user = event.data.get("user", {})
    username = user.get("display_name", "friend")

    await twitch.send_message(
        f"Welcome @{username}! Enjoy your stay! 🎉"
    )

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!help"))
async def help_command(event: Event, twitch: TwitchClient) -> None:
    """List available commands"""
    await twitch.send_message(
        "Commands: !welcome, !help, !schedule, !socials, "
        "!discord, !hype, !lurk"
    )

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!schedule"))
async def schedule_command(event: Event, twitch: TwitchClient) -> None:
    """Share stream schedule"""
    await twitch.send_message(
        "📅 Stream Schedule: Mon/Wed/Fri at 7 PM EST | "
        "Sat/Sun at 2 PM EST"
    )

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!socials"))
async def socials_command(event: Event, twitch: TwitchClient) -> None:
    """Share social media links"""
    await twitch.send_message(
        "📱 Follow me: Twitter @username | "
        "YouTube: youtube.com/username | "
        "Instagram: @username"
    )

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!discord"))
async def discord_command(event: Event, twitch: TwitchClient) -> None:
    """Share Discord invite"""
    await twitch.send_message(
        "💬 Join our Discord community: discord.gg/yourchannel"
    )

@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", "lurker")

    await twitch.send_message(
        f"Thanks for lurking @{username}! Enjoy the stream 💜"
    )

# ========== Fun Commands with Cooldowns ==========

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!hype"))
@trigger(CooldownTrigger(60))  # 60-second global cooldown
async def hype_command(event: Event, twitch: TwitchClient) -> None:
    """Hype command with cooldown"""
    user = event.data.get("user", {})
    username = user.get("display_name", "someone")

    await twitch.send_message(
        f"🔥🔥🔥 @{username} is bringing the HYPE! "
        f"Everyone get HYPED! 🔥🔥🔥"
    )

# ========== Follow Alerts ==========

@on_event("twitch.follow")
async def follow_alert(event: Event, twitch: TwitchClient, logger: logging.Logger) -> None:
    """Thank new followers"""
    user = event.data.get("user", {})
    username = user.get("display_name", "someone")

    logger.info(f"New follower: {username}")

    await twitch.send_message(
        f"Thanks for the follow @{username}! "
        f"Welcome to our amazing community! 💜"
    )

# ========== Subscription Alerts ==========

@on_event("twitch.subscription")
async def sub_alert(event: Event, twitch: TwitchClient, logger: logging.Logger) -> None:
    """Thank subscribers"""
    user = event.data.get("user", {})
    username = user.get("display_name", "someone")
    tier = event.data.get("tier", "1")

    logger.info(f"New subscription from {username} (Tier {tier})")

    tier_emotes = {"1": "⭐", "2": "🌟", "3": "💫"}
    emote = tier_emotes.get(tier, "⭐")

    await twitch.send_message(
        f"{emote} Thank you for subscribing @{username}! "
        f"You're awesome! {emote}"
    )

Step 5: Test Your Module¶

You can test your module in two ways:

Option A: Using the Web Interface (Recommended)¶

Start StarStreamer:
```
uv run python src/main.py
```
Open the web interface:
Go to http://localhost:8888
Navigate to the Modules page
Your new module should appear in the list
Click Load if not already loaded, then Enable
Test in chat:
Type !help to see commands
Type !welcome to test the welcome
Have someone new chat to trigger auto-welcome

Option B: Restart StarStreamer¶

Restart StarStreamer:
```
uv run python src/main.py
```

Check the logs:

INFO: Loading module: welcome
INFO: Registered handler: welcome_new_chatters
INFO: Registered handler: welcome_command
INFO: Registered handler: help_command

Advanced Features¶

Using Services via Dependency Injection¶

Access database and other services through dependency injection:

from starstreamer.services.economy import EconomyService
from starstreamer.services.users import UserService

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!profile"))
async def profile_command(
    event: Event,
    twitch: TwitchClient,
    economy: EconomyService,
    users: UserService,
    logger: logging.Logger
) -> None:
    """Show user profile with stats"""
    user = event.data.get("user", {})
    user_id = user.get("id")
    username = user.get("display_name")

    # Get user data from services
    balance = await economy.get_balance(user_id)
    user_data = await users.get_or_create_user(user_id, username)

    await twitch.send_message(
        f"@{username} | Balance: {balance} | "
        f"First seen: {user_data.created_at.strftime('%Y-%m-%d')}"
    )

    logger.info(f"Profile requested for {username}")

Command Arguments¶

Parse command arguments for more complex commands:

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!shoutout"))
async def shoutout_command(event: Event, twitch: TwitchClient) -> None:
    """Shoutout another streamer"""
    message = event.data.get("message", "")
    parts = message.split(maxsplit=1)

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

    # Extract target username (remove @ if present)
    target = parts[1].lstrip("@")

    await twitch.send_message(
        f"Check out @{target}! They stream amazing content! "
        f"https://twitch.tv/{target}"
    )

Combining Triggers¶

Use multiple triggers for more control:

from starstreamer.triggers import ModOnlyTrigger, SubscriberOnlyTrigger

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!modcommand"))
@trigger(ModOnlyTrigger())  # Only mods can use this
async def mod_only_command(event: Event, twitch: TwitchClient) -> None:
    """Command only moderators can use"""
    await twitch.send_message("Mod power activated! 🛡️")

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!subcommand"))
@trigger(SubscriberOnlyTrigger())  # Only subscribers can use this
async def sub_only_command(event: Event, twitch: TwitchClient) -> None:
    """Command only subscribers can use"""
    await twitch.send_message("Thanks for being a subscriber! 💜")

Tips for Writing Actions¶

Use Dependency Injection - Don't import services directly, use DI parameters
Handle Missing Data - Always use .get() with defaults for event data
Add Logging - Track important events with the logger service
Use Cooldowns - Prevent spam with CooldownTrigger
Keep It Simple - Start small and build up
Test Edge Cases - Handle missing users, empty messages, etc.

Common Patterns¶

Persistent Data with Database¶

from starstreamer.db.database import Database

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!count"))
async def count_command(
    event: Event,
    twitch: TwitchClient,
    database: Database
) -> None:
    """Persistent counter using database"""
    # Get or create counter
    result = await database.fetch_one(
        "SELECT value FROM counters WHERE name = ?",
        ("global_count",)
    )

    count = result["value"] if result else 0
    count += 1

    # Update counter
    await database.execute(
        "INSERT OR REPLACE INTO counters (name, value) VALUES (?, ?)",
        ("global_count", count)
    )

    await twitch.send_message(f"Counter: {count}")

Rate Limiting Per User¶

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!points"))
@trigger(CooldownTrigger(30, per_user=True))  # 30s per user
async def points_command(event: Event, twitch: TwitchClient) -> None:
    """Check points with per-user cooldown"""
    user = event.data.get("user", {})
    username = user.get("display_name")

    # This will only run once per 30 seconds per user
    await twitch.send_message(f"@{username} has 100 points!")

Troubleshooting¶

Module Not Loading¶

If your module doesn't load: 1. Check that it extends BaseModule 2. Verify it's in src/modules/yourmodule/ 3. Check that register_actions() imports your action files 4. Look for import errors in the logs

Commands Not Working¶

If commands don't respond: 1. Check the command trigger matches exactly (case-sensitive) 2. Verify the handler is decorated with @on_event("twitch.chat.message") 3. Check logs for any errors during execution 4. Ensure your OAuth token has chat permissions

Dependency Injection Errors¶

If you get injection errors: 1. Make sure parameter names match service names exactly 2. Common service names: twitch, database, logger, economy, users 3. Don't use type hints that conflict with service names

Next Steps¶

Now that you've created your first action:

📖 Learn about the Module System in detail
🎯 Explore Triggers and Filters
💉 Understand Dependency Injection
📊 Add Database Storage
🧪 Write Tests for your actions