Decorators API Reference

StarStreamer provides a comprehensive decorator system for registering event handlers and adding filtering, prioritization, and triggering capabilities to your actions.

Core Event Decorators

@on_event(event_type: str, **config)

on_event

on_event(event_type: str, **config: Any) -> Callable[[HandlerFunc], HandlerFunc]

Decorator to register a function as an event handler

Parameters:

Name	Type	Description	Default
event_type	str	The event type to listen for (e.g., "twitch.follow")	required
**config	Any	Optional configuration for this event type	{}

Source code in src/starstreamer/core/decorators.py

def on_event(event_type: str, **config: Any) -> Callable[[HandlerFunc], HandlerFunc]:
    """
    Decorator to register a function as an event handler

    Args:
        event_type: The event type to listen for (e.g., "twitch.follow")
        **config: Optional configuration for this event type
    """

    def decorator(func: HandlerFunc) -> HandlerFunc:
        func = _ensure_handler_metadata(func)

        # Add event type to the list
        if event_type not in func._event_types:  # type: ignore[attr-defined] # noqa: SLF001
            func._event_types.append(event_type)  # type: ignore[attr-defined] # noqa: SLF001

        # Store any config
        if config:
            func._event_configs[event_type] = config  # type: ignore[attr-defined] # noqa: SLF001

        # Try to register immediately
        _register_handler(func)

        return func

    # Handle both @on_event("event") and @on_event("event", config=value)
    if callable(event_type):
        # Called without parentheses - event_type is actually the function
        raise TypeError("on_event decorator requires an event type: @on_event('event.type')")

    return decorator

The primary decorator for registering event handlers. Every action must use this decorator to specify which events it responds to.

Parameters: - event_type: Event type string (e.g., "twitch.chat.message", "twitch.follow") - **config: Optional configuration for the event type

Basic Usage:

@on_event("twitch.chat.message")
async def handle_chat(event: Event, twitch: TwitchClient) -> None:
    """Handle all chat messages"""
    message = event.data.get("message", "")
    user = event.data.get("user", {})
    username = user.get("display_name", "Unknown")

    print(f"{username}: {message}")

Multiple Events:

@on_event("twitch.follow")
@on_event("twitch.subscription")
async def handle_engagement(event: Event, twitch: TwitchClient) -> None:
    """Handle both follows and subscriptions"""
    if event.type == "twitch.follow":
        username = event.data.get("user_name", "Unknown")
        await twitch.send_message(f"Thanks for following, {username}!")
    elif event.type == "twitch.subscription":
        username = event.data.get("user_name", "Unknown")
        await twitch.send_message(f"Thanks for subscribing, {username}!")

With Configuration:

@on_event("twitch.chat.message", timeout=30, retries=3)
async def important_handler(event: Event) -> None:
    """Handler with custom configuration"""
    # Configuration is stored and available to the event system
    pass

---

Filtering Decorators

@filter(filter_func)

filter

filter(filter_func: FilterFunc) -> Callable[[HandlerFunc], HandlerFunc]

Decorator to add a filter to an event handler

Parameters:

Name	Type	Description	Default
filter_func	FilterFunc	A function that takes an event and returns True/False	required

Source code in src/starstreamer/core/decorators.py

def filter(filter_func: FilterFunc) -> Callable[[HandlerFunc], HandlerFunc]:
    """
    Decorator to add a filter to an event handler

    Args:
        filter_func: A function that takes an event and returns True/False
    """

    def decorator(func: HandlerFunc) -> HandlerFunc:
        func = _ensure_handler_metadata(func)

        # Add filter to the list
        func._filters.append(filter_func)  # type: ignore[attr-defined] # noqa: SLF001

        # Don't re-register here, let on_event handle it

        return func

    # Support both @filter(lambda e: ...) and composed filters
    if not callable(filter_func):
        raise TypeError("filter decorator requires a callable")

    return decorator

Add custom filter logic to event handlers.

Parameters: - filter_func: Function that takes an Event and returns bool

Basic Usage:

def is_moderator(event: Event) -> bool:
    """Check if user is a moderator"""
    user = event.data.get("user", {})
    badges = user.get("badges", {})
    return "moderator" in badges or "broadcaster" in badges

@on_event("twitch.chat.message")
@filter(is_moderator)
async def mod_only_handler(event: Event, twitch: TwitchClient) -> None:
    """Only responds to moderator messages"""
    await twitch.send_message("Hello, moderator!")

Lambda Filters:

@on_event("twitch.chat.message")
@filter(lambda event: len(event.data.get("message", "")) > 50)
async def long_message_handler(event: Event) -> None:
    """Only handle messages longer than 50 characters"""
    print("Received a long message!")

Multiple Filters:

@on_event("twitch.chat.message")
@filter(lambda event: "!" in event.data.get("message", ""))
@filter(is_moderator)
async def mod_command_handler(event: Event) -> None:
    """Only handle command messages from moderators"""
    # Both filters must pass for this handler to run
    pass

Filter Helper Functions

StarStreamer provides pre-built filter functions for common conditions:

min_value(key: str, threshold: float)

@on_event("twitch.cheer")
@filter(min_value("bits", 100))
async def large_cheer_handler(event: Event) -> None:
    """Only handle cheers of 100+ bits"""
    pass

max_value(key: str, threshold: float)

@on_event("twitch.raid")
@filter(max_value("viewers", 10))
async def small_raid_handler(event: Event) -> None:
    """Only handle raids with 10 or fewer viewers"""
    pass

has_key(key: str)

@on_event("twitch.subscription.gift")
@filter(has_key("recipient_user_name"))
async def individual_gift_handler(event: Event) -> None:
    """Only handle individual gift subs (not anonymous gifts)"""
    pass

match_value(key: str, value: Any)

@on_event("twitch.subscription")
@filter(match_value("tier", "3000"))
async def tier3_sub_handler(event: Event) -> None:
    """Only handle Tier 3 subscriptions"""
    pass

---

Priority Decorator

@priority(level: int)

priority

priority(level: int) -> Callable[[HandlerFunc], HandlerFunc]

Decorator to set handler priority (lower number = higher priority)

Parameters:

Name	Type	Description	Default
level	int	Priority level (1-10, default 5)	required

Source code in src/starstreamer/core/decorators.py

def priority(level: int) -> Callable[[HandlerFunc], HandlerFunc]:
    """
    Decorator to set handler priority (lower number = higher priority)

    Args:
        level: Priority level (1-10, default 5)
    """

    def decorator(func: HandlerFunc) -> HandlerFunc:
        func = _ensure_handler_metadata(func)

        # Set priority
        func._priority = level  # type: ignore[attr-defined] # noqa: SLF001

        # Don't re-register here, let on_event handle it

        return func

    return decorator

Set handler execution priority (lower numbers execute first).

Parameters: - level: Priority level (1-10, default is 5)

Usage:

@on_event("twitch.chat.message")
@priority(1)  # High priority - runs first
async def anti_spam_handler(event: Event) -> None:
    """Check for spam before other handlers"""
    # Spam detection logic
    pass

@on_event("twitch.chat.message")
@priority(5)  # Normal priority
async def regular_handler(event: Event) -> None:
    """Regular message processing"""
    pass

@on_event("twitch.chat.message")
@priority(10)  # Low priority - runs last
async def logging_handler(event: Event) -> None:
    """Log message after all processing"""
    # Logging logic
    pass

Priority Guidelines: - 1-2: Security, anti-spam, critical validation - 3-4: Core functionality, essential commands - 5: Default priority for most handlers - 6-7: Secondary features, optional processing - 8-10: Logging, analytics, non-essential tasks

---

Trigger System

The trigger system provides specialized decorators for common event patterns, especially for chat commands and conditions.

@trigger(trigger_instance)

trigger

trigger(trigger_instance: Trigger) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Decorator to use a trigger instance

Example

my_trigger = CommandTrigger("hello")

@trigger(my_trigger) async def hello_handler(event, ctx): ...

Source code in src/starstreamer/triggers/base.py

def trigger(trigger_instance: Trigger) -> Callable[[Callable[..., Any]], Callable[..., Any]]:
    """
    Decorator to use a trigger instance

    Example:
        my_trigger = CommandTrigger("hello")

        @trigger(my_trigger)
        async def hello_handler(event, ctx):
            ...
    """
    return trigger_instance

Apply trigger-based filtering to event handlers.

Parameters: - trigger_instance: Instance of a trigger class

Command Triggers

CommandTrigger(command: str, **options)

from starstreamer.triggers import CommandTrigger, trigger

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!hello"))
async def hello_command(event: Event, twitch: TwitchClient) -> None:
    """Respond to !hello command"""
    user = event.data.get("user", {})
    username = user.get("display_name", "friend")
    await twitch.send_message(f"Hello {username}!")

# With options
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!ban", requires_mod=True))
async def ban_command(event: Event, twitch: TwitchClient) -> None:
    """Moderator-only ban command"""
    # Handle ban logic
    pass

Cooldown Triggers

CooldownTrigger(cooldown_seconds: float, per_user: bool = True, message: str = None)

from starstreamer.triggers import CooldownTrigger, CommandTrigger, trigger

@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 per-user cooldown"""
    # Work command logic
    pass

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!uptime"))
@trigger(CooldownTrigger(30, per_user=False))  # Global 30-second cooldown
async def uptime_command(event: Event, twitch: TwitchClient) -> None:
    """Uptime command with global cooldown"""
    # Uptime logic
    pass

Condition Triggers

Permission-Based Triggers

from starstreamer.triggers import (
    ModOnlyTrigger, SubscriberOnlyTrigger, VIPOnlyTrigger,
    CommandTrigger, trigger
)

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!timeout"))
@trigger(ModOnlyTrigger())
async def timeout_command(event: Event) -> None:
    """Moderator-only timeout command"""
    pass

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!subsong"))
@trigger(SubscriberOnlyTrigger())
async def subscriber_song_request(event: Event) -> None:
    """Subscriber-only song request"""
    pass

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!vip"))
@trigger(VIPOnlyTrigger())
async def vip_command(event: Event) -> None:
    """VIP-only command"""
    pass

Value-Based Triggers

from starstreamer.triggers import MinBitsTrigger, MinViewersTrigger, trigger

@on_event("twitch.cheer")
@trigger(MinBitsTrigger(1000))
async def mega_cheer_alert(event: Event, twitch: TwitchClient) -> None:
    """Special alert for 1000+ bit cheers"""
    bits = event.data.get("bits", 0)
    user = event.data.get("user_name", "Anonymous")
    await twitch.send_announcement(
        f"🚨 MEGA CHEER! {user} just cheered {bits} bits! 🚨",
        color="purple"
    )

@on_event("twitch.raid")
@trigger(MinViewersTrigger(50))
async def large_raid_handler(event: Event, twitch: TwitchClient) -> None:
    """Handle large raids (50+ viewers)"""
    raiders = event.data.get("viewers", 0)
    from_broadcaster = event.data.get("from_broadcaster_user_name", "Unknown")
    await twitch.send_announcement(
        f"🔥 MASSIVE RAID! {from_broadcaster} brought {raiders} viewers! 🔥",
        color="red"
    )

Advanced Trigger Patterns

Keyword Detection

from starstreamer.triggers import KeywordTrigger, trigger

@on_event("twitch.chat.message")
@trigger(KeywordTrigger("discord", "invite", case_sensitive=False))
async def discord_mention(event: Event, twitch: TwitchClient) -> None:
    """Respond when someone mentions Discord"""
    await twitch.send_message("Join our Discord: https://discord.gg/example")

Regex Patterns

from starstreamer.triggers import RegexTrigger, trigger
import re

@on_event("twitch.chat.message")
@trigger(RegexTrigger(r"!remind\s+(\d+)\s+(.+)", re.IGNORECASE))
async def reminder_command(event: Event, twitch: TwitchClient) -> None:
    """Parse reminder command with regex"""
    match = event.trigger_result.match  # Access regex match object
    if match:
        minutes = int(match.group(1))
        message = match.group(2)
        # Set up reminder logic
        pass

User-Specific Triggers

from starstreamer.triggers import UserTrigger, CommandTrigger, trigger

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!owner"))
@trigger(UserTrigger(["broadcaster_username"], mode="allow"))
async def owner_only_command(event: Event) -> None:
    """Command only the broadcaster can use"""
    pass

# Multiple users
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!admin"))
@trigger(UserTrigger(["user1", "user2", "user3"], mode="allow"))
async def admin_command(event: Event) -> None:
    """Command for specific admin users"""
    pass

# Exclude specific users
@on_event("twitch.chat.message")
@trigger(CommandTrigger("!public"))
@trigger(UserTrigger(["banned_user", "spam_bot"], mode="deny"))
async def public_command(event: Event) -> None:
    """Command available to everyone except banned users"""
    pass

---

Combining Decorators

Decorators can be combined in any order to create sophisticated handler behavior:

Basic Combination

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!play"))
@trigger(CooldownTrigger(60))
@filter(lambda event: "youtube.com" in event.data.get("message", ""))
@priority(3)
async def youtube_command(event: Event, twitch: TwitchClient) -> None:
    """Play YouTube videos with cooldown and URL validation"""
    message = event.data.get("message", "")
    # Extract and handle YouTube URL
    pass

Complex Permission System

def is_trusted_user(event: Event) -> bool:
    """Check if user is trusted (mod, VIP, or subscriber)"""
    user = event.data.get("user", {})
    badges = user.get("badges", {})
    return any(badge in badges for badge in ["moderator", "vip", "subscriber", "broadcaster"])

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!queue"))
@filter(is_trusted_user)
@trigger(CooldownTrigger(30, per_user=True))
@priority(4)
async def queue_command(event: Event, twitch: TwitchClient) -> None:
    """Queue management for trusted users only"""
    # Queue logic
    pass

Multi-Event Handler

@on_event("twitch.follow")
@on_event("twitch.subscription")
@on_event("twitch.cheer")
@filter(lambda event: event.data.get("user_name") not in ["StreamlabsBot", "Nightbot"])
@priority(2)
async def engagement_tracker(event: Event, analytics: AnalyticsService) -> None:
    """Track engagement events excluding bots"""
    user_name = event.data.get("user_name", "Unknown")
    await analytics.record_engagement(event.type, user_name)

---

Dependency Injection with Decorators

StarStreamer's decorator system works seamlessly with dependency injection:

Service Injection

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!balance"))
async def balance_command(
    event: Event,
    twitch: TwitchClient,        # Platform service
    economy: EconomyService,     # Business logic service
    users: UserService,          # Data access service
    logger: logging.Logger       # Utility service
) -> None:
    """Get user's currency balance with full DI"""
    user = event.data.get("user", {})
    user_id = user.get("id", "")
    username = user.get("display_name", "Unknown")

    # Use injected services
    balance = await economy.get_balance(user_id)
    await twitch.send_message(f"@{username} Your balance: {balance} coins")
    logger.info(f"Balance checked by {username}: {balance}")

Custom Service Registration

# In your main application setup
from starstreamer.core.decorators import setup_dependency_injection

container, registry = setup_dependency_injection()

# Register your services
container.register_singleton(MyCustomService, my_service_instance)
container.register_singleton(DatabaseService, db_service)

# Handlers can now inject these services
@on_event("twitch.chat.message")
async def handler(event: Event, custom: MyCustomService, db: DatabaseService) -> None:
    """Handler with custom service injection"""
    pass

---

Advanced Usage Patterns

Decorator Factories

def mod_command(command_name: str, cooldown: int = 60):
    """Factory for creating moderator-only commands with cooldown"""
    def decorator(func):
        return (
            on_event("twitch.chat.message")(
                trigger(CommandTrigger(command_name))(
                    trigger(ModOnlyTrigger())(
                        trigger(CooldownTrigger(cooldown))(
                            priority(3)(func)
                        )
                    )
                )
            )
        )
    return decorator

# Usage
@mod_command("!clear", cooldown=30)
async def clear_command(event: Event, twitch: TwitchClient) -> None:
    """Clear chat command"""
    await twitch.clear_chat()

Conditional Registration

import os

# Only register in development
if os.getenv("ENVIRONMENT") == "development":
    @on_event("twitch.chat.message")
    @trigger(CommandTrigger("!debug"))
    async def debug_command(event: Event, twitch: TwitchClient) -> None:
        """Development-only debug command"""
        await twitch.send_message("Debug info: ...")

Dynamic Handler Creation

def create_social_command(platform: str, url: str):
    """Dynamically create social media commands"""

    @on_event("twitch.chat.message")
    @trigger(CommandTrigger(f"!{platform}"))
    @trigger(CooldownTrigger(120))
    async def social_command(event: Event, twitch: TwitchClient) -> None:
        await twitch.send_message(f"Follow me on {platform.title()}: {url}")

    return social_command

# Create multiple social commands
twitter_cmd = create_social_command("twitter", "https://twitter.com/username")
youtube_cmd = create_social_command("youtube", "https://youtube.com/@channel")

---

Best Practices

Decorator Order

While decorators can be applied in any order, this conventional order improves readability:

@on_event("event.type")          # 1. Event registration (required)
@trigger(TriggerType())          # 2. Trigger conditions
@filter(filter_func)             # 3. Custom filters
@priority(level)                 # 4. Priority setting
async def handler(event, services) -> None:
    pass

Error Handling

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!risky"))
async def risky_command(event: Event, twitch: TwitchClient, logger: logging.Logger) -> None:
    """Command that might fail"""
    try:
        # Risky operation
        result = await some_external_api()
        await twitch.send_message(f"Success: {result}")
    except Exception as e:
        logger.error(f"Command failed: {e}")
        await twitch.send_message("Sorry, something went wrong!")

Type Safety

from typing import TYPE_CHECKING

if TYPE_CHECKING:
    from starstreamer.runtime.types import Event
    from starstreamer.plugins.twitch import TwitchClient

@on_event("twitch.chat.message")
@trigger(CommandTrigger("!typed"))
async def typed_handler(event: Event, twitch: TwitchClient) -> None:
    """Fully typed handler for better IDE support"""
    # Full type checking and autocomplete available
    message: str = event.data.get("message", "")
    await twitch.send_message(f"You said: {message}")

See Also

• Event System
• Triggers & Filters
• Dependency Injection
• Custom Commands Examples