Creating Custom Events

Custom events let you define domain-specific messages for your content pack. This guide covers how to create, emit, and handle custom events.

Defining Custom Events

Basic Event

Events are dataclasses that inherit from Event:

from dataclasses import dataclass
from uuid import UUID
from maid_engine.core.events import Event

@dataclass
class SpellCastEvent(Event):
    """Emitted when a spell is cast."""
    caster_id: UUID
    spell_name: str
    target_id: UUID | None = None

With Complex Types

from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from uuid import UUID
from maid_engine.core.events import Event

class DamageType(Enum):
    PHYSICAL = "physical"
    MAGICAL = "magical"
    FIRE = "fire"
    ICE = "ice"

@dataclass
class CombatDamageEvent(Event):
    """Emitted when combat damage is dealt."""
    attacker_id: UUID
    target_id: UUID
    base_damage: int
    damage_type: DamageType
    is_critical: bool = False
    modifiers: list[str] = field(default_factory=list)

With Computed Properties

@dataclass
class ItemTransferEvent(Event):
    """Emitted when an item is transferred between entities."""
    item_id: UUID
    from_entity_id: UUID | None
    to_entity_id: UUID | None
    quantity: int = 1

    @property
    def is_pickup(self) -> bool:
        """Item picked up from the ground."""
        return self.from_entity_id is None

    @property
    def is_drop(self) -> bool:
        """Item dropped to the ground."""
        return self.to_entity_id is None

    @property
    def is_trade(self) -> bool:
        """Item traded between entities."""
        return self.from_entity_id is not None and self.to_entity_id is not None

Event Inheritance

Base Events

Create base events for families of related events:

@dataclass
class CombatEvent(Event):
    """Base class for all combat-related events."""
    room_id: UUID
    tick: int

@dataclass
class AttackEvent(CombatEvent):
    """Attack initiated."""
    attacker_id: UUID
    target_id: UUID
    weapon_id: UUID | None = None

@dataclass
class DamageDealtEvent(CombatEvent):
    """Damage was dealt."""
    source_id: UUID
    target_id: UUID
    damage: int

@dataclass
class CombatEndedEvent(CombatEvent):
    """Combat ended."""
    winner_id: UUID | None = None
    loser_ids: list[UUID] = field(default_factory=list)

Subscribing to Base Events

Note: MAID's EventBus matches exact event types, not inheritance:

# This only matches AttackEvent, not CombatEvent handlers
await bus.emit(AttackEvent(...))

# To handle all combat events, subscribe to each type:
bus.subscribe(AttackEvent, handle_combat)
bus.subscribe(DamageDealtEvent, handle_combat)
bus.subscribe(CombatEndedEvent, handle_combat)

Registering Custom Events

Via Content Pack

Register events in your content pack:

class MyCombatPack:
    def get_events(self) -> list[type[Event]]:
        """Return custom event types."""
        return [
            AttackEvent,
            DamageDealtEvent,
            CombatEndedEvent,
            SpellCastEvent,
        ]

This helps with documentation and discovery - the engine knows what events your pack defines.

Emitting Custom Events

Basic Emission

from my_pack.events import SpellCastEvent

# In a system or command handler
async def cast_spell(caster_id: UUID, spell: str, target_id: UUID | None) -> None:
    # ... spell logic ...

    # Emit event
    await events.emit(SpellCastEvent(
        caster_id=caster_id,
        spell_name=spell,
        target_id=target_id,
    ))

Conditional Emission

async def process_attack(attacker: Entity, target: Entity) -> None:
    damage = calculate_damage(attacker, target)

    # Emit damage event
    await events.emit(DamageDealtEvent(
        room_id=get_room_id(attacker),
        tick=current_tick,
        source_id=attacker.id,
        target_id=target.id,
        damage=damage,
    ))

    # Conditionally emit death event
    health = target.get(HealthComponent)
    if health.current <= 0:
        await events.emit(EntityDiedEvent(
            entity_id=target.id,
            killer_id=attacker.id,
        ))

Chained Events

async def complete_quest(player_id: UUID, quest_id: str) -> None:
    # Emit quest completion
    await events.emit(QuestCompletedEvent(
        player_id=player_id,
        quest_id=quest_id,
    ))

    # Check for achievements (handlers will emit achievement events)
    # Check for level up (handlers will emit level up events)
    # The event chain handles cascading effects

Event Design Patterns

Request-Response Pattern

Use events for requests that might be modified or cancelled:

@dataclass
class HealRequestEvent(Event):
    """Request to heal an entity. Can be modified or cancelled."""
    target_id: UUID
    amount: int
    source_id: UUID | None = None

    # Mutable - handlers can modify
    final_amount: int = 0

# High priority handler - apply healing modifiers
async def apply_healing_bonus(event: HealRequestEvent) -> None:
    target = world.get_entity(event.target_id)
    bonus = target.try_get(HealingBonusComponent)
    if bonus:
        event.final_amount = int(event.amount * bonus.multiplier)
    else:
        event.final_amount = event.amount

# Normal priority handler - apply the healing
async def apply_healing(event: HealRequestEvent) -> None:
    if event.cancelled:
        return

    target = world.get_entity(event.target_id)
    health = target.get(HealthComponent)
    health.current = min(health.maximum, health.current + event.final_amount)

Notification Pattern

Events that inform but don't require action:

@dataclass
class PlayerLeveledUpEvent(Event):
    """Notification that a player leveled up."""
    player_id: UUID
    old_level: int
    new_level: int

# Handlers can react but not modify the fact
async def announce_level_up(event: PlayerLeveledUpEvent) -> None:
    room_id = get_player_room(event.player_id)
    await broadcast_to_room(
        room_id,
        f"Player leveled up to {event.new_level}!"
    )

async def award_level_up_bonus(event: PlayerLeveledUpEvent) -> None:
    player = world.get_entity(event.player_id)
    # Award stat points, etc.

Aggregation Pattern

Collect data from multiple handlers:

@dataclass
class DamageCalculationEvent(Event):
    """Calculate total damage with modifiers."""
    attacker_id: UUID
    target_id: UUID
    base_damage: int
    modifiers: dict[str, int] = field(default_factory=dict)

    @property
    def total_damage(self) -> int:
        return max(0, self.base_damage + sum(self.modifiers.values()))

# Each handler adds its modifier
async def apply_strength_bonus(event: DamageCalculationEvent) -> None:
    attacker = world.get_entity(event.attacker_id)
    stats = attacker.try_get(StatsComponent)
    if stats:
        event.modifiers["strength"] = stats.strength // 2

async def apply_weapon_bonus(event: DamageCalculationEvent) -> None:
    attacker = world.get_entity(event.attacker_id)
    weapon = get_equipped_weapon(attacker)
    if weapon:
        event.modifiers["weapon"] = weapon.damage_bonus

async def apply_armor_reduction(event: DamageCalculationEvent) -> None:
    target = world.get_entity(event.target_id)
    armor = target.try_get(ArmorComponent)
    if armor:
        event.modifiers["armor"] = -armor.defense

# Usage
event = DamageCalculationEvent(
    attacker_id=attacker.id,
    target_id=target.id,
    base_damage=10,
)
await events.emit(event)
actual_damage = event.total_damage  # Includes all modifier contributions

Organizing Events

File Structure

my-content-pack/
└── src/
    └── my_pack/
        ├── __init__.py
        ├── pack.py
        └── events/
            ├── __init__.py
            ├── combat.py    # Combat events
            ├── items.py     # Item events
            ├── quests.py    # Quest events
            └── social.py    # Social events

Event Module

# my_pack/events/combat.py
from dataclasses import dataclass, field
from uuid import UUID
from maid_engine.core.events import Event

@dataclass
class AttackEvent(Event):
    """Player or NPC attacks a target."""
    attacker_id: UUID
    target_id: UUID
    skill_name: str = "basic_attack"

@dataclass
class DamageEvent(Event):
    """Damage is dealt to a target."""
    target_id: UUID
    damage: int
    damage_type: str = "physical"
    source_id: UUID | None = None

@dataclass
class DeathEvent(Event):
    """An entity dies."""
    entity_id: UUID
    killer_id: UUID | None = None

Package Exports

# my_pack/events/__init__.py
from .combat import AttackEvent, DamageEvent, DeathEvent
from .items import ItemPickedUpEvent, ItemDroppedEvent
from .quests import QuestStartedEvent, QuestCompletedEvent

__all__ = [
    # Combat
    "AttackEvent",
    "DamageEvent",
    "DeathEvent",
    # Items
    "ItemPickedUpEvent",
    "ItemDroppedEvent",
    # Quests
    "QuestStartedEvent",
    "QuestCompletedEvent",
]

Documentation

Documenting Events

@dataclass
class CriticalHitEvent(Event):
    """Emitted when a critical hit occurs in combat.

    This event is emitted after damage calculation but before
    damage application. Handlers can:
    - Modify the critical multiplier
    - Add special effects
    - Cancel to prevent the critical (damage still applies normally)

    Attributes:
        attacker_id: Entity that scored the critical hit
        target_id: Entity receiving the critical hit
        base_damage: Damage before critical multiplier
        multiplier: Critical damage multiplier (default 2.0)

    Example:
        async def on_critical(event: CriticalHitEvent) -> None:
            # Apply assassin bonus
            if has_skill(event.attacker_id, "assassin"):
                event.multiplier = 3.0

    Related Events:
        - DamageEvent: Emitted after critical is applied
        - DeathEvent: May be emitted if target dies
    """
    attacker_id: UUID
    target_id: UUID
    base_damage: int
    multiplier: float = 2.0

    @property
    def total_damage(self) -> int:
        """Damage after critical multiplier."""
        return int(self.base_damage * self.multiplier)

Best Practices

1. Use Clear, Descriptive Names

# Good - clear what happened
class PlayerJoinedGuildEvent(Event): ...
class ItemEnchantedEvent(Event): ...
class CombatRoundEndedEvent(Event): ...

# Bad - ambiguous
class GuildEvent(Event): ...  # What about the guild?
class ItemEvent(Event): ...   # What happened to the item?

2. Include All Necessary Context

# Good - all context included
@dataclass
class TradeCompletedEvent(Event):
    trader_a_id: UUID
    trader_b_id: UUID
    items_a_to_b: list[UUID]
    items_b_to_a: list[UUID]
    gold_a_to_b: int
    gold_b_to_a: int
    room_id: UUID
    timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))

# Bad - missing context
@dataclass
class TradeCompletedEvent(Event):
    trade_id: UUID  # What was traded? Who traded?

3. Keep Events Immutable Where Possible

# Good - mostly immutable, explicit mutable fields
@dataclass
class DamageCalculationEvent(Event):
    # Immutable
    attacker_id: UUID
    target_id: UUID
    base_damage: int

    # Explicitly mutable for modifier pattern
    modifiers: dict[str, int] = field(default_factory=dict)

4. Use Optional Fields with Defaults

@dataclass
class ItemCraftedEvent(Event):
    crafter_id: UUID
    item_id: UUID
    recipe_id: str
    # Optional context
    quality: str = "normal"
    bonus_stats: dict[str, int] = field(default_factory=dict)
    critical_craft: bool = False

5. Test Your Events

def test_damage_event_creation():
    event = DamageEvent(
        target_id=UUID("12345678-1234-1234-1234-123456789abc"),
        damage=50,
    )

    assert event.damage == 50
    assert event.source_id is None
    assert event.event_type == "DamageEvent"
    assert not event.cancelled

def test_damage_event_cancellation():
    event = DamageEvent(
        target_id=UUID("12345678-1234-1234-1234-123456789abc"),
        damage=50,
    )

    event.cancel()

    assert event.cancelled

Next Steps

• Event Overview - Understanding the event system
• Writing Event Handlers - Handler patterns