Skip to content

maid-classic-rpg

The maid-classic-rpg package provides a complete classic MUD gameplay experience, building on top of maid-engine and maid-stdlib.

Installation

Install via uv sync from the monorepo root. Not yet published to PyPI.

uv sync

Module Overview

Systems (maid_classic_rpg.systems)

Game systems implementing classic MUD mechanics:

System Module Description
combat Turn-based tactical combat
magic Spellcasting and magical effects
crafting Item creation and recipes
economy Shops, trading, banking
quests Quest objectives and rewards
quests.generation Automated quest generation integration
social Guilds, factions, achievements
pvp Arenas, duels, rankings
world Time, weather, ecosystems
npc NPC behavior and AI dialogue
npc.autonomy NPC autonomy (needs, goals, schedules, social, barks)
skills Character skills and progression

Commands (maid_classic_rpg.commands)

Game-specific commands:

Category Commands
Combat attack, defend, flee, cast, use
Economy buy, sell, trade, bank, auction
Social guild, faction, party, tell, shout
Dialogue talk, ask, greet, conversations
Crafting craft, gather, recipe
Quests quest, objectives, abandon
PvP duel, arena, rankings
Debug @debug_brain — inspect NPC autonomy state (admin only)

Models (maid_classic_rpg.models)

Data models for game entities:

Module Description
combat Combat-related data structures
magic Spell definitions and effects
economy Shop, trade, and currency models
crafting Recipes and materials
social Guild and faction structures
npc NPC configuration and dialogue
skills Skill definitions and trees
world Time, weather, terrain

Events (maid_classic_rpg.events)

Game-specific events:

Module Description
combat Combat-related events
economy Economy/trade events
quests Quest progression events
autonomy NPC autonomy events (need changes, goal updates, schedule transitions, social interactions)

Components (maid_classic_rpg.components)

Game-specific ECS components:

Component Description
ClassComponent Character class (Warrior, Mage, etc.)
SkillsComponent Character skills and levels
QuestLogComponent Active and completed quests
ReputationComponent Faction standings
CraftingComponent Known recipes and materials

Systems

Combat System

Turn-based tactical combat with positioning and abilities.

from maid_classic_rpg.systems.combat import CombatSystem

# Combat system handles:
# - Initiative and turn order
# - Attack resolution
# - Damage calculation
# - Status effects
# - Death and respawn

combat

Combat systems for melee and ranged combat.

BaseCombatSystem

Bases: System, ABC

Abstract base class for combat systems.

Provides structure for combat flow: engagement, turn processing, damage calculation, and disengagement.

start_combat abstractmethod async

start_combat(attacker_id: UUID, defender_id: UUID) -> bool

Initiate combat between two entities.

end_combat abstractmethod async

end_combat(entity_id: UUID, reason: str = 'normal') -> bool

End combat for an entity.

process_attack abstractmethod async

process_attack(
    attacker: Entity,
    defender: Entity,
    weapon_id: UUID | None = None,
) -> tuple[bool, int]

Process an attack action.

calculate_damage abstractmethod async

calculate_damage(
    attacker: Entity, defender: Entity, base_damage: int
) -> int

Calculate final damage after modifiers.

is_in_combat

is_in_combat(entity_id: UUID) -> bool

Check if an entity is currently in combat.

BodyDamageResult

BodyDamageResult(
    actual_damage: int = 0,
    part_destroyed: bool = False,
    effects: list[StatusEffectType] | None = None,
    vital_hit: bool = False,
)

Result of applying damage to a body part.

is_fatal property

is_fatal: bool

Check if this damage was fatal (vital part destroyed).

BodySystem

BodySystem(world: World)

Bases: System

System for managing body parts and location-based damage.

Handles: - Applying damage to specific body parts - Tracking body part health - Determining effects when parts are damaged/destroyed - Body part regeneration

update async

update(delta: float) -> None

Update tick - handle regeneration and effect ticks.

apply_damage_to_location

apply_damage_to_location(
    body: BodyComponent,
    location: str,
    damage: int,
    armor_value: int = 0,
) -> BodyDamageResult

Apply damage to a specific body location.

Parameters:

Name Type Description Default
body BodyComponent

The BodyComponent to damage

required
location str

Name of the body part

required
damage int

Amount of damage to apply

required
armor_value int

Additional armor to apply (on top of part's armor)

0

Returns:

Type Description
BodyDamageResult

BodyDamageResult with damage details

heal_location

heal_location(
    body: BodyComponent, location: str, amount: int
) -> int

Heal a specific body location.

Parameters:

Name Type Description Default
body BodyComponent

The BodyComponent to heal

required
location str

Name of the body part

required
amount int

Amount of healing

required

Returns:

Type Description
int

Actual amount healed

restore_part

restore_part(body: BodyComponent, location: str) -> bool

Restore a destroyed body part to minimal health.

Parameters:

Name Type Description Default
body BodyComponent

The BodyComponent

required
location str

Name of the body part to restore

required

Returns:

Type Description
bool

True if part was restored

get_total_health

get_total_health(body: BodyComponent) -> tuple[int, int]

Get total health across all body parts.

Parameters:

Name Type Description Default
body BodyComponent

The BodyComponent

required

Returns:

Type Description
tuple[int, int]

Tuple of (current_health, max_health)

get_health_percentage

get_health_percentage(body: BodyComponent) -> float

Get overall health percentage.

Parameters:

Name Type Description Default
body BodyComponent

The BodyComponent

required

Returns:

Type Description
float

Health percentage (0-100)

is_vital_destroyed

is_vital_destroyed(body: BodyComponent) -> bool

Check if any vital body part is destroyed.

Parameters:

Name Type Description Default
body BodyComponent

The BodyComponent

required

Returns:

Type Description
bool

True if a vital part is destroyed (entity should be dead)

get_status_summary

get_status_summary(body: BodyComponent) -> dict[str, str]

Get a status summary of all body parts.

Parameters:

Name Type Description Default
body BodyComponent

The BodyComponent

required

Returns:

Type Description
dict[str, str]

Dict mapping part name to status string

add_part_effect

add_part_effect(
    body: BodyComponent, location: str, effect: str
) -> bool

Add a status effect to a body part.

Parameters:

Name Type Description Default
body BodyComponent

The BodyComponent

required
location str

Name of the body part

required
effect str

Effect to add

required

Returns:

Type Description
bool

True if effect was added

remove_part_effect

remove_part_effect(
    body: BodyComponent, location: str, effect: str
) -> bool

Remove a status effect from a body part.

Parameters:

Name Type Description Default
body BodyComponent

The BodyComponent

required
location str

Name of the body part

required
effect str

Effect to remove

required

Returns:

Type Description
bool

True if effect was removed

clear_part_effects

clear_part_effects(
    body: BodyComponent, location: str
) -> int

Clear all status effects from a body part.

Parameters:

Name Type Description Default
body BodyComponent

The BodyComponent

required
location str

Name of the body part

required

Returns:

Type Description
int

Number of effects cleared

DamageSystem

DamageSystem(world: World)

Bases: System

System for applying damage from combat to characters.

Handles: - Applying attack results to character vitals - Location-based damage to body parts - Status effect application - Death determination

startup async

startup() -> None

Subscribe to entity destruction events for cache cleanup.

update async

update(delta: float) -> None

Update tick - check for deaths from damage-over-time effects.

DOT damage is applied by StatusEffectManager via apply_direct_damage(). This method checks for entities that died from DOT since last tick.

shutdown async

shutdown() -> None

Clean up body component cache and unsubscribe from events.

remove_body

remove_body(character_id: UUID) -> None

Remove a character's body component from the cache.

Call when an entity is destroyed to prevent unbounded cache growth.

Parameters:

Name Type Description Default
character_id UUID

The character UUID to remove.

required

get_or_create_body

get_or_create_body(character: Character) -> BodyComponent

Get or create a body component for a character.

Parameters:

Name Type Description Default
character Character

The character

required

Returns:

Type Description
BodyComponent

BodyComponent for the character

apply_attack_result async

apply_attack_result(
    result: AttackResult, target: Character
) -> bool

Apply an attack result to a target character.

Parameters:

Name Type Description Default
result AttackResult

The attack result to apply

required
target Character

The target character

required

Returns:

Type Description
bool

True if target was killed

apply_direct_damage

apply_direct_damage(
    target: Character,
    damage: int,
    _damage_type: DamageType = TRUE,
    location: str | None = None,
) -> int

Apply direct damage to a target (no attack roll).

Parameters:

Name Type Description Default
target Character

The target character

required
damage int

Amount of damage

required
_damage_type DamageType

Type of damage (for future resistance calculations)

TRUE
location str | None

Optional body location

None

Returns:

Type Description
int

Actual damage dealt

heal_character

heal_character(
    target: Character,
    amount: int,
    location: str | None = None,
) -> int

Heal a character.

Parameters:

Name Type Description Default
target Character

The character to heal

required
amount int

Amount to heal

required
location str | None

Optional specific body part to heal

None

Returns:

Type Description
int

Actual amount healed

is_alive

is_alive(character: Character) -> bool

Check if a character is alive.

Parameters:

Name Type Description Default
character Character

The character to check

required

Returns:

Type Description
bool

True if character is alive

get_health_status

get_health_status(character: Character) -> str

Get a description of character's health status.

Parameters:

Name Type Description Default
character Character

The character

required

Returns:

Type Description
str

Health status string

get_body_status

get_body_status(character: Character) -> dict[str, str]

Get detailed body part status for a character.

Parameters:

Name Type Description Default
character Character

The character

required

Returns:

Type Description
dict[str, str]

Dict mapping body part to status

StatusEffectManager

StatusEffectManager(world: World)

Bases: System

System for managing status effects on entities.

Handles: - Adding/removing status effects - Effect duration tracking - Effect tick processing (damage over time, etc.) - Effect stacking rules

shutdown async

shutdown() -> None

Clean up effect tracking to prevent memory leak.

update async

update(delta: float) -> None

Process effect ticks and expirations.

add_effect

add_effect(
    target: Character,
    effect_type: StatusEffectType,
    source_id: UUID | None = None,
    duration: float | None = None,
    intensity: float = 1.0,
    stacks: int = 1,
) -> StatusEffect

Add a status effect to a character.

Parameters:

Name Type Description Default
target Character

Character to affect

required
effect_type StatusEffectType

Type of effect

required
source_id UUID | None

Entity that caused the effect

None
duration float | None

Override default duration

None
intensity float

Effect intensity multiplier

1.0
stacks int

Number of stacks to apply

1

Returns:

Type Description
StatusEffect

The created or updated StatusEffect

remove_effect

remove_effect(
    target: Character, effect_type: StatusEffectType
) -> bool

Remove a status effect from a character.

Parameters:

Name Type Description Default
target Character

Character to affect

required
effect_type StatusEffectType

Type of effect to remove

required

Returns:

Type Description
bool

True if effect was found and removed

remove_all_effects

remove_all_effects(target: Character) -> int

Remove all effects from a character.

Parameters:

Name Type Description Default
target Character

Character to clear

required

Returns:

Type Description
int

Number of effects removed

get_effects

get_effects(target: Character) -> list[StatusEffect]

Get all active effects on a character.

Parameters:

Name Type Description Default
target Character

Character to check

required

Returns:

Type Description
list[StatusEffect]

List of active effects (copy)

has_effect

has_effect(
    target: Character, effect_type: StatusEffectType
) -> bool

Check if character has a specific effect.

Parameters:

Name Type Description Default
target Character

Character to check

required
effect_type StatusEffectType

Effect type to look for

required

Returns:

Type Description
bool

True if effect is active

get_effect_stacks

get_effect_stacks(
    target: Character, effect_type: StatusEffectType
) -> int

Get the number of stacks of an effect.

Parameters:

Name Type Description Default
target Character

Character to check

required
effect_type StatusEffectType

Effect type

required

Returns:

Type Description
int

Number of stacks (0 if not present)

get_effect_modifier

get_effect_modifier(
    target: Character, effect_type: StatusEffectType
) -> float

Get the modifier value from an active effect.

Parameters:

Name Type Description Default
target Character

Character to check

required
effect_type StatusEffectType

Effect type

required

Returns:

Type Description
float

Effect intensity * stacks (0.0 if not present)

MeleeCombatSystem

MeleeCombatSystem(world: World)

Bases: System

System for resolving melee combat attacks.

Handles attack resolution including: - Accuracy calculation with all modifiers - Hit location selection - Damage calculation with modifiers - Critical hits - Dodge/block/parry mechanics

update async

update(delta: float) -> None

Update tick - melee combat is resolved immediately, not per-tick.

resolve_attack async

resolve_attack(
    attacker: Character,
    target: Character,
    weapon: Item | None,
    attacker_position: CombatPosition | None = None,
    target_position: CombatPosition | None = None,
    attacker_zone: CombatZone = MELEE,
    target_zone: CombatZone = MELEE,
    attacker_size: CreatureSize = MEDIUM,
    target_size: CreatureSize = MEDIUM,
    weapon_size: CreatureSize | None = None,
    _other_attackers: list[CombatPosition] | None = None,
    aimed_location: str | None = None,
    attacker_grid_state: CombatantGridState | None = None,
    target_grid_state: CombatantGridState | None = None,
) -> AttackResult

Resolve a melee attack.

Parameters:

Name Type Description Default
attacker Character

Character making the attack

required
target Character

Character being attacked

required
weapon Item | None

Weapon being used (None for unarmed)

required
attacker_position CombatPosition | None

Tactical position of attacker (legacy 3x3 grid)

None
target_position CombatPosition | None

Tactical position of target (legacy 3x3 grid)

None
attacker_zone CombatZone

Combat zone of attacker

MELEE
target_zone CombatZone

Combat zone of target

MELEE
attacker_size CreatureSize

Size of the attacker creature

MEDIUM
target_size CreatureSize

Size of the target creature

MEDIUM
weapon_size CreatureSize | None

Absolute size of the weapon (if different from wielder)

None
_other_attackers list[CombatPosition] | None

Positions of other attackers (for flanking, legacy)

None
aimed_location str | None

Specific body part to target (optional)

None
attacker_grid_state CombatantGridState | None

New variable-grid position/facing state

None
target_grid_state CombatantGridState | None

New variable-grid position/facing state for target

None

Returns:

Type Description
AttackResult

AttackResult with full attack outcome

RangedCombatSystem

RangedCombatSystem(world: World)

Bases: System

System for resolving ranged combat attacks.

Handles attacks across room boundaries including: - Line of sight checks - Range/distance modifiers - Ranged weapon mechanics - Cover system - Ammunition tracking - Reload mechanics - Moving target penalties

Note: Facing-based bonuses (backstab, flanking) do NOT apply to ranged attacks. Those bonuses only apply to melee combat.

update async

update(delta: float) -> None

Update tick - process reload timers.

shutdown async

shutdown() -> None

Clean up reload timers and ammo tracking to prevent memory leak.

resolve_attack async

resolve_attack(
    attacker: Character,
    target: Character,
    weapon: Item,
    distance_rooms: int = 0,
    attacker_zone: CombatZone = RANGED_NEAR,
    target_zone: CombatZone = MELEE,
    line_of_sight: LineOfSight | None = None,
    target_in_cover: bool = False,
    cover_level: float = 0.0,
    target_is_moving: bool = False,
    aimed_location: str | None = None,
) -> AttackResult

Resolve a ranged attack.

Parameters:

Name Type Description Default
attacker Character

Character making the attack

required
target Character

Character being attacked

required
weapon Item

Ranged weapon being used

required
distance_rooms int

Number of rooms between attacker and target

0
attacker_zone CombatZone

Combat zone of attacker

RANGED_NEAR
target_zone CombatZone

Combat zone of target

MELEE
line_of_sight LineOfSight | None

Line of sight info between positions

None
target_in_cover bool

Whether target has cover

False
cover_level float

Cover percentage (0.0-1.0, 1.0 = full cover)

0.0
target_is_moving bool

Whether target moved this tick

False
aimed_location str | None

Specific body part to target

None

Returns:

Type Description
AttackResult

AttackResult with full attack outcome

get_reload_remaining

get_reload_remaining(attacker_id: UUID) -> float

Get remaining reload time for an attacker.

is_reloading

is_reloading(attacker_id: UUID) -> bool

Check if attacker is currently reloading.

force_reload

force_reload(
    attacker_id: UUID, weapon_type: RangedWeaponType
) -> None

Manually trigger a reload (for reload command).

cancel_reload

cancel_reload(attacker_id: UUID) -> bool

Cancel an ongoing reload.

Magic System

Spellcasting with mana costs, cooldowns, and magical effects.

from maid_classic_rpg.systems.magic import MagicSystem

# Magic system handles:
# - Spell learning and memorization
# - Mana management
# - Spell effects and duration
# - Magical item enchantments

magic

Magic systems for spellcasting and effects.

ActiveBuff dataclass

ActiveBuff(
    id: UUID,
    name: str,
    source_id: UUID | None,
    target_id: UUID,
    expires_at: float,
    effect_type: str,
    stat_modifiers: dict[str, int] = dict(),
    damage_reduction: float = 0.0,
    damage_increase: float = 0.0,
    status_immunity: list[StatusEffectType] = list(),
    tick_effect: str | None = None,
    tick_interval: float = 1.0,
    last_tick: float = 0.0,
    tick_damage: int = 0,
    tick_heal: int = 0,
    tick_damage_type: DamageType | None = None,
    stacks: int = 1,
    max_stacks: int = 1,
    is_buff: bool = True,
    can_dispel: bool = True,
    icon: str = "",
)

A temporary buff/debuff on a character.

BuffAppliedEvent dataclass

BuffAppliedEvent(
    target_id: UUID,
    buff_name: str,
    source_id: UUID | None = None,
    duration: float = 0.0,
)

Bases: Event

Emitted when a buff is applied.

BuffDispelledEvent dataclass

BuffDispelledEvent(
    target_id: UUID,
    buff_name: str,
    dispeller_id: UUID | None = None,
)

Bases: Event

Emitted when a buff is dispelled.

BuffExpiredEvent dataclass

BuffExpiredEvent(target_id: UUID, buff_name: str)

Bases: Event

Emitted when a buff expires.

SpellEffectSystem

SpellEffectSystem(world: World)

Bases: System

Manages active spell effects, buffs, and debuffs.

update async

update(delta: float) -> None

Process buff expirations and tick effects.

register_handler

register_handler(
    effect_type: str, handler: EffectHandler
) -> None

Register a custom effect handler.

apply_effect async

apply_effect(
    caster: Character,
    target: Character,
    effect_type: str,
    params: dict[str, Any],
) -> dict[str, Any]

Apply a spell effect.

Parameters:

Name Type Description Default
caster Character

Character casting the spell

required
target Character

Target of the effect

required
effect_type str

Type of effect to apply

required
params dict[str, Any]

Effect parameters

required

Returns:

Type Description
dict[str, Any]

Result dictionary with effect outcome

add_buff

add_buff(target_id: UUID, buff: ActiveBuff) -> bool

Add a buff to a character.

Returns True if buff was added, False if stacking failed.

remove_buff

remove_buff(target_id: UUID, buff_name: str) -> bool

Remove a buff by name. Returns True if removed.

get_buffs

get_buffs(target_id: UUID) -> list[ActiveBuff]

Get all active buffs on a target.

get_buff

get_buff(
    target_id: UUID, buff_name: str
) -> ActiveBuff | None

Get a specific buff by name.

has_buff

has_buff(target_id: UUID, buff_name: str) -> bool

Check if target has a specific buff.

get_stat_modifier

get_stat_modifier(target_id: UUID, stat: str) -> int

Get total stat modifier from all buffs.

get_damage_reduction

get_damage_reduction(target_id: UUID) -> float

Get total damage reduction from all buffs (0.0-1.0).

get_damage_increase

get_damage_increase(target_id: UUID) -> float

Get total damage increase from all buffs (capped at 5.0 / 500%).

has_status_immunity

has_status_immunity(
    target_id: UUID, status: StatusEffectType
) -> bool

Check if target is immune to a status effect.

dispel async

dispel(
    target_id: UUID,
    dispeller_id: UUID | None = None,
    count: int = 1,
    buffs_only: bool = False,
    debuffs_only: bool = False,
) -> int

Dispel buffs/debuffs from a target.

Parameters:

Name Type Description Default
target_id UUID

Character to dispel

required
dispeller_id UUID | None

Who is dispelling

None
count int

Max effects to remove

1
buffs_only bool

Only remove positive effects

False
debuffs_only bool

Only remove negative effects

False

Returns:

Type Description
int

Number of effects dispelled

clear_buffs

clear_buffs(target_id: UUID) -> int

Remove all buffs from a target. Returns count removed.

SpellCastCompleteEvent dataclass

SpellCastCompleteEvent(
    caster_id: UUID,
    spell_name: str,
    target_id: UUID | None = None,
    success: bool = True,
    damage_dealt: int = 0,
    healing_done: int = 0,
)

Bases: Event

Emitted when a spell finishes casting.

SpellCastInterruptedEvent dataclass

SpellCastInterruptedEvent(
    caster_id: UUID, spell_name: str, reason: str = ""
)

Bases: Event

Emitted when a spell cast is interrupted.

SpellCastStartEvent dataclass

SpellCastStartEvent(
    caster_id: UUID,
    spell_name: str,
    target_id: UUID | None = None,
    cast_time: float = 0.0,
)

Bases: Event

Emitted when a spell cast begins.

SpellLearnedEvent dataclass

SpellLearnedEvent(character_id: UUID, spell_name: str)

Bases: Event

Emitted when a character learns a spell.

CastingState dataclass

CastingState(
    caster_id: UUID,
    spell_name: str,
    target_id: UUID | None,
    started_at: float,
    completes_at: float,
    room_id: UUID | None,
    mana_cost: int = 0,
    cooldown: float = 0.0,
)

Tracks an in-progress spell cast.

SpellCastResult dataclass

SpellCastResult(
    success: bool = False,
    message: str = "",
    damage_dealt: int = 0,
    healing_done: int = 0,
    effects_applied: list[StatusEffectType] = list(),
    targets_hit: list[UUID] = list(),
    mana_spent: int = 0,
)

Result of attempting to cast a spell.

SpellSystem

SpellSystem(world: World)

Bases: System, StorageAware

Handles spell casting, effects, and cooldowns.

set_storage

set_storage(store: DocumentStore) -> None

Set the document store for persistence.

startup async

startup() -> None

Load spell data from storage.

shutdown async

shutdown() -> None

Save spell data and clean up.

update async

update(delta: float) -> None

Process casting completions and clean up expired cooldowns.

start_cast async

start_cast(
    caster: Character,
    spell_name: str,
    target: Character | None = None,
) -> tuple[bool, str]

Begin casting a spell.

Parameters:

Name Type Description Default
caster Character

Character casting the spell

required
spell_name str

Internal name of the spell

required
target Character | None

Optional target character

None

Returns:

Type Description
tuple[bool, str]

Tuple of (success, message)

interrupt_cast async

interrupt_cast(
    caster_id: UUID, reason: str = "interrupted"
) -> bool

Interrupt an in-progress cast.

Refunds mana spent when cast was initiated.

Parameters:

Name Type Description Default
caster_id UUID

ID of caster to interrupt

required
reason str

Reason for interruption

'interrupted'

Returns:

Type Description
bool

True if a cast was interrupted

is_casting

is_casting(caster_id: UUID) -> bool

Check if a character is currently casting.

get_casting_state

get_casting_state(caster_id: UUID) -> CastingState | None

Get the current casting state for a character.

learn_spell

learn_spell(character_id: UUID, spell_name: str) -> bool

Teach a spell to a character.

Parameters:

Name Type Description Default
character_id UUID

Character to teach

required
spell_name str

Spell internal name

required

Returns:

Type Description
bool

True if spell was learned (not already known)

forget_spell

forget_spell(character_id: UUID, spell_name: str) -> bool

Remove a spell from a character's known spells.

Parameters:

Name Type Description Default
character_id UUID

Character to forget spell

required
spell_name str

Spell internal name

required

Returns:

Type Description
bool

True if spell was removed

get_known_spells

get_known_spells(character_id: UUID) -> list[LearnedSpell]

Get all spells known by a character.

get_known_spell

get_known_spell(
    character_id: UUID, spell_name: str
) -> LearnedSpell | None

Get a specific learned spell.

get_cooldown_remaining

get_cooldown_remaining(
    character_id: UUID, spell_name: str
) -> float

Get remaining cooldown time in game seconds.

register_magic_commands

register_magic_commands(
    spell_system: SpellSystem, skill_system: SkillSystem
) -> None

Register all magic and ability commands.

Crafting System

Item creation with recipes, materials, and skill requirements.

from maid_classic_rpg.systems.crafting import CraftingSystem

# Crafting system handles:
# - Recipe discovery and learning
# - Material gathering
# - Crafting skill progression
# - Quality and variation

crafting

Crafting system components.

CraftingResult dataclass

CraftingResult(
    success: bool,
    error: str = "",
    item: Item | None = None,
    quality: MaterialQuality = COMMON,
    skill_gained: float = 0.0,
    materials_lost: list[UUID] = list(),
)

Result of a crafting attempt.

CraftingSystem

CraftingSystem(world: World)

Bases: System

Manages all crafting operations.

recipe_manager property

recipe_manager: RecipeManager

Get the recipe manager.

station_registry property

station_registry: StationRegistry

Get the station registry.

set_item_manager

set_item_manager(item_manager: ItemManager) -> None

Set the item manager for creating crafted items.

This allows deferred wiring when the ItemManager is created after the CraftingSystem (e.g., during content pack loading).

startup async

startup() -> None

Load recipes and initialize system.

shutdown async

shutdown() -> None

Shutdown crafting system.

update async

update(delta: float) -> None

Process ongoing crafts with cast times.

attempt_craft async

attempt_craft(
    character: Character,
    recipe: Recipe,
    materials: list[MaterialItem],
    station: CraftingStation | None = None,
    item_manager: Any = None,
) -> CraftingResult

Attempt to craft an item from a recipe.

Parameters:

Name Type Description Default
character Character

Character doing the crafting

required
recipe Recipe

Recipe to craft

required
materials list[MaterialItem]

Materials to use

required
station CraftingStation | None

Crafting station (if required)

None
item_manager Any

Optional explicit item manager for creating output. If not provided, uses the system's configured ItemManager or falls back to world.items.

None

Returns:

Type Description
CraftingResult

CraftingResult with success/failure and created item

learn_recipe

learn_recipe(character: Character, recipe: Recipe) -> bool

Teach a recipe to a character.

Parameters:

Name Type Description Default
character Character

Character to teach

required
recipe Recipe

Recipe to learn

required

Returns:

Type Description
bool

True if successfully learned

character_knows_recipe

character_knows_recipe(
    character: Character, recipe: Recipe
) -> bool

Check if character knows a recipe.

Parameters:

Name Type Description Default
character Character

Character to check

required
recipe Recipe

Recipe to check

required

Returns:

Type Description
bool

True if character knows the recipe

find_materials_for_recipe

find_materials_for_recipe(
    recipe: Recipe, inventory: list[MaterialItem]
) -> tuple[list[MaterialItem], list[str]]

Find matching materials for a recipe from an inventory.

Parameters:

Name Type Description Default
recipe Recipe

Recipe to match

required
inventory list[MaterialItem]

Available materials

required

Returns:

Type Description
tuple[list[MaterialItem], list[str]]

Tuple of (matched materials, missing material types)

EnchantingSystem

EnchantingSystem()

Handles item enchantment.

enchant_item async

enchant_item(
    character: Character,
    item: Item,
    enchantment_type: EnchantmentType,
    reagent_consumed: bool = True,
    station: CraftingStation | None = None,
) -> EnchantResult

Apply an enchantment to an item.

Parameters:

Name Type Description Default
character Character

Character enchanting

required
item Item

Item to enchant

required
enchantment_type EnchantmentType

Type of enchantment

required
reagent_consumed bool

Whether reagent cost was paid

True
station CraftingStation | None

Enchanting station if available

None

Returns:

Type Description
EnchantResult

EnchantResult with success/failure

Enchantment

Bases: MAIDBaseModel

An enchantment applied to an item.

EnchantmentType

Bases: str, Enum

Types of enchantments.

EnchantResult dataclass

EnchantResult(
    success: bool,
    error: str = "",
    enchantment: Enchantment | None = None,
    item_damaged: bool = False,
)

Result of an enchanting attempt.

GemType

Bases: str, Enum

Types of socketable gems.

RepairResult dataclass

RepairResult(
    success: bool,
    error: str = "",
    durability_restored: int = 0,
    new_durability: int = 0,
)

Result of a repair attempt.

RepairSystem

RepairSystem()

Handles item repair and durability.

repair_item async

repair_item(
    character: Character,
    item: Item,
    use_materials: bool = False,
    station: CraftingStation | None = None,
) -> RepairResult

Repair a damaged item.

Parameters:

Name Type Description Default
character Character

Character doing the repair

required
item Item

Item to repair

required
use_materials bool

Whether materials are being used for bonus

False
station CraftingStation | None

Crafting station if available

None

Returns:

Type Description
RepairResult

RepairResult with success/failure

damage_item

damage_item(item: Item, amount: int = 1) -> bool

Apply durability damage to an item.

Parameters:

Name Type Description Default
item Item

Item to damage

required
amount int

Durability points to remove

1

Returns:

Type Description
bool

True if item is now broken (0 durability)

SocketedGem

Bases: MAIDBaseModel

A gem socketed into an item.

SocketingSystem

Handles gem socketing into items.

socket_gem async

socket_gem(
    character: Character, item: Item, gem: Item
) -> SocketResult

Socket a gem into an item.

Parameters:

Name Type Description Default
character Character

Character doing the socketing

required
item Item

Item to socket into

required
gem Item

Gem item to socket

required

Returns:

Type Description
SocketResult

SocketResult with success/failure

remove_gem async

remove_gem(
    character: Character, item: Item, socket_index: int
) -> SocketResult

Remove a gem from an item (destroys the gem).

Parameters:

Name Type Description Default
character Character

Character doing the removal

required
item Item

Item to remove gem from

required
socket_index int

Index of socket to empty

required

Returns:

Type Description
SocketResult

SocketResult indicating success/failure

SocketResult dataclass

SocketResult(
    success: bool,
    error: str = "",
    gem: SocketedGem | None = None,
)

Result of a gem socketing attempt.

GatheringResult dataclass

GatheringResult(
    success: bool,
    error: str = "",
    materials: list[MaterialItem] = list(),
    skill_gained: float = 0.0,
)

Result of a gathering attempt.

GatheringSystem

GatheringSystem(world: World)

Bases: System, StorageAware

Manages resource gathering in the world.

set_storage

set_storage(store: DocumentStore) -> None

Set the document store for persistence.

startup async

startup() -> None

Load gathering nodes from storage on startup.

shutdown async

shutdown() -> None

Save gathering nodes to storage on shutdown.

update async

update(delta: float) -> None

Regenerate resources over time.

register_node

register_node(node: ResourceNode) -> None

Register a resource node.

Parameters:

Name Type Description Default
node ResourceNode

Node to register

required

unregister_node

unregister_node(node_id: UUID) -> ResourceNode | None

Unregister a resource node.

Parameters:

Name Type Description Default
node_id UUID

ID of node to remove

required

Returns:

Type Description
ResourceNode | None

Removed node, or None if not found

get_node

get_node(node_id: UUID) -> ResourceNode | None

Get a node by ID.

Parameters:

Name Type Description Default
node_id UUID

Node ID

required

Returns:

Type Description
ResourceNode | None

Node if found

get_nodes_in_room

get_nodes_in_room(
    room_id: UUID, visible_only: bool = True
) -> list[ResourceNode]

Get all resource nodes in a room.

Parameters:

Name Type Description Default
room_id UUID

Room ID

required
visible_only bool

If True, exclude hidden nodes

True

Returns:

Type Description
list[ResourceNode]

List of nodes in the room

attempt_gather async

attempt_gather(
    character: Character, node: ResourceNode
) -> GatheringResult

Attempt to gather from a resource node.

Parameters:

Name Type Description Default
character Character

Character gathering

required
node ResourceNode

Node to gather from

required

Returns:

Type Description
GatheringResult

GatheringResult with success/failure and materials

attempt_skinning async

attempt_skinning(
    character: Character, corpse: Item
) -> GatheringResult

Attempt to skin a corpse for leather/materials.

Parameters:

Name Type Description Default
character Character

Character skinning

required
corpse Item

Corpse item to skin

required

Returns:

Type Description
GatheringResult

GatheringResult with success/failure and materials

all_nodes

all_nodes() -> list[ResourceNode]

Get all registered nodes.

Returns:

Type Description
list[ResourceNode]

List of all nodes

clear

clear() -> None

Clear all registered nodes.

MaterialQuality

Bases: IntEnum

Material quality tiers with numeric values for calculations.

RecipeManager

RecipeManager(
    data_dir: Path | None = None, store: Any = None
)

Manages recipe loading, storage, and queries.

Parameters:

Name Type Description Default
data_dir Path | None

Directory containing recipe YAML files

None
store Any

DocumentStore instance for persistence

None

load async

load() -> int

Load all recipes from storage and YAML files.

Returns:

Type Description
int

Number of recipes loaded

save_recipe async

save_recipe(recipe: Recipe) -> None

Save a recipe to persistent storage.

Parameters:

Name Type Description Default
recipe Recipe

Recipe to save

required

register

register(recipe: Recipe) -> None

Register a recipe.

Parameters:

Name Type Description Default
recipe Recipe

Recipe to register

required

unregister

unregister(recipe_id: UUID) -> Recipe | None

Unregister a recipe.

Parameters:

Name Type Description Default
recipe_id UUID

ID of recipe to remove

required

Returns:

Type Description
Recipe | None

Removed recipe, or None if not found

get

get(recipe_id: UUID) -> Recipe | None

Get a recipe by ID.

Parameters:

Name Type Description Default
recipe_id UUID

Recipe ID

required

Returns:

Type Description
Recipe | None

Recipe if found

find_by_name

find_by_name(name: str) -> Recipe | None

Find a recipe by name (case-insensitive).

Parameters:

Name Type Description Default
name str

Recipe name to search for

required

Returns:

Type Description
Recipe | None

Matching recipe, or None

get_recipes_for_skill

get_recipes_for_skill(
    skill: str, max_level: int = 100
) -> list[Recipe]

Get all recipes for a skill up to a level.

Parameters:

Name Type Description Default
skill str

Skill name

required
max_level int

Maximum skill level filter

100

Returns:

Type Description
list[Recipe]

List of matching recipes

get_learnable_recipes

get_learnable_recipes(
    character: Character, skill: str
) -> list[Recipe]

Get recipes a character can learn but doesn't know.

Parameters:

Name Type Description Default
character Character

Character to check

required
skill str

Skill to check recipes for

required

Returns:

Type Description
list[Recipe]

List of learnable recipes

get_known_recipes

get_known_recipes(character: Character) -> list[Recipe]

Get all recipes known by a character.

Parameters:

Name Type Description Default
character Character

Character to check

required

Returns:

Type Description
list[Recipe]

List of known recipes

get_auto_learned_recipes

get_auto_learned_recipes(
    skill: str, skill_level: int
) -> list[Recipe]

Get recipes that are automatically learned.

Parameters:

Name Type Description Default
skill str

Skill name

required
skill_level int

Character's skill level

required

Returns:

Type Description
list[Recipe]

List of auto-learned recipes

get_recipe_by_output

get_recipe_by_output(
    output_template_id: UUID,
) -> Recipe | None

Get recipe that produces a specific item template.

Parameters:

Name Type Description Default
output_template_id UUID

Item template ID

required

Returns:

Type Description
Recipe | None

Recipe if found

all_recipes

all_recipes() -> list[Recipe]

Get all registered recipes.

Returns:

Type Description
list[Recipe]

List of all recipes

clear

clear() -> None

Clear all registered recipes.

StationRegistry

StationRegistry()

Manages crafting stations in the world.

register

register(station: CraftingStation) -> None

Register a crafting station.

Parameters:

Name Type Description Default
station CraftingStation

Station to register

required

unregister

unregister(station_id: UUID) -> CraftingStation | None

Unregister a crafting station.

Parameters:

Name Type Description Default
station_id UUID

ID of station to unregister

required

Returns:

Type Description
CraftingStation | None

The removed station, or None if not found

get

get(station_id: UUID) -> CraftingStation | None

Get a station by ID.

Parameters:

Name Type Description Default
station_id UUID

Station ID

required

Returns:

Type Description
CraftingStation | None

Station if found, None otherwise

get_stations_in_room

get_stations_in_room(
    room_id: UUID, station_type: StationType | None = None
) -> list[CraftingStation]

Get crafting stations in a room.

Parameters:

Name Type Description Default
room_id UUID

Room ID to search

required
station_type StationType | None

Optional filter by station type

None

Returns:

Type Description
list[CraftingStation]

List of matching stations

get_stations_by_type

get_stations_by_type(
    station_type: StationType,
) -> list[CraftingStation]

Get all stations of a specific type.

Parameters:

Name Type Description Default
station_type StationType

Type to search for

required

Returns:

Type Description
list[CraftingStation]

List of matching stations

find_nearest_station

find_nearest_station(
    room_id: UUID,
    station_type: StationType,
    world: World,
    max_distance: int = 10,
) -> tuple[CraftingStation | None, int]

Find nearest station of type using BFS.

Parameters:

Name Type Description Default
room_id UUID

Starting room ID

required
station_type StationType

Type of station to find

required
world World

World instance for room graph traversal

required
max_distance int

Maximum rooms to search

10

Returns:

Type Description
tuple[CraftingStation | None, int]

Tuple of (station or None, distance). Distance is -1 if not found.

all_stations

all_stations() -> list[CraftingStation]

Get all registered stations.

Returns:

Type Description
list[CraftingStation]

List of all stations

clear

clear() -> None

Clear all registered stations.

apply_quality_to_item

apply_quality_to_item(
    item: Item, quality: MaterialQuality
) -> None

Apply quality modifiers to a crafted item's stats.

Modifies: - Weapon damage (dice bonus) - Armor value - Item value (gold) - Stores quality in metadata

Parameters:

Name Type Description Default
item Item

The crafted item to modify

required
quality MaterialQuality

Quality tier to apply

required

calculate_output_quality

calculate_output_quality(
    crafter_skill: int,
    recipe_difficulty: int,
    materials: list[MaterialItem],
    quality_variance: float = 0.5,
    skill_bonus_divisor: float = 20.0,
) -> MaterialQuality

Calculate the quality of a crafted item.

Formula: - Base quality from average material quality - Skill bonus: (skill - difficulty) / 20 quality levels - Random variance: ± 0.5 quality levels (configurable) - Clamped to valid quality range (1-5)

Parameters:

Name Type Description Default
crafter_skill int

Crafter's skill level (1-100)

required
recipe_difficulty int

Recipe difficulty (1-100)

required
materials list[MaterialItem]

List of materials used

required
quality_variance float

Random variance range (default 0.5)

0.5
skill_bonus_divisor float

Points above difficulty per quality level

20.0

Returns:

Type Description
MaterialQuality

Final quality tier for the crafted item

Economy System

Shops, trading, banking, and auction house.

from maid_classic_rpg.systems.economy import EconomySystem

# Economy system handles:
# - NPC shops with inventory
# - Player-to-player trading
# - Banking and storage
# - Auction house

economy

Economy systems for shops, trading, banking, and auctions.

AuctionHouse

AuctionHouse(world: World)

Bases: System, StorageAware

Manages the auction house system.

Handles listing, searching, bidding, and purchasing.

active_count property

active_count: int

Return number of active listings.

set_storage

set_storage(store: DocumentStore) -> None

Receive storage from the engine via StorageAware protocol.

startup async

startup() -> None

Initialize auction house and load data from storage.

shutdown async

shutdown() -> None

Persist all auction data before shutdown.

update async

update(delta: float) -> None

Process expired auctions and deliver pending gold periodically.

list_item async

list_item(
    seller: Character,
    item: Item,
    starting_price: int,
    buyout_price: int | None = None,
    duration_hours: int = 24,
) -> AuctionResult

List an item for auction.

Parameters:

Name Type Description Default
seller Character

Character listing the item

required
item Item

Item to list

required
starting_price int

Starting bid price

required
buyout_price int | None

Optional instant purchase price

None
duration_hours int

Auction duration

24

Returns:

Type Description
AuctionResult

AuctionResult with success/failure and listing

place_bid async

place_bid(
    bidder: Character, listing_id: UUID, bid_amount: int
) -> BidResult

Place a bid on an auction.

Parameters:

Name Type Description Default
bidder Character

Character placing bid

required
listing_id UUID

Listing to bid on

required
bid_amount int

Amount to bid

required

Returns:

Type Description
BidResult

BidResult with success/failure

buyout async

buyout(
    buyer: Character,
    listing_id: UUID,
    item_manager: Any = None,
) -> BuyoutResult

Instantly purchase an item at buyout price.

Parameters:

Name Type Description Default
buyer Character

Character buying

required
listing_id UUID

Listing to purchase

required
item_manager Any

Item manager for getting item

None

Returns:

Type Description
BuyoutResult

BuyoutResult with success/failure

cancel_listing async

cancel_listing(
    seller: Character,
    listing_id: UUID,
    item_manager: Any = None,
) -> CancelResult

Cancel an auction listing.

Parameters:

Name Type Description Default
seller Character

Character cancelling

required
listing_id UUID

Listing to cancel

required
item_manager Any

Item manager for getting item

None

Returns:

Type Description
CancelResult

CancelResult with success/failure

search async

search(
    query: str | None = None,
    item_type: str | None = None,
    min_price: int | None = None,
    max_price: int | None = None,
    seller_name: str | None = None,
    sort_by: str = "expires_at",
    limit: int = 50,
    offset: int = 0,
) -> list[AuctionListing]

Search active auction listings.

Parameters:

Name Type Description Default
query str | None

Search term for item name

None
item_type str | None

Filter by item type

None
min_price int | None

Minimum current bid

None
max_price int | None

Maximum current bid

None
seller_name str | None

Filter by seller name

None
sort_by str

Sort field (expires_at, price_asc, price_desc, bid_count)

'expires_at'
limit int

Maximum results

50
offset int

Result offset for pagination

0

Returns:

Type Description
list[AuctionListing]

List of matching listings

get_listing

get_listing(listing_id: UUID) -> AuctionListing | None

Get a listing by ID.

get_listing_by_short_id

get_listing_by_short_id(
    short_id: str,
) -> AuctionListing | None

Get a listing by short ID prefix.

Parameters:

Name Type Description Default
short_id str

First 7 characters of listing ID

required

Returns:

Type Description
AuctionListing | None

Matching listing, or None

get_seller_listings

get_seller_listings(
    seller_id: UUID, active_only: bool = False
) -> list[AuctionListing]

Get all listings by a seller.

Parameters:

Name Type Description Default
seller_id UUID

Seller character ID

required
active_only bool

If True, only return active listings

False

Returns:

Type Description
list[AuctionListing]

List of listings

get_pending_gold

get_pending_gold(character_id: UUID) -> int

Get pending gold for a character (from sales/refunds).

Parameters:

Name Type Description Default
character_id UUID

Character ID

required

Returns:

Type Description
int

Pending gold amount

collect_pending_gold

collect_pending_gold(character: Character) -> int

Collect pending gold for a character.

Parameters:

Name Type Description Default
character Character

Character collecting

required

Returns:

Type Description
int

Amount collected

get_pending_items

get_pending_items(character_id: UUID) -> list[UUID]

Get pending items for a character (from wins).

Parameters:

Name Type Description Default
character_id UUID

Character ID

required

Returns:

Type Description
list[UUID]

List of pending item IDs

collect_pending_items

collect_pending_items(character_id: UUID) -> list[UUID]

Collect pending items for a character.

Parameters:

Name Type Description Default
character_id UUID

Character ID

required

Returns:

Type Description
list[UUID]

List of item IDs collected

auto_deliver_pending_gold async

auto_deliver_pending_gold() -> int

Deliver pending gold to all online players.

Called periodically from update() and on player connection.

Returns:

Type Description
int

Total gold delivered across all players.

clear

clear() -> None

Clear all auction data.

BankSystem

BankSystem(world: World)

Bases: System, StorageAware

System for bank operations.

Banking uses Character.bank_gold for storage. This system handles deposit/withdraw transactions. Bank NPC registrations are persisted via the DocumentStore.

get_transactions

get_transactions(
    character_id: UUID | None = None,
) -> list[BankTransaction]

Get transaction history, optionally filtered by character.

Parameters:

Name Type Description Default
character_id UUID | None

If provided, only return transactions for this character.

None

Returns:

Type Description
list[BankTransaction]

List of BankTransaction records.

set_storage

set_storage(store: DocumentStore) -> None

Set the document store for persistence.

startup async

startup() -> None

Load banks from storage on startup.

shutdown async

shutdown() -> None

Save banks to storage on shutdown.

register_bank

register_bank(bank: BankNPC) -> None

Register a bank NPC.

Parameters:

Name Type Description Default
bank BankNPC

The bank NPC to register.

required

Raises:

Type Description
ValueError

If the bank's room_id does not correspond to an existing entity in the world.

unregister_bank

unregister_bank(bank_character_id: UUID) -> BankNPC | None

Unregister a bank NPC.

get_bank_in_room

get_bank_in_room(room_id: UUID) -> BankNPC | None

Get bank NPC in a room.

get_bank

get_bank(bank_character_id: UUID) -> BankNPC | None

Get bank NPC by character ID.

all_banks

all_banks() -> list[BankNPC]

Get all registered banks.

update async

update(delta: float) -> None

Bank system tick (interest calculation if enabled).

check_balance async

check_balance(
    character_id: UUID, character_manager: CharacterManager
) -> tuple[bool, int]

Check bank balance.

Returns:

Type Description
tuple[bool, int]

Tuple of (success, balance)

deposit async

deposit(
    character_id: UUID,
    amount: int,
    bank: BankNPC | None,
    character_manager: CharacterManager,
) -> tuple[bool, str, int]

Deposit gold into bank.

Parameters:

Name Type Description Default
character_id UUID

Character depositing

required
amount int

Amount to deposit

required
bank BankNPC | None

Bank NPC (for fees)

required
character_manager CharacterManager

Character manager

required

Returns:

Type Description
tuple[bool, str, int]

Tuple of (success, message, new_balance)

withdraw async

withdraw(
    character_id: UUID,
    amount: int,
    bank: BankNPC | None,
    character_manager: CharacterManager,
) -> tuple[bool, str, int]

Withdraw gold from bank.

Parameters:

Name Type Description Default
character_id UUID

Character withdrawing

required
amount int

Amount to withdraw

required
bank BankNPC | None

Bank NPC (for fees)

required
character_manager CharacterManager

Character manager

required

Returns:

Type Description
tuple[bool, str, int]

Tuple of (success, message, amount_received)

transfer async

transfer(
    from_character_id: UUID,
    to_character_id: UUID,
    amount: int,
    character_manager: CharacterManager,
) -> tuple[bool, str]

Transfer gold between bank accounts.

Parameters:

Name Type Description Default
from_character_id UUID

Character sending

required
to_character_id UUID

Character receiving

required
amount int

Amount to transfer

required
character_manager CharacterManager

Character manager

required

Returns:

Type Description
tuple[bool, str]

Tuple of (success, message)

ShopManager

ShopManager(world: World | None = None)

Manages shop data and operations.

Handles shop registration, inventory, and pricing calculations.

register_shop

register_shop(shop: Shop) -> None

Register a shop.

Validates room_id exists in world (if world is set) and warns on overwriting an existing shop in the same room.

register_merchant

register_merchant(merchant: MerchantNPC) -> None

Register a merchant NPC linked to a shop.

get_merchant_for_shop

get_merchant_for_shop(shop_id: UUID) -> MerchantNPC | None

Get the MerchantNPC associated with a shop.

unregister_shop

unregister_shop(shop_id: UUID) -> Shop | None

Unregister a shop. Returns the shop if found.

get_shop

get_shop(shop_id: UUID) -> Shop | None

Get shop by ID.

get_shop_in_room

get_shop_in_room(room_id: UUID) -> Shop | None

Get shop operating in a room.

get_shop_by_merchant

get_shop_by_merchant(merchant_id: UUID) -> Shop | None

Get shop by merchant entity ID.

all_shops

all_shops() -> list[Shop]

Get all registered shops.

ShopSystem

ShopSystem(world: World)

Bases: System, StorageAware

System for processing shop transactions and restocking.

Handles: - Buy/sell transactions - Price calculations with modifiers - Inventory restocking over time - Shop gold management

manager property

manager: ShopManager

Get the shop manager.

set_storage

set_storage(store: DocumentStore) -> None

Set the document store for persistence.

startup async

startup() -> None

Load shops from storage on startup.

shutdown async

shutdown() -> None

Save dirty shops to storage on shutdown.

update async

update(delta: float) -> None

Process shop restocking and persist dirty shops.

list_inventory async

list_inventory(
    shop: Shop,
    item_manager: ItemManager,
    charisma_mod: int = 0,
    reputation: int = 0,
) -> list[tuple[str, int, int]]

Get shop inventory for display.

Parameters:

Name Type Description Default
shop Shop

Shop to list

required
item_manager ItemManager

Item manager

required
charisma_mod int

Player's charisma modifier for price calculation

0
reputation int

Player's reputation for price calculation

0

Returns:

Type Description
list[tuple[str, int, int]]

List of (item_name, price, quantity)

buy_item async

buy_item(
    shop: Shop,
    buyer_id: UUID,
    item_template_id: UUID,
    character_manager: CharacterManager,
    item_manager: ItemManager,
    charisma_mod: int = 0,
    reputation: int = 0,
) -> tuple[bool, str, Item | None]

Process a purchase.

Uses MerchantNPC dialogue strings when available for custom messages.

Parameters:

Name Type Description Default
shop Shop

Shop to buy from

required
buyer_id UUID

Character buying

required
item_template_id UUID

Template of item to buy

required
character_manager CharacterManager

Character manager

required
item_manager ItemManager

Item manager

required
charisma_mod int

Buyer's charisma modifier

0
reputation int

Buyer's reputation

0

Returns:

Type Description
tuple[bool, str, Item | None]

Tuple of (success, message, item_or_none)

sell_item async

sell_item(
    shop: Shop,
    seller_id: UUID,
    item_id: UUID,
    character_manager: CharacterManager,
    item_manager: ItemManager,
    charisma_mod: int = 0,
    reputation: int = 0,
) -> tuple[bool, str, int]

Process a sale.

Uses MerchantNPC dialogue strings when available for custom messages.

Parameters:

Name Type Description Default
shop Shop

Shop to sell to

required
seller_id UUID

Character selling

required
item_id UUID

Item being sold

required
character_manager CharacterManager

Character manager

required
item_manager ItemManager

Item manager

required
charisma_mod int

Seller's charisma modifier

0
reputation int

Seller's reputation

0

Returns:

Type Description
tuple[bool, str, int]

Tuple of (success, message, gold_received)

get_value async

get_value(
    shop: Shop,
    item: Item,
    charisma_mod: int = 0,
    reputation: int = 0,
) -> tuple[bool, int]

Get what a shop would pay for an item.

Returns:

Type Description
tuple[bool, int]

Tuple of (would_buy, price)

get_buy_price async

get_buy_price(
    shop: Shop,
    entry: ShopInventoryEntry,
    charisma_mod: int = 0,
    reputation: int = 0,
) -> int

Get the buy price for a shop item entry.

TradeSystem

TradeSystem(world: World)

Bases: System

System for managing player-to-player trades.

Trade flow: 1. Player A initiates trade with Player B 2. Player B accepts (PENDING -> ACTIVE) 3. Both players add/remove items and gold 4. Both players lock their offers (ACTIVE -> LOCKED) 5. Both players confirm (LOCKED -> CONFIRMED -> COMPLETED)

Either player can cancel at any time before completion.

startup async

startup() -> None

Initialize trade system.

shutdown async

shutdown() -> None

Shutdown trade system.

update async

update(_delta: float) -> None

Check for expired trades.

get_player_session

get_player_session(player_id: UUID) -> TradeSession | None

Get the active trade session for a player.

initiate_trade async

initiate_trade(
    initiator_id: UUID, target_id: UUID
) -> tuple[bool, str, TradeSession | None]

Start a trade with another player.

Returns:

Type Description
tuple[bool, str, TradeSession | None]

Tuple of (success, message, session_or_none)

accept_trade async

accept_trade(player_id: UUID) -> tuple[bool, str]

Accept a pending trade invitation.

decline_trade async

decline_trade(player_id: UUID) -> tuple[bool, str]

Decline a pending trade.

add_item async

add_item(
    player_id: UUID,
    item_id: UUID,
    character_manager: CharacterManager,
) -> tuple[bool, str]

Add an item to trade offer.

remove_item async

remove_item(
    player_id: UUID, item_id: UUID
) -> tuple[bool, str]

Remove an item from trade offer.

set_gold async

set_gold(
    player_id: UUID,
    amount: int,
    character_manager: CharacterManager,
) -> tuple[bool, str]

Set gold amount in trade offer.

lock_offer async

lock_offer(player_id: UUID) -> tuple[bool, str]

Lock your trade offer.

unlock_offer async

unlock_offer(player_id: UUID) -> tuple[bool, str]

Unlock your trade offer.

confirm_trade async

confirm_trade(
    player_id: UUID,
    character_manager: CharacterManager,
    item_manager: ItemManager,
) -> tuple[bool, str]

Confirm and execute trade.

cancel_trade async

cancel_trade(
    session_id: UUID,
    cancelled_by: UUID | None = None,
    reason: str = "cancelled",
) -> None

Cancel a trade session.

get_trade_status

get_trade_status(
    player_id: UUID,
) -> dict[str, object] | None

Get current trade status for a player.

Returns:

Type Description
dict[str, object] | None

Dict with trade details or None if not in trade.

Quest System

Quest tracking with objectives, rewards, and progression.

from maid_classic_rpg.systems.quests import QuestSystem

# Quest system handles:
# - Quest discovery and acceptance
# - Objective tracking
# - Reward distribution
# - Quest chains and prerequisites

quests

Quest system for MAID.

Provides quest definitions, tracking, and rewards.

QuestManager

QuestManager(
    quest_data_dir: Path | None = None,
    quest_log_dir: Path | None = None,
    event_bus: EventBus | None = None,
)

Manages quest definitions and character quest logs.

Responsibilities: - Load quest definitions from data files - Store and retrieve quest logs per character - Check quest prerequisites - Manage quest state transitions

Example

manager = QuestManager() await manager.load()

Get available quests for a character

available = await manager.get_available_quests(character)

Accept a quest

await manager.accept_quest(character, "starter_quest_1")

Parameters:

Name Type Description Default
quest_data_dir Path | None

Directory containing quest definition files

None
quest_log_dir Path | None

Directory for storing character quest logs (fallback)

None

set_storage

set_storage(store: DocumentStore) -> None

Set the document store for persistence.

load async

load() -> None

Load quest definitions and quest logs from storage.

get_quest

get_quest(quest_id: str) -> Quest | None

Get a quest definition by ID.

get_quest_log

get_quest_log(character_id: UUID) -> QuestLog

Get or create a quest log for a character.

shutdown async

shutdown() -> None

Await all pending saves to prevent data loss on shutdown.

get_all_quest_logs

get_all_quest_logs() -> list[tuple[UUID, QuestLog]]

Get all character quest logs.

Returns:

Type Description
list[tuple[UUID, QuestLog]]

List of (character_id, quest_log) tuples.

save_quest_log async

save_quest_log(log: QuestLog) -> None

Save a character's quest log to storage.

Public wrapper for _save_quest_log.

check_prerequisites

check_prerequisites(
    quest: Quest, character: Character, quest_log: QuestLog
) -> tuple[bool, str]

Check if a character meets quest prerequisites.

Parameters:

Name Type Description Default
quest Quest

Quest to check

required
character Character

Character to check

required
quest_log QuestLog

Character's quest log

required

Returns:

Type Description
tuple[bool, str]

Tuple of (meets_prereqs, reason_if_not)

get_available_quests async

get_available_quests(
    character: Character, npc_id: UUID | None = None
) -> list[Quest]

Get quests available to a character.

Parameters:

Name Type Description Default
character Character

Character to check

required
npc_id UUID | None

Optional NPC to filter by (only quests this NPC gives)

None

Returns:

Type Description
list[Quest]

List of available quests

accept_quest async

accept_quest(
    character: Character, quest_id: str
) -> tuple[bool, str]

Accept a quest for a character.

Parameters:

Name Type Description Default
character Character

Character accepting the quest

required
quest_id str

Quest to accept

required

Returns:

Type Description
tuple[bool, str]

Tuple of (success, message)

abandon_quest async

abandon_quest(
    character: Character, quest_id: str
) -> tuple[bool, str]

Abandon an active quest.

Parameters:

Name Type Description Default
character Character

Character abandoning the quest

required
quest_id str

Quest to abandon

required

Returns:

Type Description
tuple[bool, str]

Tuple of (success, message)

expire_quest async

expire_quest(character_id: UUID, quest_id: str) -> bool

Expire an active quest for a character.

Parameters:

Name Type Description Default
character_id UUID

Character with the active quest.

required
quest_id str

Quest identifier to expire.

required

Returns:

Type Description
bool

True if quest was expired.

register_generated_quest

register_generated_quest(quest: Quest) -> None

Register a generated quest at runtime.

Parameters:

Name Type Description Default
quest Quest

Generated quest model.

required

Raises:

Type Description
ValueError

If quest_id already exists.

invalidate_quest

invalidate_quest(quest_id: str) -> None

Invalidate a generated quest and remove pending offers.

get_generated_quests

get_generated_quests() -> list[Quest]

Get all currently registered generated quests.

get_npc_quests async

get_npc_quests(
    character: Character, npc_id: UUID
) -> dict[str, list[Quest]]

Get quests related to an NPC for a character.

Returns dict with keys: - available: Quests NPC can offer - in_progress: Quests in progress related to NPC - ready_to_turn_in: Completed quests to turn in to NPC

get_quests_by_category async

get_quests_by_category(
    character: Character, category: str
) -> list[tuple[Quest, QuestProgress]]

Get active quests by category.

get_all_quests

get_all_quests() -> list[Quest]

Get all registered quest definitions.

DeliverObjectiveHandler

DeliverObjectiveHandler(quest_manager: QuestManager)

Handler for DELIVER objective type.

Deliver objectives require giving a specific item to a specific NPC. This is checked when interacting with the target NPC.

check_delivery async

check_delivery(
    character_id: UUID,
    npc_id: UUID,
    item_id: UUID,
    item_template_id: UUID | None,
) -> list[str]

Check if giving this item completes any deliver objectives.

Returns:

Type Description
list[str]

List of quest IDs where delivery was successful

EscortObjectiveHandler

EscortObjectiveHandler(quest_manager: QuestManager)

Handler for ESCORT objective type.

Escort objectives require an NPC to reach a destination room while keeping them alive.

check_escort_arrival async

check_escort_arrival(
    npc_id: UUID, room_id: UUID
) -> list[tuple[UUID, str]]

Check if NPC arriving at room completes any escort objectives.

Returns:

Type Description
list[tuple[UUID, str]]

List of (character_id, quest_id) for completed escorts

ObjectiveTracker

ObjectiveTracker(world: World, quest_manager: QuestManager)

Bases: System, StorageAware

System for tracking quest objective progress.

Subscribes to game events and updates quest progress accordingly. Handles all objective types: kill, collect, deliver, visit, talk, etc.

Inherits from StorageAware to receive storage injection and pass it to QuestManager.

quest_manager property

quest_manager: QuestManager

Get the quest manager for quest operations.

get_reward_distributor

get_reward_distributor(
    item_manager: Any = None,
) -> RewardDistributor

Get or create the reward distributor for quest turn-ins.

Parameters:

Name Type Description Default
item_manager Any

Optional item manager for item rewards.

None

Returns:

Type Description
RewardDistributor

RewardDistributor instance.

set_storage

set_storage(store: DocumentStore) -> None

Receive storage from the engine and pass it to QuestManager.

startup async

startup() -> None

Subscribe to relevant events.

shutdown async

shutdown() -> None

Unsubscribe from events.

update async

update(_delta: float) -> None

Check timed objectives.

RewardDistributor

RewardDistributor(
    world: World,
    quest_manager: QuestManager,
    item_manager: ItemManager | None = None,
)

Handles quest reward distribution.

Responsibilities: - Validate quest can be turned in - Grant XP, gold, items, reputation - Handle reward choices - Emit reward events - Update quest state to TURNED_IN

set_item_manager

set_item_manager(item_manager: ItemManager) -> None

Set or update the item manager reference.

This allows deferred wiring when the ItemManager is created after the RewardDistributor (e.g., during content pack loading).

turn_in_quest async

turn_in_quest(
    character: Character,
    quest_id: str,
    reward_choice_index: int | None = None,
) -> tuple[bool, str, dict[str, Any] | None]

Turn in a completed quest and grant rewards.

Parameters:

Name Type Description Default
character Character

Character turning in quest

required
quest_id str

Quest to turn in

required
reward_choice_index int | None

Index of chosen reward (for choice rewards)

None

Returns:

Type Description
tuple[bool, str, dict[str, Any] | None]

Tuple of (success, message, rewards_granted)

format_rewards

format_rewards(rewards: QuestReward) -> str

Format rewards for display to player.

Social System

Guilds, factions, achievements, and social features.

from maid_classic_rpg.systems.social import SocialSystem

# Social system handles:
# - Guild creation and management
# - Faction reputation
# - Achievement tracking
# - Friend lists and ignore

social

Social systems for player communication and groups.

AchievementSystem

AchievementSystem(world: World)

Bases: System, StorageAware

System for tracking achievements.

Listens to game events and updates achievement progress accordingly.

set_storage

set_storage(store: DocumentStore) -> None

Set the document store for persistence.

update async

update(delta: float) -> None

No per-tick processing needed — achievements are event-driven via update_criterion().

startup async

startup() -> None

Subscribe to relevant game events and load data.

shutdown async

shutdown() -> None

Save data, clean up and unsubscribe from events.

register_achievement

register_achievement(achievement: Achievement) -> None

Register an achievement.

unregister_achievement

unregister_achievement(achievement_id: UUID) -> bool

Unregister an achievement.

get_achievement

get_achievement(achievement_id: UUID) -> Achievement | None

Get an achievement by ID.

get_achievement_by_name

get_achievement_by_name(name: str) -> Achievement | None

Get an achievement by name.

get_achievements_by_category

get_achievements_by_category(
    category: AchievementCategory,
) -> list[Achievement]

Get all achievements in a category.

get_all_achievements

get_all_achievements() -> list[Achievement]

Get all registered achievements.

get_progress

get_progress(
    character_id: UUID, achievement_id: UUID
) -> CharacterAchievementProgress

Get a character's progress on an achievement.

get_all_progress

get_all_progress(
    character_id: UUID,
) -> dict[UUID, CharacterAchievementProgress]

Get all achievement progress for a character.

get_earned_achievements

get_earned_achievements(
    character_id: UUID,
) -> list[Achievement]

Get all achievements a character has earned.

get_total_points

get_total_points(character_id: UUID) -> int

Get total achievement points for a character.

has_achievement

has_achievement(
    character_id: UUID, achievement_id: UUID
) -> bool

Check if a character has earned an achievement.

update_criterion async

update_criterion(
    character_id: UUID,
    criterion_type: str,
    target_id: UUID | None = None,
    target_type: str = "",
    amount: int = 1,
) -> list[Achievement]

Update progress on all matching criteria.

Returns:

Type Description
list[Achievement]

List of newly completed achievements

grant_achievement async

grant_achievement(
    character_id: UUID, achievement_id: UUID
) -> bool

Manually grant an achievement to a character.

Returns:

Type Description
bool

True if achievement was newly granted

claim_rewards async

claim_rewards(
    character_id: UUID, achievement_id: UUID
) -> list[AchievementReward]

Claim rewards for a completed achievement.

Returns:

Type Description
list[AchievementReward]

List of rewards granted

clear_character_progress

clear_character_progress(character_id: UUID) -> None

Clear all achievement progress for a character (for cleanup).

FactionSystem

FactionSystem(world: World)

Bases: System, StorageAware

System for tracking faction reputation.

Manages faction definitions and per-character reputation values.

set_storage

set_storage(store: DocumentStore) -> None

Set the document store for persistence.

update async

update(delta: float) -> None

Slowly decay reputation toward neutral each tick.

Decay is ~1 point per 10 real-time minutes. We accumulate elapsed time and apply integer decrements when the accumulator exceeds the threshold.

startup async

startup() -> None

Load factions and reputations from storage on startup.

shutdown async

shutdown() -> None

Save factions and reputations to storage on shutdown.

register_faction

register_faction(faction: Faction) -> None

Register a faction.

unregister_faction

unregister_faction(faction_id: UUID) -> bool

Unregister a faction.

get_faction

get_faction(faction_id: UUID) -> Faction | None

Get a faction by ID.

get_faction_by_name

get_faction_by_name(name: str) -> Faction | None

Get a faction by name.

get_all_factions

get_all_factions() -> list[Faction]

Get all registered factions.

get_reputation

get_reputation(
    character_id: UUID, faction_id: UUID
) -> CharacterReputation

Get a character's reputation with a faction.

get_reputation_level

get_reputation_level(
    character_id: UUID, faction_id: UUID
) -> ReputationLevel

Get a character's reputation level with a faction.

get_all_reputations

get_all_reputations(
    character_id: UUID,
) -> dict[UUID, CharacterReputation]

Get all reputations for a character.

modify_reputation async

modify_reputation(
    character_id: UUID,
    faction_id: UUID,
    amount: int,
    propagate: bool = True,
) -> tuple[int, ReputationLevel | None]

Modify a character's reputation with a faction.

Parameters:

Name Type Description Default
character_id UUID

Character whose reputation to modify

required
faction_id UUID

Faction to modify reputation with

required
amount int

Amount to add (negative to subtract)

required
propagate bool

If True, also modify related factions

True

Returns:

Type Description
tuple[int, ReputationLevel | None]

Tuple of (new_value, new_level if changed)

set_reputation

set_reputation(
    character_id: UUID, faction_id: UUID, value: int
) -> ReputationLevel

Set a character's reputation to a specific value.

Returns:

Type Description
ReputationLevel

The new reputation level

get_unlocked_rewards

get_unlocked_rewards(
    character_id: UUID, faction_id: UUID
) -> list[FactionReward]

Get rewards a character has unlocked with a faction.

get_npc_reaction

get_npc_reaction(
    character_id: UUID, npc_faction_id: UUID
) -> str

Get how an NPC should react based on reputation.

Returns:

Type Description
str

Reaction type: "attack", "refuse", "neutral", "friendly", "eager"

get_price_modifier

get_price_modifier(
    character_id: UUID, faction_id: UUID
) -> float

Get shop price modifier based on reputation.

Returns:

Type Description
float

Multiplier for prices (1.0 = normal, 0.8 = 20% discount)

can_access_vendor

can_access_vendor(
    character_id: UUID,
    faction_id: UUID,
    required_level: ReputationLevel = NEUTRAL,
) -> bool

Check if a character can access a faction vendor.

clear_character_reputations

clear_character_reputations(character_id: UUID) -> None

Clear all reputations for a character (for cleanup).

GroupSystem

GroupSystem(
    world: World, sessions: SessionManager | None = None
)

Bases: System

Manages player groups/parties.

Features: - Group creation and management - Follow leader movement - XP sharing - Loot distribution modes - Group chat (gtell)

set_sessions

set_sessions(sessions: SessionManager) -> None

Set the session manager after initialization.

startup async

startup() -> None

Subscribe to movement events for follow.

shutdown async

shutdown() -> None

Unsubscribe from events.

update async

update(_delta: float) -> None

Remove stale members and dissolve empty groups.

create_group

create_group(
    leader_id: UUID, leader_name: str
) -> tuple[Group | None, str]

Create a new group with player as leader.

disband_group

disband_group(player_id: UUID) -> tuple[bool, str]

Disband a group. Only leader can disband.

invite_player async

invite_player(
    inviter_id: UUID,
    inviter_name: str,
    invitee_id: UUID,
    invitee_name: str,
) -> tuple[bool, str]

Invite a player to group.

accept_invite async

accept_invite(
    player_id: UUID, player_name: str
) -> tuple[bool, str]

Accept a group invitation.

decline_invite

decline_invite(player_id: UUID) -> tuple[bool, str]

Decline a group invitation.

leave_group async

leave_group(player_id: UUID) -> tuple[bool, str]

Leave current group.

kick_member async

kick_member(
    kicker_id: UUID, target_id: UUID
) -> tuple[bool, str]

Kick a member from the group.

promote_member

promote_member(
    promoter_id: UUID, target_id: UUID
) -> tuple[bool, str]

Promote a member to assistant.

demote_member

demote_member(
    demoter_id: UUID, target_id: UUID
) -> tuple[bool, str]

Demote an assistant to member.

transfer_leadership

transfer_leadership(
    current_leader_id: UUID, new_leader_id: UUID
) -> tuple[bool, str]

Transfer group leadership to another member.

set_follow

set_follow(
    follower_id: UUID, leader_id: UUID
) -> tuple[bool, str]

Set a player to follow another.

stop_following

stop_following(player_id: UUID) -> tuple[bool, str]

Stop following.

group_tell async

group_tell(
    sender_id: UUID, sender_name: str, message: str
) -> tuple[bool, str]

Send message to group.

set_loot_mode

set_loot_mode(
    player_id: UUID, mode: LootMode
) -> tuple[bool, str]

Set group loot mode. Leader only.

get_looter

get_looter(group_id: UUID) -> UUID | None

Get the next player who should loot (for round-robin).

set_xp_share

set_xp_share(
    player_id: UUID, enabled: bool
) -> tuple[bool, str]

Enable or disable XP sharing. Leader only.

calculate_xp_share

calculate_xp_share(
    group_id: UUID, base_xp: int, _source_level: int
) -> dict[UUID, int]

Calculate XP distribution for group members.

get_player_group

get_player_group(player_id: UUID) -> Group | None

Get the group a player is in.

get_group

get_group(group_id: UUID) -> Group | None

Get group by ID.

is_in_group

is_in_group(player_id: UUID) -> bool

Check if player is in a group.

get_pending_invite

get_pending_invite(
    player_id: UUID,
) -> tuple[UUID, UUID] | None

Get pending invite for a player (group_id, inviter_id).

GuildBankError

Bases: GuildError

Error specific to guild bank operations.

GuildError

Bases: Exception

Base exception for guild operations.

GuildFullError

Bases: GuildError

Guild is at maximum capacity.

GuildNotFoundError

Bases: GuildError

Guild does not exist.

GuildPermissionError

Bases: GuildError

Character lacks permission for operation.

GuildSystem

GuildSystem(world: World)

Bases: System, StorageAware

System for managing player guilds.

Handles guild creation, membership, ranks, and progression.

set_storage

set_storage(store: DocumentStore) -> None

Set the document store for persistence.

update async

update(_delta: float) -> None

Persist dirty guilds and expire stale invitations.

startup async

startup() -> None

Load guilds from storage on startup.

shutdown async

shutdown() -> None

Save guilds to storage on shutdown.

get_guild

get_guild(guild_id: UUID) -> Guild | None

Get a guild by ID.

get_guild_by_name

get_guild_by_name(name: str) -> Guild | None

Get a guild by name (case-insensitive).

get_character_guild

get_character_guild(character_id: UUID) -> Guild | None

Get the guild a character belongs to.

is_in_guild

is_in_guild(character_id: UUID) -> bool

Check if a character is in any guild.

get_all_guilds

get_all_guilds() -> list[Guild]

Get all guilds.

create_guild async

create_guild(
    name: str, tag: str, leader_id: UUID, leader_name: str
) -> Guild

Create a new guild.

Parameters:

Name Type Description Default
name str

Guild name (must be unique)

required
tag str

Short guild tag (2-5 chars)

required
leader_id UUID

Character ID of the founder

required
leader_name str

Character name of the founder

required

Returns:

Type Description
Guild

The created Guild

Raises:

Type Description
GuildError

If character already in guild or name taken

disband_guild async

disband_guild(guild_id: UUID, requester_id: UUID) -> None

Disband a guild.

Parameters:

Name Type Description Default
guild_id UUID

Guild to disband

required
requester_id UUID

Character requesting disbandment

required

Raises:

Type Description
GuildNotFoundError

If guild doesn't exist

GuildPermissionError

If requester lacks permission

invite_member async

invite_member(
    guild_id: UUID, inviter_id: UUID, invitee_id: UUID
) -> None

Invite a player to the guild.

Parameters:

Name Type Description Default
guild_id UUID

Guild doing the inviting

required
inviter_id UUID

Character sending the invite

required
invitee_id UUID

Character being invited

required

Raises:

Type Description
GuildNotFoundError

If guild doesn't exist

GuildPermissionError

If inviter lacks permission

GuildError

If invitee already in a guild or guild full

get_pending_invites

get_pending_invites(character_id: UUID) -> list[Guild]

Get all pending guild invites for a character.

accept_invite async

accept_invite(
    character_id: UUID, character_name: str, guild_id: UUID
) -> Guild

Accept a guild invitation.

Parameters:

Name Type Description Default
character_id UUID

Character accepting

required
character_name str

Character's name

required
guild_id UUID

Guild to join

required

Returns:

Type Description
Guild

The joined Guild

Raises:

Type Description
GuildError

If no pending invite or already in guild

decline_invite async

decline_invite(character_id: UUID, guild_id: UUID) -> None

Decline a guild invitation.

leave_guild async

leave_guild(character_id: UUID) -> None

Leave current guild.

Raises:

Type Description
GuildError

If not in a guild or is leader

kick_member async

kick_member(
    guild_id: UUID, kicker_id: UUID, target_id: UUID
) -> None

Kick a member from the guild.

Raises:

Type Description
GuildPermissionError

If kicker lacks permission or outranked

promote_member async

promote_member(
    guild_id: UUID, promoter_id: UUID, target_id: UUID
) -> str

Promote a member to the next rank.

Returns:

Type Description
str

New rank name

demote_member async

demote_member(
    guild_id: UUID, demoter_id: UUID, target_id: UUID
) -> str

Demote a member to the previous rank.

Returns:

Type Description
str

New rank name

transfer_leadership async

transfer_leadership(
    guild_id: UUID,
    current_leader_id: UUID,
    new_leader_id: UUID,
) -> None

Transfer guild leadership to another member.

add_guild_experience async

add_guild_experience(guild_id: UUID, amount: int) -> bool

Add experience to a guild.

Returns:

Type Description
bool

True if guild leveled up

get_bank_tabs

get_bank_tabs(
    guild_id: UUID, character_id: UUID
) -> list[GuildBankTab]

Get accessible bank tabs for a character.

get_tab_contents

get_tab_contents(
    guild_id: UUID, character_id: UUID, tab_index: int
) -> tuple[GuildBankTab, list[UUID]]

Get contents of a specific bank tab.

Returns:

Type Description
tuple[GuildBankTab, list[UUID]]

Tuple of (tab, item_ids)

deposit_item async

deposit_item(
    guild_id: UUID,
    character_id: UUID,
    character_name: str,
    item_id: UUID,
    item_name: str,
    tab_index: int = 0,
    character: Any | None = None,
) -> None

Deposit an item into the guild bank.

Parameters:

Name Type Description Default
guild_id UUID

Guild to deposit into.

required
character_id UUID

Character depositing.

required
character_name str

Character's name.

required
item_id UUID

Item to deposit.

required
item_name str

Item's name.

required
tab_index int

Bank tab index.

0
character Any | None

Optional character object for atomic inventory removal.

None

withdraw_item async

withdraw_item(
    guild_id: UUID,
    character_id: UUID,
    character_name: str,
    item_id: UUID,
    item_name: str,
    tab_index: int = 0,
    character: Any | None = None,
) -> None

Withdraw an item from the guild bank.

deposit_gold async

deposit_gold(
    guild_id: UUID,
    character_id: UUID,
    character_name: str,
    amount: int,
    character: Any | None = None,
) -> None

Deposit gold into the guild bank.

Parameters:

Name Type Description Default
guild_id UUID

Guild to deposit into.

required
character_id UUID

Character depositing.

required
character_name str

Character's name.

required
amount int

Amount to deposit.

required
character Any | None

Optional character object for atomic gold deduction.

None

withdraw_gold async

withdraw_gold(
    guild_id: UUID,
    character_id: UUID,
    character_name: str,
    amount: int,
    character: Any | None = None,
) -> None

Withdraw gold from the guild bank.

get_bank_logs

get_bank_logs(
    guild_id: UUID, character_id: UUID, limit: int = 50
) -> list[GuildBankLog]

Get recent bank transaction logs.

get_gold_balance

get_gold_balance(guild_id: UUID, character_id: UUID) -> int

Get the guild's gold balance.

send_guild_message async

send_guild_message(
    guild_id: UUID,
    sender_id: UUID,
    sender_name: str,
    message: str,
    is_announcement: bool = False,
) -> None

Send a message to the guild chat.

set_motd

set_motd(
    guild_id: UUID, character_id: UUID, motd: str
) -> None

Set the guild message of the day.

get_motd

get_motd(guild_id: UUID) -> str

Get the guild message of the day.

register_social_commands

register_social_commands(
    channels: ChannelSystem,
    groups: GroupSystem,
    friends: FriendsSystem,
    mail: MailSystem,
) -> None

Register all social commands.

PvP System

Player versus player combat with arenas and rankings.

from maid_classic_rpg.systems.pvp import PvPSystem

# PvP system handles:
# - Duel challenges
# - Arena matchmaking
# - Ranking and leaderboards
# - PvP zones and rules

pvp

PvP systems for player-vs-player combat.

ArenaSystem

ArenaSystem(world: World)

Bases: System

System for managing arena matches and queues.

Features: - Queue system with rating-based matchmaking - Multiple match types (1v1, team, FFA) - Dedicated arena rooms - Rating changes based on match outcome

Arena Lifecycle: 1. QUEUED - Players/teams wait for opponents 2. PREPARING - Match found, players teleport to arena 3. IN_PROGRESS - Match active 4. COMPLETED - Match ends, ratings updated

startup async

startup() -> None

Initialize arena system.

shutdown async

shutdown() -> None

Unsubscribe from events and clean up.

update async

update(delta: float) -> None

Update tick - process queues and match timers.

register_arena_room

register_arena_room(
    room_id: UUID, match_types: list[ArenaMatchType]
) -> None

Register a room as an arena for specific match types.

Parameters:

Name Type Description Default
room_id UUID

Room to register

required
match_types list[ArenaMatchType]

Types of matches this room supports

required

join_queue async

join_queue(
    character_ids: list[UUID], match_type: ArenaMatchType
) -> tuple[bool, str]

Join the arena queue.

Parameters:

Name Type Description Default
character_ids list[UUID]

Characters joining as a team/solo

required
match_type ArenaMatchType

Type of match to queue for

required

Returns:

Type Description
tuple[bool, str]

Tuple of (success, message)

leave_queue async

leave_queue(character_id: UUID) -> tuple[bool, str]

Leave the arena queue.

Parameters:

Name Type Description Default
character_id UUID

Character leaving queue

required

Returns:

Type Description
tuple[bool, str]

Tuple of (success, message)

forfeit_match async

forfeit_match(character_id: UUID) -> tuple[bool, str]

Forfeit an active arena match.

Parameters:

Name Type Description Default
character_id UUID

Character forfeiting

required

Returns:

Type Description
tuple[bool, str]

Tuple of (success, message)

get_queue_position

get_queue_position(
    character_id: UUID,
) -> tuple[int, int] | None

Get queue position for a character.

Returns:

Type Description
tuple[int, int] | None

Tuple of (position, total_in_queue) or None if not queued

get_active_match

get_active_match(character_id: UUID) -> ArenaMatch | None

Get active match for a character.

is_in_arena

is_in_arena(character_id: UUID) -> bool

Check if character is in an arena match.

get_queue_status

get_queue_status() -> dict[str, int]

Get current queue sizes for all match types.

DuelSystem

DuelSystem(world: World)

Bases: System

System for managing duels between players.

Duels are consensual PvP fights with special rules: - Both players must agree to fight - Combat stops at a configurable HP threshold (default 1 HP) - No permanent death or item loss - Optional gold betting - Spectators can place bets

Duel Lifecycle: 1. PENDING - Challenger issues challenge, target has time to respond 2. ACCEPTED - Challenge accepted, countdown begins 3. IN_PROGRESS - Duel is active, combat proceeds 4. COMPLETED - Duel ends when HP threshold reached or forfeit

startup async

startup() -> None

Subscribe to relevant events.

shutdown async

shutdown() -> None

Clean up on shutdown.

update async

update(delta: float) -> None

Update tick - handle timeouts and countdowns.

challenge async

challenge(
    challenger_id: UUID,
    target_id: UUID,
    room_id: UUID,
    bet_amount: int = 0,
    stop_at_hp: int | None = None,
) -> tuple[bool, str, UUID | None]

Issue a duel challenge.

Parameters:

Name Type Description Default
challenger_id UUID

Character issuing challenge

required
target_id UUID

Character being challenged

required
room_id UUID

Room where duel will take place

required
bet_amount int

Gold to wager (0 for no bet)

0
stop_at_hp int | None

HP threshold to stop fight

None

Returns:

Type Description
tuple[bool, str, UUID | None]

Tuple of (success, message, duel_id)

accept async

accept(character_id: UUID) -> tuple[bool, str]

Accept a pending duel challenge.

Parameters:

Name Type Description Default
character_id UUID

Character accepting (must be target)

required

Returns:

Type Description
tuple[bool, str]

Tuple of (success, message)

decline async

decline(character_id: UUID) -> tuple[bool, str]

Decline a pending duel challenge.

Parameters:

Name Type Description Default
character_id UUID

Character declining (must be target)

required

Returns:

Type Description
tuple[bool, str]

Tuple of (success, message)

forfeit async

forfeit(character_id: UUID) -> tuple[bool, str]

Forfeit an active duel.

Parameters:

Name Type Description Default
character_id UUID

Character forfeiting

required

Returns:

Type Description
tuple[bool, str]

Tuple of (success, message)

place_bet async

place_bet(
    duel_id: UUID,
    bettor_id: UUID,
    backed_id: UUID,
    amount: int,
) -> tuple[bool, str]

Place a bet on an active duel.

Parameters:

Name Type Description Default
duel_id UUID

Duel to bet on

required
bettor_id UUID

Character placing bet

required
backed_id UUID

Character being bet on

required
amount int

Gold to wager

required

Returns:

Type Description
tuple[bool, str]

Tuple of (success, message)

get_active_duel

get_active_duel(character_id: UUID) -> DuelChallenge | None

Get active duel for a character.

get_pending_challenges

get_pending_challenges(
    character_id: UUID,
) -> list[DuelChallenge]

Get all pending challenges for a character (as target).

is_in_duel

is_in_duel(character_id: UUID) -> bool

Check if character is in an active duel.

get_opponent

get_opponent(character_id: UUID) -> UUID | None

Get duel opponent for a character.

get_countdown

get_countdown(duel_id: UUID) -> float

Get remaining countdown time for a duel.

ArenaMatchEndEvent dataclass

ArenaMatchEndEvent(
    match_id: UUID | None = None,
    match_type: ArenaMatchType | None = None,
    winning_team: str | None = None,
    individual_winner: UUID | None = None,
    rating_changes: dict[str, int] = dict(),
)

Bases: Event

Emitted when an arena match ends.

ArenaMatchFoundEvent dataclass

ArenaMatchFoundEvent(
    match_id: UUID | None = None,
    match_type: ArenaMatchType | None = None,
    team_a: list[UUID] = list(),
    team_b: list[UUID] = list(),
)

Bases: Event

Emitted when a match is found for queued players.

ArenaMatchStateChangedEvent dataclass

ArenaMatchStateChangedEvent(
    match_id: UUID | None = None,
    old_state: ArenaMatchState | None = None,
    new_state: ArenaMatchState | None = None,
)

Bases: Event

Emitted when an arena match state changes.

ArenaQueueJoinEvent dataclass

ArenaQueueJoinEvent(
    character_ids: list[UUID] = list(),
    match_type: ArenaMatchType | None = None,
)

Bases: Event

Emitted when a player/team joins arena queue.

ArenaQueueLeaveEvent dataclass

ArenaQueueLeaveEvent(
    character_ids: list[UUID] = list(),
    match_type: ArenaMatchType | None = None,
)

Bases: Event

Emitted when a player/team leaves arena queue.

DuelBetPlacedEvent dataclass

DuelBetPlacedEvent(
    duel_id: UUID | None = None,
    bettor_id: UUID | None = None,
    backed_id: UUID | None = None,
    amount: int = 0,
)

Bases: Event

Emitted when a spectator places a bet on a duel.

DuelChallengeEvent dataclass

DuelChallengeEvent(
    duel_id: UUID | None = None,
    challenger_id: UUID | None = None,
    target_id: UUID | None = None,
    room_id: UUID | None = None,
    bet_amount: int = 0,
)

Bases: Event

Emitted when a duel challenge is issued.

DuelEndEvent dataclass

DuelEndEvent(
    duel_id: UUID | None = None,
    winner_id: UUID | None = None,
    loser_id: UUID | None = None,
    reason: str = "",
)

Bases: Event

Emitted when a duel ends.

DuelResponseEvent dataclass

DuelResponseEvent(
    duel_id: UUID | None = None,
    challenger_id: UUID | None = None,
    target_id: UUID | None = None,
    accepted: bool = False,
)

Bases: Event

Emitted when a duel challenge is accepted or declined.

DuelStartEvent dataclass

DuelStartEvent(
    duel_id: UUID | None = None,
    challenger_id: UUID | None = None,
    target_id: UUID | None = None,
    room_id: UUID | None = None,
)

Bases: Event

Emitted when a duel begins.

PvPDeathEvent dataclass

PvPDeathEvent(
    victim_id: UUID | None = None,
    killer_id: UUID | None = None,
    room_id: UUID | None = None,
    loot_dropped: list[UUID] = list(),
)

Bases: Event

Emitted when a player dies in PvP.

PvPFlagChangedEvent dataclass

PvPFlagChangedEvent(
    character_id: UUID | None = None,
    old_flag: PvPFlag | None = None,
    new_flag: PvPFlag | None = None,
)

Bases: Event

Emitted when a character's PvP flag changes.

PvPFlaggedEvent dataclass

PvPFlaggedEvent(
    character_id: UUID | None = None,
    target_id: UUID | None = None,
    duration_seconds: int = 0,
)

Bases: Event

Emitted when a character becomes temporarily flagged.

PvPKillEvent dataclass

PvPKillEvent(
    killer_id: UUID | None = None,
    victim_id: UUID | None = None,
    room_id: UUID | None = None,
    kill_streak: int = 0,
)

Bases: Event

Emitted when a player kills another player.

RankingChangedEvent dataclass

RankingChangedEvent(
    character_id: UUID | None = None,
    old_rating: int = 0,
    new_rating: int = 0,
    old_rank: int = 0,
    new_rank: int = 0,
)

Bases: Event

Emitted when a player's ranking changes.

SeasonEndEvent dataclass

SeasonEndEvent(
    season_id: UUID | None = None,
    season_name: str = "",
    final_rankings: list[tuple[UUID, int, int]] = list(),
)

Bases: Event

Emitted when a PvP season ends.

TitleEarnedEvent dataclass

TitleEarnedEvent(
    character_id: UUID | None = None,
    title_id: str = "",
    title_name: str = "",
)

Bases: Event

Emitted when a player earns a PvP title.

ArenaMatch

Bases: MAIDBaseModel

An arena match instance.

all_participants property

all_participants: list[UUID]

Get all participants in the match.

ArenaMatchState

Bases: str, Enum

Arena match lifecycle states.

ArenaMatchType

Bases: str, Enum

Types of arena matches.

ArenaQueueEntry

Bases: MAIDBaseModel

An entry in the arena queue.

team_size property

team_size: int

Get number of players in this entry.

DuelBet

Bases: MAIDBaseModel

A bet placed on a duel by a spectator.

DuelChallenge

Bases: MAIDBaseModel

A duel challenge between two players.

is_expired

is_expired() -> bool

Check if challenge has expired.

DuelState

Bases: str, Enum

Duel lifecycle states.

PvPCharacterState

Bases: MAIDBaseModel

PvP state for a single character.

is_pvp_enabled property

is_pvp_enabled: bool

Check if character can engage in PvP.

is_flagged property

is_flagged: bool

Check if character is temporarily flagged.

kill_death_ratio property

kill_death_ratio: float

Calculate K/D ratio.

PvPFlag

Bases: str, Enum

PvP flag states.

PvPLootMode

Bases: str, Enum

Loot rules for PvP death.

PvPSeason

Bases: MAIDBaseModel

A competitive PvP season.

is_current

is_current() -> bool

Check if this season is currently active.

PvPTitle

Bases: BaseModel

A title earned through PvP achievements.

RankingEntry

Bases: MAIDBaseModel

A single ranking entry on the leaderboard.

win_rate property

win_rate: float

Calculate win percentage.

PvPSystem

PvPSystem(world: World)

Bases: System

System for managing PvP state, rules, and combat modifications.

Responsibilities: - Track PvP flag state per character - Validate PvP combat is allowed - Apply PvP-specific combat modifiers - Handle PvP death consequences - Manage temporary flagging

The PvP system hooks into combat events to enforce rules and track outcomes for the ranking system.

startup async

startup() -> None

Subscribe to combat events.

shutdown async

shutdown() -> None

Clean up subscriptions.

update async

update(delta: float) -> None

Update tick - check for expired flags and protections.

get_state

get_state(character_id: UUID) -> PvPCharacterState

Get or create PvP state for a character.

enable_pvp async

enable_pvp(character_id: UUID) -> bool

Enable PvP for a character.

Parameters:

Name Type Description Default
character_id UUID

Character to enable PvP for

required

Returns:

Type Description
bool

True if PvP was enabled, False if already enabled or blocked

disable_pvp async

disable_pvp(character_id: UUID) -> bool

Disable PvP for a character.

Parameters:

Name Type Description Default
character_id UUID

Character to disable PvP for

required

Returns:

Type Description
bool

True if PvP was disabled, False if flagged or in combat

can_attack

can_attack(
    attacker_id: UUID, target_id: UUID, room_id: UUID
) -> tuple[bool, str]

Check if PvP attack is allowed.

Parameters:

Name Type Description Default
attacker_id UUID

Character attempting attack

required
target_id UUID

Target character

required
room_id UUID

Room where combat would occur

required

Returns:

Type Description
tuple[bool, str]

Tuple of (allowed, reason_if_denied)

get_damage_modifier

get_damage_modifier(
    attacker_id: UUID, target_id: UUID
) -> float

Get damage modifier for PvP combat.

Returns a multiplier to apply to damage dealt. This allows balancing PvP separately from PvE.

Parameters:

Name Type Description Default
attacker_id UUID

Character dealing damage

required
target_id UUID

Character receiving damage

required

Returns:

Type Description
float

Damage multiplier (e.g., 0.7 for 70% damage)

is_pvp_combat

is_pvp_combat(attacker_id: UUID, target_id: UUID) -> bool

Check if combat between two entities is PvP.

is_in_pvp_combat

is_in_pvp_combat(character_id: UUID) -> bool

Check if character is currently in PvP combat.

record_kill async

record_kill(
    killer_id: UUID, victim_id: UUID, room_id: UUID
) -> None

Record a PvP kill and update statistics.

Parameters:

Name Type Description Default
killer_id UUID

Character who got the kill

required
victim_id UUID

Character who died

required
room_id UUID

Room where kill occurred

required

handle_pvp_death async

handle_pvp_death(
    victim_id: UUID, killer_id: UUID | None, room_id: UUID
) -> list[UUID]

Handle PvP death consequences.

Parameters:

Name Type Description Default
victim_id UUID

Character who died

required
killer_id UUID | None

Character who killed them (if any)

required
room_id UUID

Room where death occurred

required

Returns:

Type Description
list[UUID]

List of item IDs that were dropped as loot

get_protection_remaining

get_protection_remaining(character_id: UUID) -> float

Get remaining protection time in seconds.

Parameters:

Name Type Description Default
character_id UUID

Character to check

required

Returns:

Type Description
float

Seconds remaining, or 0 if not protected

get_flag_remaining

get_flag_remaining(character_id: UUID) -> float

Get remaining flag time in seconds.

Parameters:

Name Type Description Default
character_id UUID

Character to check

required

Returns:

Type Description
float

Seconds remaining, or 0 if not flagged

RankingSystem

RankingSystem(world: World)

Bases: System, StorageAware

System for managing PvP rankings and leaderboards.

Features: - Arena rating leaderboards - Kill/death rankings - Seasonal resets - Title rewards

Rankings are updated in response to PvP events and periodically persisted to the database.

set_storage

set_storage(store: DocumentStore) -> None

Receive storage from the engine via StorageAware protocol.

startup async

startup() -> None

Subscribe to PvP events and load data.

shutdown async

shutdown() -> None

Unsubscribe from events and persist rankings before shutdown.

update async

update(delta: float) -> None

Periodic persistence and season checks.

get_ranking

get_ranking(character_id: UUID) -> RankingEntry | None

Get ranking entry for a character.

get_leaderboard

get_leaderboard(
    offset: int = 0,
    limit: int = 10,
    sort_by: str = "rating",
) -> list[RankingEntry]

Get leaderboard entries.

Parameters:

Name Type Description Default
offset int

Starting position

0
limit int

Maximum entries to return

10
sort_by str

Field to sort by ('rating', 'wins', 'kills')

'rating'

Returns:

Type Description
list[RankingEntry]

List of ranking entries

get_rank

get_rank(character_id: UUID) -> int

Get a character's current rank position.

get_earned_titles

get_earned_titles(character_id: UUID) -> list[PvPTitle]

Get all titles earned by a character.

get_available_titles

get_available_titles(character_id: UUID) -> list[PvPTitle]

Get titles a character qualifies for but hasn't earned.

claim_title async

claim_title(
    character_id: UUID, title_id: str
) -> tuple[bool, str]

Claim an earned title.

Parameters:

Name Type Description Default
character_id UUID

Character claiming title

required
title_id str

Title to claim

required

Returns:

Type Description
tuple[bool, str]

Tuple of (success, message)

get_current_season

get_current_season() -> PvPSeason | None

Get the current PvP season.

start_season async

start_season(
    name: str,
    duration_days: int = 90,
    reward_tiers: dict[str, Any] | None = None,
) -> PvPSeason

Start a new PvP season.

Parameters:

Name Type Description Default
name str

Season name

required
duration_days int

Length of season

90
reward_tiers dict[str, Any] | None

Rewards by rank tier

None

Returns:

Type Description
PvPSeason

The new season

World System

Time cycles, weather, and dynamic world events.

from maid_classic_rpg.systems.world import WorldSystem

# World system handles:
# - Day/night cycles
# - Weather patterns
# - Seasonal changes
# - World events

world

World dynamics systems for MAID.

This package contains systems for managing dynamic world elements: - Time system: Game clock, day/night cycles, seasons - Weather system: Regional weather patterns with gameplay effects - World events: Random encounters, world bosses, invasions - Area resets: Periodic respawning of monsters and items - Ecosystem: Predator/prey relationships, population dynamics

EcosystemSystem

EcosystemSystem(
    world: World, config: EcosystemConfig | None = None
)

Bases: System

System for simulating dynamic ecosystems.

Manages predator/prey relationships, resource consumption, reproduction, migration, and population dynamics.

startup async

startup() -> None

Initialize ecosystem system.

shutdown async

shutdown() -> None

Shut down ecosystem system.

update async

update(delta: float) -> None

Update ecosystem simulation.

register_species

register_species(species: SpeciesDefinition) -> None

Register a species definition.

unregister_species

unregister_species(species_id: UUID) -> bool

Unregister a species definition.

add_resource_node

add_resource_node(resource: ResourceNode) -> None

Add a resource node to the ecosystem.

remove_resource_node

remove_resource_node(
    room_id: UUID, resource_type: ResourceType
) -> bool

Remove a resource node from the ecosystem.

initialize_population

initialize_population(
    area_id: UUID, species_id: UUID, initial_count: int
) -> PopulationState | None

Initialize a population for a species in an area.

get_population_info

get_population_info(area_id: UUID) -> dict[str, Any]

Get population information for an area.

get_ecosystem_status

get_ecosystem_status() -> dict[str, Any]

Get overall ecosystem status.

get_resource_info

get_resource_info(area_id: UUID) -> dict[str, Any]

Get resource information for an area.

WorldEventSystem

WorldEventSystem(world: World)

Bases: System

System for managing world events.

Handles event scheduling, triggering, progression, and rewards.

startup async

startup() -> None

Initialize the event system.

shutdown async

shutdown() -> None

Shut down the event system.

update async

update(delta: float) -> None

Update event system.

register_definition

register_definition(
    definition: WorldEventDefinition,
) -> None

Register an event definition.

get_active_events

get_active_events() -> list[WorldEventInstance]

Get all currently active events.

get_event

get_event(instance_id: UUID) -> WorldEventInstance | None

Get a specific event instance.

get_definition

get_definition(
    definition_id: UUID,
) -> WorldEventDefinition | None

Get an event definition by ID.

trigger_event_by_name async

trigger_event_by_name(name: str) -> UUID | None

Trigger an event by its name (admin command).

cancel_event async

cancel_event(instance_id: UUID) -> bool

Cancel an active event (admin command).

record_kill

record_kill(
    instance_id: UUID, template_id: str, killer_id: UUID
) -> None

Record a kill for event progress tracking.

schedule_event

schedule_event(
    definition_id: UUID, game_time_minutes: int
) -> bool

Schedule an event to trigger at a specific game time.

get_event_info

get_event_info(instance_id: UUID) -> dict[str, Any] | None

Get detailed information about an active event.

AreaResetSystem

AreaResetSystem(world: World)

Bases: System

System for managing area resets.

Handles periodic respawning of monsters, items, and restoration of room states.

startup async

startup() -> None

Initialize reset system.

shutdown async

shutdown() -> None

Shut down reset system.

update async

update(delta: float) -> None

Check and execute area resets.

register_config

register_config(config: AreaResetConfig) -> None

Register a reset configuration for an area.

unregister_config

unregister_config(area_id: UUID) -> bool

Unregister a reset configuration.

force_reset

force_reset(area_id: UUID) -> bool

Force an immediate area reset (admin command).

get_reset_info

get_reset_info(area_id: UUID) -> dict[str, Any] | None

Get reset information for an area.

get_all_configs

get_all_configs() -> list[AreaResetConfig]

Get all registered reset configurations.

set_enabled

set_enabled(area_id: UUID, enabled: bool) -> bool

Enable or disable resets for an area.

GameTimeSystem

GameTimeSystem(
    world: World, config: TimeConfig | None = None
)

Bases: System

System managing game time progression.

The time system advances game time based on real elapsed time, scaled by a configurable ratio. It emits events for significant time changes that other systems can react to.

Example

system = GameTimeSystem(world)

Get current time

current = system.current_time print(f"It is {current.format_time()}")

Check time of day

if current.is_night: print("Darkness surrounds you.")

current_time property

current_time: GameTime

Get the current game time.

config property

config: TimeConfig

Get the time configuration.

time_of_day property

time_of_day: TimeOfDay

Get current time of day.

season property

season: Season

Get current season.

is_night property

is_night: bool

Check if it's night.

set_storage

set_storage(store: DocumentStore) -> None

Set the document store for persistence.

startup async

startup() -> None

Initialize the time system, loading saved time if available.

shutdown async

shutdown() -> None

Shut down the time system, saving current time.

update async

update(delta: float) -> None

Advance game time based on elapsed real time.

Parameters:

Name Type Description Default
delta float

Real seconds since last tick

required

set_time

set_time(game_time: GameTime) -> None

Set the current game time directly.

Use for admin commands or loading saved state.

get_visibility_modifier

get_visibility_modifier() -> float

Get visibility modifier based on time of day.

Returns:

Type Description
float

Multiplier for visibility (1.0 = full, 0.0 = none)

WeatherSystem

WeatherSystem(world: World)

Bases: System

System for managing weather across regions.

The weather system tracks weather per region and handles: - Weather state progression - Weather change events - Periodic weather effects (damage, messages) - Weather effect queries for other systems

startup async

startup() -> None

Initialize weather system.

shutdown async

shutdown() -> None

Shut down weather system.

update async

update(delta: float) -> None

Update weather states.

get_weather

get_weather(
    region_id: UUID | None = None,
) -> RegionalWeather

Get weather for a region (or global if not specified).

get_weather_effects

get_weather_effects(
    region_id: UUID | None = None,
) -> WeatherEffect

Get current weather effects for a region.

get_weather_description

get_weather_description(
    region_id: UUID | None = None,
) -> str

Get a text description of current weather.

set_weather

set_weather(
    region_id: UUID | None, weather_type: WeatherType
) -> None

Manually set weather (admin command).

register_region

register_region(config: RegionWeatherConfig) -> None

Register a region with custom weather configuration.

NPC System

NPC behavior, schedules, and AI-powered dialogue.

from maid_classic_rpg.systems.npc import NPCSystem
from maid_classic_rpg.systems.npc.dialogue import DialogueSystem

# NPC system handles:
# - NPC spawning and despawning
# - Behavior patterns
# - Schedules and routines
# - AI-powered conversations

npc

NPC behavior, spawning, dialogue, and autonomy systems.

AutonomySystem

AutonomySystem(world: Any)

Bases: System

Selects and executes NPC autonomous actions each tick.

startup async

startup() -> None

Restore persisted demotion/failure state if available.

update async

update(delta: float) -> None

Run autonomy processing within tick-budget constraints.

catch_up_npc async

catch_up_npc(
    npc_id: UUID,
    elapsed_game_hours: float,
    schedule: ScheduleComponent,
    needs: NeedsComponent,
    goals: GoalsComponent,
) -> None

Analytically advance NPC state after background period.

compute_daily_need_delta

compute_daily_need_delta(
    schedule: ScheduleComponent, needs: NeedsComponent
) -> dict[NeedCategory, float]

Pre-compute net need delta for a 24-hour cycle.

BarkLibrary

BarkLibrary(
    templates: dict[str, list[BarkTemplate]] | None = None,
)

Loads and selects bark templates.

set_templates

set_templates(
    templates: dict[str, list[BarkTemplate]],
) -> None

Replace bark templates.

select_bark

select_bark(
    needs: NeedsComponent,
    mood: float,
    archetype_id: str,
    npc_id: UUID | None = None,
    now_minutes: float | None = None,
) -> str | None

Select bark text based on needs, mood, and cooldown.

BarkSystem

BarkSystem(world: Any, library: BarkLibrary | None = None)

Bases: System

Emits ambient NPC barks in rooms containing players.

BehaviorSystem

BehaviorSystem(world: World)

Bases: System

System for executing NPC behaviors.

Processes: - Aggression checks (aggressive NPCs attack on sight) - Patrol/wander movement - Flee behavior when low health - Territory defense - Pack/ally assistance

startup async

startup() -> None

Initialize behavior system.

shutdown async

shutdown() -> None

Shutdown behavior system.

register_npc

register_npc(
    entity_id: UUID,
    behavior: BehaviorConfig,
    territory_center: UUID | None = None,
) -> NPCState

Register an NPC for behavior processing.

Parameters:

Name Type Description Default
entity_id UUID

NPC entity ID

required
behavior BehaviorConfig

Behavior configuration

required
territory_center UUID | None

Optional territory center room

None

Returns:

Type Description
NPCState

The created NPCState

unregister_npc

unregister_npc(entity_id: UUID) -> None

Unregister an NPC from behavior processing.

get_npc_state

get_npc_state(entity_id: UUID) -> NPCState | None

Get the state for an NPC.

get_behavior_config

get_behavior_config(
    entity_id: UUID,
) -> BehaviorConfig | None

Get the behavior configuration for an NPC.

update async

update(delta: float) -> None

Process NPC behaviors.

check_aggro async

check_aggro(
    npc_id: UUID,
    npc_room_id: UUID,
    behavior: BehaviorType,
    _aggro_range: int = 1,
) -> list[UUID]

Check for entities that should trigger aggro.

Parameters:

Name Type Description Default
npc_id UUID

NPC checking for targets

required
npc_room_id UUID

NPC's current room

required
behavior BehaviorType

NPC's behavior type

required
_aggro_range int

Rooms to check

1

Returns:

Type Description
list[UUID]

List of entity IDs to aggro on

should_flee async

should_flee(
    _npc_id: UUID,
    current_health: int,
    max_health: int,
    flee_threshold: float,
) -> bool

Check if NPC should attempt to flee.

Parameters:

Name Type Description Default
_npc_id UUID

NPC to check

required
current_health int

Current HP

required
max_health int

Maximum HP

required
flee_threshold float

Health percentage to flee at

required

Returns:

Type Description
bool

True if should flee

choose_patrol_destination async

choose_patrol_destination(
    _npc_id: UUID,
    _current_room_id: UUID,
    patrol_rooms: list[UUID],
    patrol_index: int,
) -> tuple[UUID | None, int]

Choose next patrol destination.

Parameters:

Name Type Description Default
_npc_id UUID

NPC patrolling

required
_current_room_id UUID

Current room

required
patrol_rooms list[UUID]

Patrol route

required
patrol_index int

Current position in route

required

Returns:

Type Description
tuple[UUID | None, int]

Tuple of (destination_room_id, new_patrol_index)

choose_wander_destination async

choose_wander_destination(
    _npc_id: UUID,
    current_room_id: UUID,
    wander_chance: float,
    territory_center: UUID | None = None,
    territory_radius: int = 3,
) -> UUID | None

Choose random wander destination.

Parameters:

Name Type Description Default
_npc_id UUID

NPC wandering

required
current_room_id UUID

Current room

required
wander_chance float

Probability to wander

required
territory_center UUID | None

Center of allowed territory

None
territory_radius int

Rooms from center

3

Returns:

Type Description
UUID | None

Destination room ID or None

call_for_help async

call_for_help(
    npc_id: UUID,
    npc_room_id: UUID,
    target_id: UUID,
    _call_radius: int = 1,
) -> list[UUID]

Call nearby allies for help.

Parameters:

Name Type Description Default
npc_id UUID

NPC calling for help

required
npc_room_id UUID

NPC's room

required
target_id UUID

Entity to aggro allies on

required
_call_radius int

Rooms to check for allies

1

Returns:

Type Description
list[UUID]

List of ally IDs that responded

set_in_combat

set_in_combat(npc_id: UUID, in_combat: bool) -> None

Set whether an NPC is in combat.

update_threat

update_threat(
    npc_id: UUID, entity_id: UUID, threat: int
) -> None

Update threat level for an entity.

record_sighting

record_sighting(npc_id: UUID, entity_id: UUID) -> None

Record when an NPC sees an entity.

NPCDialogueSystem

NPCDialogueSystem(world: World)

Bases: System, StorageAware

System for handling AI-powered NPC dialogue.

Manages conversations between players and NPCs, including: - AI provider integration for generating responses - Rate limiting to prevent abuse - Conversation history tracking - Context building from world state

Example

system = NPCDialogueSystem(world)

Process a player's message to an NPC

success = await system.process_dialogue( player_entity_id=player.id, npc_entity_id=npc.id, message="Hello, how are you?", session=player_session, )

Parameters:

Name Type Description Default
world World

The game world instance

required

set_storage

set_storage(store: DocumentStore) -> None

Set the document store for conversation persistence.

Called automatically by the engine when the system is registered if the system implements StorageAware.

Parameters:

Name Type Description Default
store DocumentStore

The document store instance

required

startup async

startup() -> None

Initialize the dialogue system.

Sets up rate limiting and gets provider registry from engine.

shutdown async

shutdown() -> None

Shut down the dialogue system.

Saves conversations to storage (if persistence enabled) and cleans up.

update async

update(delta: float) -> None

Periodic update for cleanup of stale conversations.

Parameters:

Name Type Description Default
delta float

Time since last update in seconds

required

process_dialogue async

process_dialogue(
    player_entity_id: UUID,
    npc_entity_id: UUID,
    message: str,
    session: Any,
) -> bool

Process a dialogue request from player to NPC.

This is the main entry point for NPC dialogue. It handles: - Validating the NPC has a DialogueComponent - Checking rate limits and cooldowns - Building context from world state - Generating AI response - Sending the complete response to player (buffered for content filtering)

Parameters:

Name Type Description Default
player_entity_id UUID

UUID of the player entity

required
npc_entity_id UUID

UUID of the NPC entity

required
message str

The player's message to the NPC

required
session Any

Player's session for sending responses

required

Returns:

Type Description
bool

True if dialogue was processed, False on error

find_npc_in_room

find_npc_in_room(
    room_id: UUID, keyword: str
) -> Entity | None

Find an NPC in a room by keyword.

Searches for NPCs in the specified room that match the given keyword in their name or keywords list.

Parameters:

Name Type Description Default
room_id UUID

The room to search in

required
keyword str

Keyword to match against NPC names/keywords

required

Returns:

Type Description
Entity | None

The matching Entity or None if not found

get_conversation async

get_conversation(
    player_id: UUID, npc_id: UUID
) -> Conversation | None

Get an existing conversation between player and NPC.

Parameters:

Name Type Description Default
player_id UUID

UUID of the player

required
npc_id UUID

UUID of the NPC

required

Returns:

Type Description
Conversation | None

The Conversation if it exists, None otherwise

get_player_conversations async

get_player_conversations(
    player_id: UUID,
) -> list[Conversation]

Get all active conversations for a player.

Parameters:

Name Type Description Default
player_id UUID

UUID of the player

required

Returns:

Type Description
list[Conversation]

List of Conversation objects for this player

get_entity_name

get_entity_name(entity_id: UUID) -> str

Get the display name for an entity.

Parameters:

Name Type Description Default
entity_id UUID

UUID of the entity

required

Returns:

Type Description
str

Entity's name or "Someone" if not found

end_conversation async

end_conversation(player_id: UUID, npc_id: UUID) -> bool

End a conversation between player and NPC.

Parameters:

Name Type Description Default
player_id UUID

UUID of the player

required
npc_id UUID

UUID of the NPC

required

Returns:

Type Description
bool

True if a conversation was ended, False if not found

ActionExecutor

Executes autonomy actions by mutating intent components.

execute async

execute(
    npc_id: UUID, scored_action: ScoredAction, world: Any
) -> ActionResult

Execute a scored action for an NPC.

NPCMovementHandler

Bases: System

Moves NPCs one step per tick based on NavigationIntent.

update async

update(delta: float) -> None

Advance NPC navigation intents by one path step.

FallbackResponseProvider

Provides contextually appropriate canned responses when AI is unavailable.

This provider selects responses based on the NPC's personality type (derived from their role) and the interaction context. Responses are chosen randomly within the matching category to avoid repetition.

When a fallback is used, the provider logs the event with the reason for monitoring and debugging AI provider health.

Example

provider = FallbackResponseProvider()

Get a fallback response for a merchant greeting

response = provider.get_response( personality="merchant", context="greeting", )

Log the fallback usage

provider.log_fallback( npc_name="Grimjaw", reason=FallbackReason.TIMEOUT, personality="merchant", context="greeting", )

get_response

get_response(
    personality: str = "", context: str = ""
) -> str

Get a contextually appropriate fallback response.

Resolves the personality type from the NPC role string and the interaction context, then selects a random response from the matching category.

Parameters:

Name Type Description Default
personality str

NPC role/personality string (e.g., "merchant", "guard"). Maps to a PersonalityType via resolve_personality().

''
context str

Interaction context string (e.g., "greeting", "trade"). Maps to an InteractionContext via resolve_context().

''

Returns:

Type Description
str

A fallback response string appropriate for the NPC and context.

log_fallback

log_fallback(
    npc_name: str,
    reason: FallbackReason,
    personality: str = "",
    context: str = "",
    error_detail: str = "",
) -> None

Log that a fallback response was used.

This method logs at WARNING level to aid in monitoring AI provider health. The log includes the NPC name, the reason for fallback, and additional context about the failure.

Parameters:

Name Type Description Default
npc_name str

Display name of the NPC.

required
reason FallbackReason

Why the fallback was triggered (from FallbackReason).

required
personality str

NPC personality/role string.

''
context str

Interaction context string.

''
error_detail str

Additional detail about the error (e.g., exception message).

''

get_response_and_log

get_response_and_log(
    npc_name: str,
    reason: FallbackReason,
    personality: str = "",
    context: str = "",
    error_detail: str = "",
) -> str

Get a fallback response and log the event in one call.

Convenience method that combines get_response() and log_fallback().

Parameters:

Name Type Description Default
npc_name str

Display name of the NPC.

required
reason FallbackReason

Why the fallback was triggered.

required
personality str

NPC role/personality string.

''
context str

Interaction context string.

''
error_detail str

Additional error detail.

''

Returns:

Type Description
str

A fallback response string appropriate for the NPC and context.

NeedDecaySystem

NeedDecaySystem(
    world: Any,
    memory_provider: MemoryProvider | None = None,
)

Bases: System

Decays NPC needs and updates mood.

update async

update(delta: float) -> None

Decay needs by elapsed game hours and emit mood changes.

ScheduleSystem

ScheduleSystem(world: Any)

Bases: System

Applies schedule blocks to NPCs as world time changes.

startup async

startup() -> None

Subscribe to time and weather events.

shutdown async

shutdown() -> None

Unsubscribe event handlers.

update async

update(delta: float) -> None

Schedule reacts via event handlers only.

StorySignalSystem

StorySignalSystem(
    world: Any, detector: StorySignalDetector | None = None
)

Bases: System

Periodic detector execution and signal emission.

SocialFabricSystem

SocialFabricSystem(
    world: Any,
    relationship_provider: RelationshipProvider
    | None = None,
    resolver: SocialInteractionResolver | None = None,
)

Bases: System

Processes social intents and relationship-derived interaction state.

update async

update(delta: float) -> None

Sweep reservations and process social intents.

get_friends

get_friends(npc_id: UUID) -> list[UUID]

Return friend IDs where trust > 0.7.

get_rivals

get_rivals(npc_id: UUID) -> list[UUID]

Return rival IDs where trust and respect are both low.

get_allies

get_allies(npc_id: UUID) -> list[UUID]

Return allies based on loyalty relation if available.

reserve_interaction

reserve_interaction(
    initiator_id: UUID, target_id: UUID, duration: float
) -> bool

Reserve interaction target for a duration.

SpawnerSystem

SpawnerSystem(world: World)

Bases: System

System for spawning and respawning NPCs/monsters.

Handles: - Initial spawning at startup - Respawning after death (with timers) - Population limits per room and area - Spawn conditions (time, player presence)

startup async

startup() -> None

Initialize spawner and do initial spawns.

shutdown async

shutdown() -> None

Shutdown spawner system.

register_spawn_point

register_spawn_point(spawn_point: SpawnPoint) -> None

Register a spawn point.

unregister_spawn_point

unregister_spawn_point(
    spawn_point_id: UUID,
) -> SpawnPoint | None

Unregister a spawn point.

register_template

register_template(template: MonsterTemplate) -> None

Register a monster template.

unregister_template

unregister_template(
    template_id: UUID,
) -> MonsterTemplate | None

Unregister a monster template.

get_spawn_point

get_spawn_point(spawn_point_id: UUID) -> SpawnPoint | None

Get a spawn point by ID.

get_template

get_template(template_id: UUID) -> MonsterTemplate | None

Get a monster template by ID.

get_monster

get_monster(entity_id: UUID) -> Monster | None

Get a spawned monster by ID.

register_spawn_rule

register_spawn_rule(rule: SpawnRule) -> None

Register a spawn rule that governs spawning behavior.

unregister_spawn_rule

unregister_spawn_rule(rule_id: UUID) -> SpawnRule | None

Unregister a spawn rule.

register_spawn_group

register_spawn_group(group: SpawnGroup) -> None

Register a coordinated spawn group.

unregister_spawn_group

unregister_spawn_group(group_id: UUID) -> SpawnGroup | None

Unregister a spawn group.

get_spawn_rule

get_spawn_rule(rule_id: UUID) -> SpawnRule | None

Get a spawn rule by ID.

get_spawn_group

get_spawn_group(group_id: UUID) -> SpawnGroup | None

Get a spawn group by ID.

update async

update(delta: float) -> None

Process spawn timers and respawns.

on_entity_death async

on_entity_death(entity_id: UUID) -> None

Handle entity death for respawn tracking.

Parameters:

Name Type Description Default
entity_id UUID

Entity that died

required

despawn_entity async

despawn_entity(
    entity_id: UUID, reason: str = "cleanup"
) -> bool

Manually despawn an entity.

Parameters:

Name Type Description Default
entity_id UUID

Entity to despawn

required
reason str

Reason for despawn

'cleanup'

Returns:

Type Description
bool

True if entity was despawned

get_spawn_point_status

get_spawn_point_status(
    spawn_point_id: UUID,
) -> dict[str, object] | None

Get status of a spawn point.

Returns:

Type Description
dict[str, object] | None

Dict with spawn point status or None

get_all_spawn_points

get_all_spawn_points() -> list[SpawnPoint]

Get all registered spawn points.

get_all_monsters

get_all_monsters() -> list[Monster]

Get all spawned monsters.

get_monsters_in_room

get_monsters_in_room(room_id: UUID) -> list[Monster]

Get all monsters in a specific room.

NPC Autonomy

NPC autonomy systems providing independent NPC behavior including needs, goals, schedules, social interactions, ambient barks, action scoring, signal processing, gossip propagation, and LLM-driven decision queuing.

Submodules: autonomy, needs, goals, schedule, social, barks, executor, scoring, signals, gossip, llm_queue.

autonomy

Core autonomy decision loop with tick-budget safeguards.

TickBudgetGovernor

TickBudgetGovernor()

Tracks autonomy budget pressure and dynamic demotions.

dynamic_demotions property

dynamic_demotions: set[UUID]

Get currently demoted NPC IDs.

should_process

should_process(npc_id: UUID, elapsed_ms: float) -> bool

Check whether processing may continue for current tick budget.

mark_demoted

mark_demoted(npc_id: UUID) -> None

Mark an NPC as dynamically demoted.

end_tick

end_tick(elapsed_ms: float) -> None

Finalize tick accounting and clear demotions on recovery.

AutonomySystem

AutonomySystem(world: Any)

Bases: System

Selects and executes NPC autonomous actions each tick.

startup async

startup() -> None

Restore persisted demotion/failure state if available.

update async

update(delta: float) -> None

Run autonomy processing within tick-budget constraints.

catch_up_npc async

catch_up_npc(
    npc_id: UUID,
    elapsed_game_hours: float,
    schedule: ScheduleComponent,
    needs: NeedsComponent,
    goals: GoalsComponent,
) -> None

Analytically advance NPC state after background period.

compute_daily_need_delta

compute_daily_need_delta(
    schedule: ScheduleComponent, needs: NeedsComponent
) -> dict[NeedCategory, float]

Pre-compute net need delta for a 24-hour cycle.

Quest Generation

Automated quest generation integration, connecting world state and NPC needs to the quest generation framework in maid-stdlib.

generation

Automated quest generation package.

QuestHistory

QuestHistory(document_store: DocumentStore | None)

In-memory history index with optional DocumentStore persistence.

record

record(quest: GeneratedQuest) -> None

Record newly generated quest metadata.

mark_offered

mark_offered(character_id: UUID, quest_id: str) -> None

Track a pending quest offer for a character.

mark_accepted

mark_accepted(character_id: UUID, quest_id: str) -> None

Mark quest as accepted by character.

set_outcome

set_outcome(
    quest_id: str,
    outcome: QuestOutcome,
    character_id: UUID | None = None,
) -> None

Set quest outcome and completion timestamp.

recent_quest_from_npc

recent_quest_from_npc(
    npc_id: UUID, window_seconds: float
) -> bool

Return True when NPC gave a quest within cooldown window.

recent_types

recent_types(count: int) -> list[QuestSeedType]

Return most recent seed types.

active_generated_quest_ids

active_generated_quest_ids() -> set[str]

Get active generated quest identifiers.

get

get(quest_id: str) -> QuestHistoryEntry | None

Get quest history entry by quest ID.

get_chain

get_chain(chain_id: str) -> list[QuestHistoryEntry]

Return history entries belonging to a quest chain.

count_active_in_region

count_active_in_region(region_id: str) -> int

Count active generated quests in a region.

count_pending_offers

count_pending_offers(character_id: UUID) -> int

Count pending generated quest offers for character.

GeneratedQuest dataclass

GeneratedQuest(
    id: str,
    seed_id: UUID,
    seed_type: QuestSeedType,
    archetype_id: str,
    importance: float,
    title: str,
    summary: str,
    detailed_description: str,
    objectives: list[QuestObjective],
    quest_giver_id: UUID,
    quest_giver_motivation: str,
    quest_giver_dialogue: QuestDialogue,
    involved_npcs: dict[UUID, str],
    involved_locations: list[UUID],
    rewards: QuestReward,
    consequences: list[QuestConsequence],
    failure_consequences: list[QuestConsequence],
    difficulty: DifficultyTier,
    time_limit: float | None,
    expiry: datetime,
    chain_potential: float,
    chain_depth: int = 0,
    grounding_score: float = 1.0,
    coherence_score: float = 1.0,
    novelty_score: float = 1.0,
    delivery_method: QuestDeliveryMethod = DIRECT_OFFER,
)

Fully built generated quest payload before conversion to Quest model.

QuestOutcome

Bases: str, Enum

Outcome grade emitted for quest completion/failure.

QuestSeed dataclass

QuestSeed(
    source_signals: list[str],
    seed_type: QuestSeedType,
    importance: float,
    urgency: float,
    primary_npc: UUID | None,
    primary_npc_motivation: str,
    backstory: str,
    stakes: str,
    location: UUID | None,
    threat_source: str,
    evidence_location: UUID | None,
    involved_npcs: list[UUID],
    context_keys: dict[str, str],
    created_at: datetime,
    expires_at: datetime | None,
    attempt_count: int = 0,
    chain_depth: int = 0,
    seed_id: UUID = uuid4(),
)

Input signal bundle used to generate a quest.

QuestSeedTypes

Known quest seed type identifiers.

Skills System

Character skills, experience, and progression.

from maid_classic_rpg.systems.skills import SkillsSystem

# Skills system handles:
# - Skill learning
# - Experience tracking
# - Level progression
# - Skill trees and prerequisites

skills

Skill systems for training and skill checks.

CheckDifficulty

Bases: Enum

Difficulty levels for skill checks.

SkillCheckResult dataclass

SkillCheckResult(
    success: bool = False,
    roll: int = 0,
    target: int = 0,
    margin: int = 0,
    critical: bool = False,
    skill_level: int = 0,
)

Result of a skill check.

SkillLearnedEvent dataclass

SkillLearnedEvent(character_id: UUID, skill_name: str)

Bases: Event

Emitted when a character learns a skill.

SkillLevelUpEvent dataclass

SkillLevelUpEvent(
    character_id: UUID, skill_name: str, new_level: int
)

Bases: Event

Emitted when a character's skill levels up.

SkillSystem

SkillSystem(world: World)

Bases: System, StorageAware

Manages skill training, checks, and progression.

set_storage

set_storage(store: DocumentStore) -> None

Set the document store for persistence.

startup async

startup() -> None

Load skill data from storage.

shutdown async

shutdown() -> None

Save skill data and clean up.

update async

update(_delta: float) -> None

Process any time-based skill updates and clean up cooldowns.

learn_skill

learn_skill(character_id: UUID, skill_name: str) -> bool

Teach a skill to a character.

Parameters:

Name Type Description Default
character_id UUID

Character to teach

required
skill_name str

Skill internal name

required

Returns:

Type Description
bool

True if skill was learned (not already known)

get_skill_level

get_skill_level(character_id: UUID, skill_name: str) -> int

Get a character's level in a skill.

get_character_skill

get_character_skill(
    character_id: UUID, skill_name: str
) -> CharacterSkill | None

Get a character's skill data.

get_all_skills

get_all_skills(character_id: UUID) -> list[CharacterSkill]

Get all skills for a character.

get_skills_by_category

get_skills_by_category(
    character_id: UUID, category: SkillCategory
) -> list[CharacterSkill]

Get skills in a category for a character.

add_experience

add_experience(
    character_id: UUID,
    skill_name: str,
    amount: float,
    respect_cooldown: bool = True,
) -> tuple[float, int]

Add experience to a skill.

Parameters:

Name Type Description Default
character_id UUID

Character ID

required
skill_name str

Skill internal name

required
amount float

Experience to add

required
respect_cooldown bool

If True, check cooldown (1 second between gains)

True

Returns:

Type Description
tuple[float, int]

Tuple of (actual exp added, levels gained)

train_skill

train_skill(
    character_id: UUID,
    skill_name: str,
    training_points: int = 1,
) -> tuple[bool, int]

Train a skill using training points.

Parameters:

Name Type Description Default
character_id UUID

Character ID

required
skill_name str

Skill to train

required
training_points int

Points to spend

1

Returns:

Type Description
tuple[bool, int]

Tuple of (success, levels gained)

skill_check

skill_check(
    character: Character,
    skill_name: str,
    difficulty: CheckDifficulty | int,
    stat_bonus: bool = True,
) -> SkillCheckResult

Perform a skill check.

Parameters:

Name Type Description Default
character Character

Character making the check

required
skill_name str

Skill being checked

required
difficulty CheckDifficulty | int

Target number or CheckDifficulty enum

required
stat_bonus bool

Whether to add stat bonuses

True

Returns:

Type Description
SkillCheckResult

SkillCheckResult with outcome details

opposed_check

opposed_check(
    attacker: Character,
    defender: Character,
    attack_skill: str,
    defend_skill: str,
) -> tuple[SkillCheckResult, SkillCheckResult]

Perform an opposed skill check.

Parameters:

Name Type Description Default
attacker Character

Character initiating the check

required
defender Character

Character defending

required
attack_skill str

Attacker's skill

required
defend_skill str

Defender's skill

required

Returns:

Type Description
tuple[SkillCheckResult, SkillCheckResult]

Tuple of (attacker_result, defender_result)

get_available_skills

get_available_skills(
    character: Character,
) -> list[SkillDefinition]

Get all skills available to a character based on class.

Parameters:

Name Type Description Default
character Character

Character to check

required

Returns:

Type Description
list[SkillDefinition]

List of available skill definitions

get_trainable_skills

get_trainable_skills(
    character: Character,
) -> list[SkillDefinition]

Get skills that can be trained at a guild.

Parameters:

Name Type Description Default
character Character

Character to check

required

Returns:

Type Description
list[SkillDefinition]

List of trainable skill definitions


Commands

Dialogue Commands

Commands for interacting with AI-powered NPCs:

talk <npc> <message>      - Talk to an NPC
ask <npc> about <topic>   - Ask NPC about a topic
greet <npc>               - Greet an NPC (aliases: hello, hi)
conversations             - List active conversations (alias: convos)
endconversation <npc>     - End conversation (aliases: endconv, bye)

dialogue

Dialogue command handlers for AI NPC conversations.

cmd_talk async

cmd_talk(ctx: CommandContext) -> None

Talk to an NPC using AI dialogue.

Usage: talk Example: talk bartender Hello, do you have any news?

cmd_ask async

cmd_ask(ctx: CommandContext) -> None

Ask an NPC about a specific topic.

Usage: ask about Example: ask guard about trouble in town

cmd_conversations async

cmd_conversations(ctx: CommandContext) -> None

List active conversations with NPCs.

cmd_endconversation async

cmd_endconversation(ctx: CommandContext) -> None

End a conversation with an NPC.

Usage: endconversation Example: endconversation bartender

cmd_greet async

cmd_greet(ctx: CommandContext) -> None

Greet an NPC to initiate conversation.

This sends a generic greeting to the NPC, which can be useful to start a conversation or see the NPC's introduction.

Examples:

greet blacksmith hello merchant hi guard

Economy Commands

Commands for economic activities:

buy <item> [from <npc>]   - Purchase an item
sell <item> [to <npc>]    - Sell an item
trade <player>            - Initiate a trade
bank deposit <amount>     - Deposit currency
bank withdraw <amount>    - Withdraw currency
auction list              - View auction house
auction bid <id> <amount> - Bid on an item

economy

Economy command handlers for shops, banking, and trading.

cmd_gold async

cmd_gold(ctx: CommandContext) -> None

Display current gold on hand.

cmd_list async

cmd_list(ctx: CommandContext) -> None

List shop inventory.

cmd_buy async

cmd_buy(ctx: CommandContext) -> None

Buy an item from shop.

cmd_sell async

cmd_sell(ctx: CommandContext) -> None

Sell an item to shop.

cmd_value async

cmd_value(ctx: CommandContext) -> None

Get sale price for an item.

cmd_appraise async

cmd_appraise(ctx: CommandContext) -> None

Get detailed info on shop item.

cmd_balance async

cmd_balance(ctx: CommandContext) -> None

Check bank balance.

cmd_deposit async

cmd_deposit(ctx: CommandContext) -> None

Deposit gold into bank.

cmd_withdraw async

cmd_withdraw(ctx: CommandContext) -> None

Withdraw gold from bank.

cmd_trade async

cmd_trade(ctx: CommandContext) -> None

Start a trade with another player.

cmd_accept async

cmd_accept(ctx: CommandContext) -> None

Accept trade request or confirm trade.

cmd_decline async

cmd_decline(ctx: CommandContext) -> None

Decline or cancel trade.

cmd_offer async

cmd_offer(ctx: CommandContext) -> None

Add to trade offer.

cmd_remove async

cmd_remove(ctx: CommandContext) -> None

Remove from trade offer.

cmd_lock async

cmd_lock(ctx: CommandContext) -> None

Lock trade offer.

cmd_unlock async

cmd_unlock(ctx: CommandContext) -> None

Unlock trade offer.

cmd_tradestatus async

cmd_tradestatus(ctx: CommandContext) -> None

View trade status.

Information Commands

Commands for getting information:

score                     - View character sheet
skills                    - List your skills
quests                    - View quest log
who                       - List online players
time                      - View in-game time
weather                   - View current weather

information

Information command handlers.

cmd_score async

cmd_score(ctx: CommandContext) -> None

Show character stats and vitals.

cmd_inventory async

cmd_inventory(ctx: CommandContext) -> None

Show the player's inventory.

cmd_equipment async

cmd_equipment(ctx: CommandContext) -> None

Show equipped items.

cmd_examine async

cmd_examine(ctx: CommandContext) -> None

Examine something in detail.

cmd_who async

cmd_who(ctx: CommandContext) -> None

Show list of online players.

cmd_help async

cmd_help(ctx: CommandContext) -> None

Show help information.

Displays command documentation using the HelpGenerator system. When called without arguments, lists all available commands grouped by category. When called with a command name, shows detailed help for that specific command.

Examples:

help help look help attack help teleport

See Also

commands


Content Pack

The Classic RPG content pack registers all systems, commands, and game content:

from maid_engine import GameEngine
from maid_stdlib import StdlibContentPack
from maid_classic_rpg import ClassicRPGContentPack

engine = GameEngine(settings)

# Load packs in dependency order
engine.load_content_pack(StdlibContentPack())
engine.load_content_pack(ClassicRPGContentPack())

await engine.start()

ClassicRPGContentPack

ClassicRPGContentPack()

Classic RPG content pack.

Provides a complete traditional MUD experience with: - Combat system with tactical positioning - Magic system with spells and effects - Crafting and gathering - Economy (shops, trading, banking, auctions) - Social features (guilds, factions, achievements) - Quest system with objectives and rewards - PvP (arenas, duels, rankings) - World dynamics (time, weather, ecosystems)

manifest property

Get pack manifest.

get_dependencies

get_dependencies() -> list[str]

Get pack dependencies.

get_systems

get_systems(world: World) -> list[System]

Get systems provided by this pack.

get_events

get_events() -> list[type[Event]]

Get events defined by this pack.

register_commands

register_commands(registry: AnyCommandRegistry) -> None

Register commands provided by this pack.

register_document_schemas

register_document_schemas(store: DocumentStore) -> None

Register document schemas used by this pack.

register_component_types

register_component_types(
    registry: ComponentRegistry,
) -> None

Register component types for persistence serialization.

register_api_routes

register_api_routes(router: APIRouter) -> None

Register API routes. Classic RPG has no custom admin routes.

on_load async

on_load(engine: GameEngine) -> None

Called when pack is loaded.

Initializes game data including: - Quest definitions from data files - Crafting recipes - Default world setup if needed

on_unload async

on_unload(engine: GameEngine) -> None

Called when pack is unloaded.

Performs cleanup: - Saves any pending data - Releases resources


Configuration

Classic RPG specific configuration options:

# Combat settings
MAID_CLASSIC_RPG__COMBAT__TICK_DURATION=6.0
MAID_CLASSIC_RPG__COMBAT__FLEE_CHANCE=0.25

# Economy settings
MAID_CLASSIC_RPG__ECONOMY__STARTING_GOLD=100
MAID_CLASSIC_RPG__ECONOMY__TAX_RATE=0.05

# World settings
MAID_CLASSIC_RPG__WORLD__DAY_DURATION_MINUTES=60
MAID_CLASSIC_RPG__WORLD__WEATHER_CHANGE_INTERVAL=300

Package API

Full package API documentation:

maid_classic_rpg

MAID Classic RPG - Traditional MUD gameplay content pack.

This package provides a complete classic MUD experience:

  • Combat system with tactical grid positioning
  • Magic system with spells and effects
  • Crafting and gathering
  • Economy (shops, trading, banking, auctions)
  • Social features (guilds, factions, achievements)
  • Quest system with objectives and rewards
  • PvP (arenas, duels, rankings)
  • World dynamics (time, weather, ecosystems)

ClassicRPGContentPack

ClassicRPGContentPack()

Classic RPG content pack.

Provides a complete traditional MUD experience with: - Combat system with tactical positioning - Magic system with spells and effects - Crafting and gathering - Economy (shops, trading, banking, auctions) - Social features (guilds, factions, achievements) - Quest system with objectives and rewards - PvP (arenas, duels, rankings) - World dynamics (time, weather, ecosystems)

Source code in packages/maid-classic-rpg/src/maid_classic_rpg/pack.py
def __init__(self) -> None:
    """Initialize the content pack."""
    from typing import Any

    self._recipe_manager: Any = None  # RecipeManager stored to prevent GC

manifest property

Get pack manifest.

get_dependencies

get_dependencies() -> list[str]

Get pack dependencies.

Source code in packages/maid-classic-rpg/src/maid_classic_rpg/pack.py
def get_dependencies(self) -> list[str]:
    """Get pack dependencies."""
    return ["stdlib"]

get_events

get_events() -> list[type[Event]]

Get events defined by this pack.

Source code in packages/maid-classic-rpg/src/maid_classic_rpg/pack.py
def get_events(self) -> list[type[Event]]:
    """Get events defined by this pack."""
    from maid_classic_rpg.events import get_classic_rpg_events

    return get_classic_rpg_events()

get_systems

get_systems(world: World) -> list[System]

Get systems provided by this pack.

Source code in packages/maid-classic-rpg/src/maid_classic_rpg/pack.py
def get_systems(self, world: World) -> list[System]:
    """Get systems provided by this pack."""
    from maid_classic_rpg.systems import get_classic_rpg_systems

    return get_classic_rpg_systems(world)

on_load async

on_load(engine: GameEngine) -> None

Called when pack is loaded.

Initializes game data including: - Quest definitions from data files - Crafting recipes - Default world setup if needed

Source code in packages/maid-classic-rpg/src/maid_classic_rpg/pack.py
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
async def on_load(self, engine: GameEngine) -> None:
    """Called when pack is loaded.

    Initializes game data including:
    - Quest definitions from data files
    - Crafting recipes
    - Default world setup if needed
    """
    import logging
    from importlib.resources import files

    logger = logging.getLogger(__name__)
    logger.info("Loading Classic RPG content pack...")

    # Initialize Managers
    from pathlib import Path

    from maid_classic_rpg.auth.manager import AccountManager
    from maid_classic_rpg.managers.character_manager import CharacterManager
    from maid_classic_rpg.managers.item_manager import ItemManager
    from maid_classic_rpg.world.room_manager import RoomManager

    # Attach managers to engine/world for global access
    # Use data directory from settings or default
    data_dir = Path.cwd() / "data"
    if (
        hasattr(engine, "settings")
        and engine.settings
        and hasattr(engine.settings, "storage")
        and hasattr(engine.settings.storage, "data_dir")
    ):
        data_dir = Path(engine.settings.storage.data_dir)
    account_dir = data_dir / "accounts"
    char_dir = data_dir / "characters"
    items_dir = data_dir / "items"
    rooms_dir = data_dir / "rooms"

    self.account_manager = AccountManager(data_dir=account_dir)
    self.character_manager = CharacterManager(data_dir=char_dir)
    self.item_manager = ItemManager(data_dir=items_dir)
    self.room_manager = RoomManager(data_dir=rooms_dir)

    # Attach managers to engine/world for access by systems/commands
    # Note: engine.characters, engine.items, engine.rooms are properties
    # that delegate to world, so only set on world
    engine.accounts = self.account_manager  # type: ignore[attr-defined]
    engine._character_manager = self.character_manager  # type: ignore[attr-defined]
    if engine.world:
        engine.world.characters = self.character_manager  # type: ignore[attr-defined]
        engine.world.items = self.item_manager  # type: ignore[attr-defined]
        engine.world.rooms = self.room_manager  # type: ignore[attr-defined]

    # Load manager data
    await self.item_manager.load()
    await self.room_manager.load()

    # Load starter area from YAML files (expanded Millbrook)
    from maid_classic_rpg.world.load_yaml_starter_area import (
        load_expanded_starter_area,
    )

    if engine.world:
        await load_expanded_starter_area(
            self.room_manager, self.item_manager, engine.world,
        )

    # Sync all loaded items to ECS entities
    if engine.world:
        from maid_classic_rpg.managers.ecs_item_bridge import (
            sync_all_items_to_ecs,
        )

        synced = await sync_all_items_to_ecs(engine.world, self.item_manager)
        logger.info("Synced %d items to ECS entities", synced)

    # Fall back to legacy starter area if YAML loading produced no rooms
    from maid_classic_rpg.world.setup_starter_area import (
        STARTING_ROOM_ID,
        setup_starter_area,
        setup_starter_npcs,
    )

    if not await self.room_manager.get_by_id(STARTING_ROOM_ID):
        await setup_starter_area(self.room_manager)
        logger.info("Created legacy starter area (Town Square and surroundings)")

    # Create starter NPCs (Guard Captain Aldric, etc.) if not already present
    if engine.world:
        setup_starter_npcs(engine.world)

    # Sync rooms from room_manager into the World room index
    if engine.world:
        all_rooms = await self.room_manager.list_rooms()
        for room in all_rooms:
            engine.world.register_room(room.id, room)
        logger.debug("Registered %d rooms with World", len(all_rooms))

    # Load NPC autonomy archetypes and bark templates
    from maid_classic_rpg.models.npc.archetypes import ArchetypeRegistry
    from maid_classic_rpg.systems.npc.barks import load_barks_from_yaml

    npc_data_dir = files("maid_classic_rpg") / "data" / "npcs"
    archetypes_path = Path(str(npc_data_dir / "archetypes.yaml"))
    barks_path = Path(str(npc_data_dir / "barks.yaml"))

    archetype_registry = ArchetypeRegistry()
    if archetypes_path.exists():
        loaded = archetype_registry.load_from_yaml(archetypes_path)
        if engine.world:
            engine.world.set_data("npc_archetype_registry", archetype_registry)
            engine.world.set_data(
                "npc_archetype_max_need_delta",
                {
                    archetype.archetype_id: archetype.max_need_delta_per_tick
                    for archetype in loaded
                },
            )
            logger.debug("Loaded %d NPC archetypes", len(loaded))

    if barks_path.exists() and engine.world:
        bark_templates = load_barks_from_yaml(barks_path)
        engine.world.set_data("npc_bark_templates", bark_templates)
        logger.debug("Loaded bark templates for %d archetypes", len(bark_templates))

    # Wire cached memory/relationship adapters into NPC systems.
    # Systems are created with null providers; replace them now if
    # the real services are available on the world.
    if engine.world:
        from maid_classic_rpg.systems.npc.needs import NeedDecaySystem
        from maid_classic_rpg.systems.npc.social import SocialFabricSystem
        from maid_classic_rpg.systems.npc.stubs import (
            CachedMemoryAdapter,
            CachedRelationshipAdapter,
        )

        memory_adapter = CachedMemoryAdapter(engine.world)
        relationship_adapter = CachedRelationshipAdapter(engine.world)

        systems_mgr_npc = getattr(engine.world, "systems", None)
        if systems_mgr_npc:
            for system in getattr(systems_mgr_npc, "systems", []):
                if isinstance(system, NeedDecaySystem):
                    system._memory_provider = memory_adapter  # noqa: SLF001
                elif isinstance(system, SocialFabricSystem):
                    system._relationship_provider = relationship_adapter  # noqa: SLF001

        engine.world.set_data("_memory_adapter", memory_adapter)
        engine.world.set_data("_relationship_adapter", relationship_adapter)
        logger.debug("Wired cached memory/relationship adapters into NPC systems")

    # Wire ItemManager into RewardDistributor and CraftingSystem.
    # These are created during get_classic_rpg_systems() before the
    # ItemManager exists, so we connect them now.
    if engine.world:
        # Wire into RewardDistributor
        reward_distributor = getattr(engine.world, "reward_distributor", None)
        if reward_distributor and hasattr(reward_distributor, "set_item_manager"):
            reward_distributor.set_item_manager(self.item_manager)
            logger.debug("Wired ItemManager into RewardDistributor")

        # Wire into CraftingSystem
        from maid_classic_rpg.systems.crafting.crafting import CraftingSystem

        systems_manager = getattr(engine.world, "systems", None)
        if systems_manager:
            systems_list = getattr(systems_manager, "systems", [])
            from maid_classic_rpg.systems.quests.generation.models import (
                QuestSeedTypes,
            )
            from maid_classic_rpg.systems.quests.generation.system import (
                QuestGenerationSystem,
            )

            for system in systems_list:
                if isinstance(system, CraftingSystem):
                    system.set_item_manager(self.item_manager)
                    logger.debug("Wired ItemManager into CraftingSystem")
                if isinstance(system, QuestGenerationSystem):
                    for seed_type in QuestSeedTypes.EXTENSION_TYPES:
                        system._seed_registry.register(  # noqa: SLF001
                            seed_type,
                            f"Registered extension seed type: {seed_type}",
                        )
                    from maid_classic_rpg.systems.quests.generation.archetypes import (
                        DEFAULT_ARCHETYPES,
                    )

                    system.register_archetypes(DEFAULT_ARCHETYPES)
                    logger.debug("Registered %d quest archetypes", len(DEFAULT_ARCHETYPES))

    # Register Connection Handler
    if engine.server:
        from maid_classic_rpg.auth.character_handler import CharacterHandler
        from maid_classic_rpg.auth.login_handler import LoginHandler
        from maid_classic_rpg.entities.character_entity import create_character_entity
        from maid_engine.net.server import MAIDServer
        from maid_engine.net.session import Session, SessionState

        async def classic_rpg_connection_handler(session: Session, server: MAIDServer) -> None:
            """Handle new connections for Classic RPG."""
            character = None
            try:
                # 1. Login
                login_handler = LoginHandler(session, self.account_manager)
                account = await login_handler.run()

                if not account:
                    await session.close()
                    return

                session.metadata["account_id"] = account.id

                # 2. Character Selection
                from maid_classic_rpg.world.setup_starter_area import (
                    STARTING_ROOM_ID,
                )

                char_handler = CharacterHandler(
                    session, account, self.character_manager,
                    starting_room_id=STARTING_ROOM_ID,
                )
                character = await char_handler.run()

                if not character:
                    await session.close()
                    return

                # 3. Enter World
                # Create ECS entity for the character
                # Check if entity already exists (reconnect), otherwise create
                entity = engine.world.get_entity(character.id)
                if not entity:
                    # Create new entity from character model
                    entity = create_character_entity(engine.world.entities, character)

                    # Recalculate inventory weight on load
                    if character.inventory and self.item_manager:
                        weight = await self.item_manager.recalculate_inventory_weight(character.inventory)
                        character.current_weight = weight
                        # Also update the inventory component
                        from maid_stdlib.components import InventoryComponent
                        inv_comp = entity.try_get(InventoryComponent)
                        if inv_comp:
                            inv_comp.current_weight = weight

                # Link session to player entity
                session.player_id = character.id
                session.metadata["character_name"] = character.name
                session.metadata["character"] = character
                server.sessions.link_player(session.id, character.id)
                session.state = SessionState.PLAYING

                # Place entity in room (via World room index)
                if character.room_id:
                    engine.world.place_entity_in_room(character.id, character.room_id)

                # Notify login
                await session.send_line("\r\n[Entering World...]")

                # Initial look
                from maid_engine.commands.registry import CommandContext

                ctx = CommandContext(
                    session=session,
                    player_id=character.id,
                    command="look",
                    args=[],
                    raw_input="look",
                    world=engine.world,
                    document_store=engine.document_store,
                    metadata={"engine": engine, "server": server},
                )
                await engine.command_registry.execute(ctx)
                await session.send_prompt()

                # 4. Main Loop
                while session.state not in (SessionState.DISCONNECTING, SessionState.DISCONNECTED):
                    try:
                        # Wait for input
                        line = await session.receive()
                        line = line.strip()

                        if not line:
                            await session.send_prompt()
                            continue

                        if line.lower() in ("quit", "exit"):
                            await session.send_line("Goodbye!")
                            break

                        # Execute command
                        parts = line.split()
                        cmd = parts[0].lower()
                        args = parts[1:]

                        ctx = CommandContext(
                            session=session,
                            player_id=character.id,
                            command=cmd,
                            args=args,
                            raw_input=line,
                            world=engine.world,
                            document_store=engine.document_store,
                            target=args[0] if args else None,
                            metadata={"engine": engine, "server": server},
                        )

                        try:
                            handled = await engine.command_registry.execute(ctx)
                            if not handled:
                                await session.send_line("I don't understand that.")
                        except PermissionError as pe:
                            await session.send_line(str(pe))
                        except Exception as cmd_err:
                            logger.exception("Error executing command %r", cmd)
                            await session.send_line(f"An error occurred: {cmd_err}")

                        await session.send_prompt()

                    except (ConnectionError, asyncio.CancelledError):
                        break

            except Exception as e:
                logger.error(f"Error in connection handler: {e}")
                try:
                    await session.send_line(f"An error occurred: {e}")
                    await session.close()
                except Exception:
                    pass
            finally:
                # Always clean up: remove character entity from room on disconnect
                if character and character.id:
                    # Remove ECS entity entirely to prevent PositionSyncSystem
                    # from re-adding it to the room index on the next tick
                    engine.world.destroy_entity(character.id)
                    # Clear character's room_id so they don't appear
                    # in CharacterManager.get_in_room() queries
                    character.room_id = None
                    cached = self.character_manager._cache.get(character.id)
                    if cached and cached is not character:
                        cached.room_id = None

        engine.server.set_connection_handler(classic_rpg_connection_handler)

    # Load quest data
    # Quest data is loaded by QuestManager when systems initialize
    logger.debug("Quest system initialized")

    # Load crafting recipes (non-critical - game can run without recipes)
    try:
        from maid_classic_rpg.systems.crafting.recipes import RecipeManager

        # Get the data directory path using importlib.resources
        recipe_data_dir = files("maid_classic_rpg") / "data" / "crafting" / "recipes"
        if recipe_data_dir.is_dir():
            # Convert Traversable to Path for RecipeManager
            recipe_path = Path(str(recipe_data_dir))
            self._recipe_manager = RecipeManager(data_dir=recipe_path)
            count = await self._recipe_manager.load()
            logger.debug(f"Loaded {count} crafting recipes")
        else:
            logger.debug("No crafting recipes directory found, skipping")
    except ImportError:
        logger.debug("RecipeManager not available, crafting disabled")
    except Exception:
        # Log full traceback for unexpected errors, but don't crash
        logger.exception("Failed to load crafting recipes")

    # Note: Storage injection is handled automatically by the engine
    # for systems implementing StorageAware protocol

    # QuestManager is not a System, so we need to manually inject storage
    if engine.world and hasattr(engine.world, "quest_manager"):
        quest_manager = engine.world.quest_manager
        if hasattr(quest_manager, "set_storage"):
            quest_manager.set_storage(engine.document_store)
            logger.debug("Injected storage into QuestManager")

    logger.info("Classic RPG content pack loaded")

    # Register social commands (requires system instances from the world)
    if engine.world:
        systems_mgr = getattr(engine.world, "systems", None)
        if systems_mgr:
            from maid_classic_rpg.systems.social.groups import GroupSystem
            from maid_stdlib.systems.social.channels import ChannelSystem
            from maid_stdlib.systems.social.friends import FriendsSystem
            from maid_stdlib.systems.social.mail import MailSystem

            all_systems = getattr(systems_mgr, "systems", [])
            channels = None
            groups = None
            friends = None
            mail = None
            for system in all_systems:
                if isinstance(system, ChannelSystem):
                    channels = system
                elif isinstance(system, GroupSystem):
                    groups = system
                elif isinstance(system, FriendsSystem):
                    friends = system
                elif isinstance(system, MailSystem):
                    mail = system

            if channels and groups and friends and mail:
                from maid_classic_rpg.systems.social.commands import (
                    register_social_commands,
                )

                register_social_commands(channels, groups, friends, mail)
                logger.debug("Registered social commands")
            else:
                logger.warning("Could not find all social systems for command registration")

    # Subscribe audit logger to player connect/disconnect events
    if engine.world and engine.world.events:
        from maid_engine.core.events import (
            PlayerConnectedEvent,
            PlayerDisconnectedEvent,
        )
        from maid_engine.logging.audit import (
            AuditEntry,
            AuditEventType,
            get_audit_logger,
        )

        audit = get_audit_logger()

        async def _on_player_connected(event: PlayerConnectedEvent) -> None:
            await audit.log(AuditEntry(
                event_type=AuditEventType.ACCOUNT_PROMOTE,  # closest available type
                action="player_login",
                actor_id=event.player_id,
                details=f"session={event.session_id}",
            ))

        async def _on_player_disconnected(event: PlayerDisconnectedEvent) -> None:
            await audit.log(AuditEntry(
                event_type=AuditEventType.ACCOUNT_DEMOTE,  # closest available type
                action="player_logout",
                actor_id=event.player_id,
                details=f"session={event.session_id}",
            ))

        engine.world.events.subscribe(
            PlayerConnectedEvent, _on_player_connected,
        )
        engine.world.events.subscribe(
            PlayerDisconnectedEvent, _on_player_disconnected,
        )
        logger.debug("Subscribed audit logger to player connect/disconnect events")

on_unload async

on_unload(engine: GameEngine) -> None

Called when pack is unloaded.

Performs cleanup: - Saves any pending data - Releases resources

Source code in packages/maid-classic-rpg/src/maid_classic_rpg/pack.py
async def on_unload(self, engine: GameEngine) -> None:
    """Called when pack is unloaded.

    Performs cleanup:
    - Saves any pending data
    - Releases resources
    """
    import logging

    logger = logging.getLogger(__name__)
    logger.info("Unloading Classic RPG content pack...")

    # Persist all data before unloading
    systems_manager = engine.world.systems
    systems_list = getattr(systems_manager, "systems", [])
    for system in systems_list:
        if hasattr(system, "shutdown"):
            try:
                await system.shutdown()
            except Exception as e:
                logger.warning(f"Error during system shutdown: {e}")

    # Shutdown QuestManager (not a System, but has pending saves to await)
    if engine.world and hasattr(engine.world, "quest_manager"):
        quest_manager = engine.world.quest_manager
        if hasattr(quest_manager, "shutdown"):
            try:
                await quest_manager.shutdown()
            except Exception as e:
                logger.warning(f"Error during QuestManager shutdown: {e}")

    logger.info("Classic RPG content pack unloaded")

register_api_routes

register_api_routes(router: APIRouter) -> None

Register API routes. Classic RPG has no custom admin routes.

Source code in packages/maid-classic-rpg/src/maid_classic_rpg/pack.py
def register_api_routes(self, router: APIRouter) -> None:
    """Register API routes. Classic RPG has no custom admin routes."""
    pass

register_commands

register_commands(registry: AnyCommandRegistry) -> None

Register commands provided by this pack.

Source code in packages/maid-classic-rpg/src/maid_classic_rpg/pack.py
def register_commands(self, registry: AnyCommandRegistry) -> None:
    """Register commands provided by this pack."""
    from maid_classic_rpg.commands import register_classic_rpg_commands

    register_classic_rpg_commands(registry, pack_name="classic-rpg")

    # Register hooks for command execution
    # These hooks are defined in maid-engine but should be configured and
    # registered by content packs that implement gameplay
    from maid_engine.commands import LayeredCommandRegistry

    if isinstance(registry, LayeredCommandRegistry):
        # Block most commands while the player is a ghost
        from maid_classic_rpg.commands.handlers.ghost import (
            ghost_restriction_hook,
        )
        from maid_engine.commands import (
            CombatStateHook,
            CooldownHook,
            CooldownRecordHook,
            HookPriority,
        )

        registry.register_pre_hook(
            name="ghost_restriction",
            handler=ghost_restriction_hook,
            priority=HookPriority.FIRST,
        )

        # Block commands that shouldn't be used during combat:
        # - quit/logout: Can't leave the game while fighting
        # - crafting category: Can't craft items while fighting
        # - economy category: Can't shop/trade while fighting
        # - builder category: Can't build while fighting
        combat_hook = CombatStateHook(
            blocked_commands=["quit", "logout"],
            blocked_categories=["crafting", "economy", "builder"],
        )
        registry.register_pre_hook(
            name="combat_state_block",
            handler=combat_hook,
            priority=HookPriority.HIGH,
        )

        # Register CooldownHook to enforce command cooldowns
        # Commands can specify cooldown times in their metadata (in seconds)
        # e.g., @command(name="teleport", metadata={"cooldown": 60})
        cooldown_hook = CooldownHook()
        registry.register_pre_hook(
            name="cooldown_check",
            handler=cooldown_hook,
            priority=HookPriority.HIGH,
        )

        # Register CooldownRecordHook to record successful command usage
        # This must share the same CooldownHook instance to track cooldowns
        cooldown_record_hook = CooldownRecordHook(cooldown_hook)
        registry.register_post_hook(
            name="cooldown_record",
            handler=cooldown_record_hook,
            priority=HookPriority.NORMAL,
        )

register_component_types

register_component_types(
    registry: ComponentRegistry,
) -> None

Register component types for persistence serialization.

Source code in packages/maid-classic-rpg/src/maid_classic_rpg/pack.py
def register_component_types(self, registry: ComponentRegistry) -> None:
    """Register component types for persistence serialization."""
    from maid_classic_rpg import components as classic_components
    from maid_engine.core.ecs.component import Component

    for name in getattr(classic_components, "__all__", []):
        candidate = getattr(classic_components, name, None)
        if (
            isinstance(candidate, type)
            and issubclass(candidate, Component)
            and candidate is not Component
        ):
            registry.register(
                candidate,
                pack_name="classic-rpg",
                allow_overwrite=True,
            )

register_document_schemas

register_document_schemas(store: DocumentStore) -> None

Register document schemas used by this pack.

Source code in packages/maid-classic-rpg/src/maid_classic_rpg/pack.py
def register_document_schemas(self, store: DocumentStore) -> None:
    """Register document schemas used by this pack."""
    from maid_classic_rpg.models import register_classic_rpg_schemas
    from maid_classic_rpg.systems.quests.generation.history import (
        register_generation_schemas,
    )

    register_classic_rpg_schemas(store)
    register_generation_schemas(store)