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

pip install maid-classic-rpg

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
social	Guilds, factions, achievements
pvp	Arenas, duels, rankings
world	Time, weather, ecosystems
npc	NPC behavior and AI dialogue
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

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

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.

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

update async

update(delta: float) -> None

Update tick - process damage over time effects.

shutdown async

shutdown() -> None

Clean up body component cache to prevent memory leak.

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.

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.

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	Item manager for creating output	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

Manages resource gathering in the world.

startup async

startup() -> None

Initialize gathering system.

shutdown async

shutdown() -> None

Shutdown gathering system.

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 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

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.

set_storage

set_storage(store: DocumentStore) -> None

Set the document store for persistence.

startup async

startup() -> None

Initialize bank system.

shutdown async

shutdown() -> None

Shutdown bank system.

register_bank

register_bank(bank: BankNPC) -> None

Register a bank NPC.

TODO: Validate room_id exists in world before registration.

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)

TODO: Emit BankTransferEvent for audit trail consistency.

ShopManager

ShopManager()

Manages shop data and operations.

Handles shop registration, inventory, and pricing calculations.

register_shop

register_shop(shop: Shop) -> None

Register a shop.

TODO: Validate room_id exists in world before registration. TODO: Warn or prevent overwriting existing shop in same room.

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.

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.

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,
)

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)

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,
)

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

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 updates needed.

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).

ChannelSystem

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

Bases: System

Manages chat channels and messaging.

Handles: - Global channels (ooc, newbie, auction) - Custom player-created channels - Channel moderation (mute, ban) - Message history - Cross-node messaging via Redis pub/sub

set_sessions

set_sessions(sessions: SessionManager) -> None

Set the session manager after initialization.

startup async

startup() -> None

Subscribe to connection events.

shutdown async

shutdown() -> None

Unsubscribe from events.

update async

update(delta: float) -> None

No per-tick updates needed for channels.

get_channel

get_channel(name: str) -> Channel | None

Get channel by name.

create_channel

create_channel(
    name: str,
    owner_id: UUID,
    channel_type: ChannelType = CUSTOM,
    is_public: bool = False,
    password: str | None = None,
) -> tuple[Channel | None, str]

Create a new channel.

Returns:

Type	Description
tuple[Channel | None, str]	Tuple of (channel, error_message)

delete_channel

delete_channel(
    name: str, requester_id: UUID
) -> tuple[bool, str]

Delete a channel. Only owner or admin can delete.

join_channel

join_channel(
    player_id: UUID,
    channel_name: str,
    password: str | None = None,
) -> tuple[bool, str]

Join a channel.

leave_channel

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

Leave a channel.

send_message async

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

Send a message to a channel.

get_history

get_history(
    channel_name: str, limit: int = 20
) -> list[dict[str, Any]]

Get recent channel history.

mute_player

mute_player(
    channel_name: str,
    player_id: UUID,
    duration_minutes: int,
    moderator_id: UUID,
) -> tuple[bool, str]

Mute a player in a channel.

unmute_player

unmute_player(
    channel_name: str, player_id: UUID, moderator_id: UUID
) -> tuple[bool, str]

Unmute a player in a channel.

ban_player

ban_player(
    channel_name: str, player_id: UUID, moderator_id: UUID
) -> tuple[bool, str]

Ban a player from a channel.

unban_player

unban_player(
    channel_name: str, player_id: UUID, moderator_id: UUID
) -> tuple[bool, str]

Unban a player from a channel.

hide_channel

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

Hide a channel from a player (mute locally).

unhide_channel

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

Unhide a channel for a player.

get_player_channels

get_player_channels(player_id: UUID) -> list[str]

Get list of channels player is in.

get_membership

get_membership(player_id: UUID) -> ChannelMembership | None

Get player's channel membership.

set_last_tell

set_last_tell(
    player_id: UUID, from_player_id: UUID
) -> None

Set the last player who sent a tell to this player.

get_last_tell

get_last_tell(player_id: UUID) -> UUID | None

Get the last player who sent a tell to this player.

list_channels

list_channels(
    include_private: bool = False,
) -> list[Channel]

Get a list of all channels.

deliver_external_message async

deliver_external_message(
    channel_name: str, sender_name: str, message: str
) -> bool

Deliver a message from an external bridge to the game.

This method implements the ExternalMessageReceiver protocol, allowing the engine to deliver messages from external services (Discord, IRC, etc.) to game channels without accessing private attributes.

Parameters:

Name	Type	Description	Default
channel_name	str	The game channel name to send to (e.g., "ooc")	required
sender_name	str	The formatted sender name (e.g., "[Discord] User")	required
message	str	The message content	required

Returns:

Type	Description
bool	True if the message was delivered, False otherwise

ChannelMessageEvent dataclass

ChannelMessageEvent(
    channel_name: str,
    sender_id: UUID,
    sender_name: str,
    message: str,
)

Bases: Event

Emitted when a message is sent to a channel.

FriendAddedEvent dataclass

FriendAddedEvent(
    player_id: UUID, friend_id: UUID, friend_name: str
)

Bases: Event

Emitted when a player adds someone as a friend.

FriendOfflineEvent dataclass

FriendOfflineEvent(
    player_id: UUID,
    player_name: str,
    friend_ids: list[UUID] = list(),
)

Bases: Event

Emitted when a friend goes offline.

FriendOnlineEvent dataclass

FriendOnlineEvent(
    player_id: UUID,
    player_name: str,
    friend_ids: list[UUID] = list(),
)

Bases: Event

Emitted when a friend comes online.

FriendRemovedEvent dataclass

FriendRemovedEvent(player_id: UUID, friend_id: UUID)

Bases: Event

Emitted when a player removes a friend.

GroupDisbandEvent dataclass

GroupDisbandEvent(group_id: UUID, leader_id: UUID)

Bases: Event

Emitted when a group is disbanded.

GroupInviteEvent dataclass

GroupInviteEvent(
    group_id: UUID,
    inviter_id: UUID,
    inviter_name: str,
    invitee_id: UUID,
)

Bases: Event

Emitted when a player is invited to a group.

GroupJoinEvent dataclass

GroupJoinEvent(
    group_id: UUID, player_id: UUID, player_name: str
)

Bases: Event

Emitted when a player joins a group.

GroupLeaveEvent dataclass

GroupLeaveEvent(
    group_id: UUID,
    player_id: UUID,
    player_name: str,
    was_kicked: bool = False,
)

Bases: Event

Emitted when a player leaves a group.

MailReceivedEvent dataclass

MailReceivedEvent(
    mail_id: UUID,
    sender_id: UUID,
    sender_name: str,
    recipient_id: UUID,
    subject: str,
    has_attachment: bool = False,
)

Bases: Event

Emitted when a player receives mail.

PlayerIgnoredEvent dataclass

PlayerIgnoredEvent(
    player_id: UUID, target_id: UUID, target_name: str
)

Bases: Event

Emitted when a player ignores another player.

PlayerUnignoredEvent dataclass

PlayerUnignoredEvent(player_id: UUID, target_id: UUID)

Bases: Event

Emitted when a player unignores another player.

TellEvent dataclass

TellEvent(
    sender_id: UUID,
    sender_name: str,
    recipient_id: UUID,
    message: str,
)

Bases: Event

Emitted when a player sends a tell.

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

No per-tick updates needed.

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).

FriendsSystem

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

Bases: System

Manages friend and ignore lists.

Features: - Add/remove friends - Add/remove ignored players - Online notifications for friends - Privacy settings

set_sessions

set_sessions(sessions: SessionManager) -> None

Set the session manager after initialization.

startup async

startup() -> None

Subscribe to connection events.

shutdown async

shutdown() -> None

Unsubscribe from events.

update async

update(delta: float) -> None

No per-tick updates needed.

get_social_list

get_social_list(player_id: UUID) -> SocialList

Get or create social list for player.

add_friend

add_friend(
    player_id: UUID, friend_id: UUID, friend_name: str
) -> tuple[bool, str]

Add a player to friends list.

remove_friend

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

Remove a player from friends list.

get_friends

get_friends(player_id: UUID) -> dict[UUID, str]

Get player's friends list.

is_friend

is_friend(player_id: UUID, other_id: UUID) -> bool

Check if other is on player's friends list.

is_mutual_friend

is_mutual_friend(
    player1_id: UUID, player2_id: UUID
) -> bool

Check if two players are mutual friends.

add_ignore

add_ignore(
    player_id: UUID, target_id: UUID, target_name: str
) -> tuple[bool, str]

Add a player to ignore list.

remove_ignore

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

Remove a player from ignore list.

get_ignored

get_ignored(player_id: UUID) -> dict[UUID, str]

Get player's ignore list.

is_ignored

is_ignored(player_id: UUID, other_id: UUID) -> bool

Check if other is on player's ignore list.

can_message

can_message(
    sender_id: UUID, recipient_id: UUID
) -> tuple[bool, str]

Check if sender can message recipient.

set_allow_tells

set_allow_tells(player_id: UUID, allow: bool) -> None

Set whether player accepts tells from non-friends.

set_allow_finger

set_allow_finger(player_id: UUID, allow: bool) -> None

Set whether player allows finger info.

set_show_online

set_show_online(player_id: UUID, show: bool) -> None

Set whether player shows online status.

get_privacy_settings

get_privacy_settings(player_id: UUID) -> dict[str, bool]

Get player's privacy settings.

get_friends_who_added

get_friends_who_added(player_id: UUID) -> list[UUID]

Get list of players who have this player as a friend.

is_online

is_online(player_id: UUID) -> bool

Check if a player is online.

get_online_friends async

get_online_friends(
    player_id: UUID,
) -> list[tuple[UUID, str]]

Get list of online friends.

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

No per-tick updates needed.

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 any dirty guilds.

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,
) -> None

Deposit an item into the guild bank.

withdraw_item async

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

Withdraw an item from the guild bank.

deposit_gold async

deposit_gold(
    guild_id: UUID,
    character_id: UUID,
    character_name: str,
    amount: int,
) -> None

Deposit gold into the guild bank.

withdraw_gold async

withdraw_gold(
    guild_id: UUID,
    character_id: UUID,
    character_name: str,
    amount: int,
) -> 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.

MailSystem

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

Bases: System, StorageAware

Manages the in-game mail system.

Features: - Send mail to offline players - Attachments (gold, items) - Mail expiration - Reply chains

set_storage

set_storage(store: DocumentStore) -> None

Set the document store for persistence.

set_sessions

set_sessions(sessions: SessionManager) -> None

Set the session manager after initialization.

startup async

startup() -> None

Subscribe to connection events and load data.

shutdown async

shutdown() -> None

Unsubscribe from events and save data.

update async

update(delta: float) -> None

Process mail expiration periodically.

get_mailbox

get_mailbox(player_id: UUID) -> list[MailMessage]

Get player's mailbox.

get_inbox

get_inbox(player_id: UUID) -> list[MailMessage]

Get non-deleted messages for player.

get_unread_count

get_unread_count(player_id: UUID) -> int

Get count of unread mail.

send_mail async

send_mail(
    sender_id: UUID,
    sender_name: str,
    recipient_id: UUID,
    recipient_name: str,
    subject: str,
    body: str,
    attachment: MailAttachment | None = None,
    reply_to_id: UUID | None = None,
) -> tuple[MailMessage | None, str]

Send a mail message.

get_message

get_message(
    player_id: UUID, message_id: UUID
) -> MailMessage | None

Get a specific message.

get_message_by_index

get_message_by_index(
    player_id: UUID, index: int
) -> MailMessage | None

Get a message by inbox index (0-based).

read_message

read_message(
    player_id: UUID, message_id: UUID
) -> tuple[MailMessage | None, str]

Read a message (marks as read).

claim_attachment

claim_attachment(
    player_id: UUID, message_id: UUID
) -> tuple[MailAttachment | None, str]

Claim attachment from a message.

delete_message

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

Soft delete a message.

purge_deleted

purge_deleted(player_id: UUID) -> int

Permanently remove deleted messages.

cleanup_expired

cleanup_expired() -> int

Remove expired messages (run periodically).

reply_to_mail async

reply_to_mail(
    player_id: UUID,
    player_name: str,
    original_id: UUID,
    body: str,
) -> tuple[MailMessage | None, str]

Reply to a message.

send_system_mail async

send_system_mail(
    recipient_id: UUID,
    recipient_name: str,
    subject: str,
    body: str,
    attachment: MailAttachment | None = None,
) -> tuple[MailMessage | None, str]

Send a system-generated mail message.

get_outbox

get_outbox(player_id: UUID) -> list[UUID]

Get list of message IDs sent by player.

has_unread_mail

has_unread_mail(player_id: UUID) -> bool

Check if player has unread mail.

get_inbox_summary

get_inbox_summary(player_id: UUID) -> dict[str, int]

Get summary of inbox.

Channel

Bases: MAIDBaseModel

A chat channel definition.

can_speak

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

Check if player can speak in channel.

add_message

add_message(
    sender_id: UUID, sender_name: str, message: str
) -> None

Add message to history.

ChannelMembership

Bases: MAIDBaseModel

Player's channel membership and preferences.

ChannelPermission

Bases: StrEnum

Permissions for channel actions.

ChannelType

Bases: StrEnum

Types of chat channels.

Group

Bases: MAIDBaseModel

A player group/party.

member_ids property

member_ids: set[UUID]

Get all member IDs including leader.

get_member

get_member(player_id: UUID) -> GroupMember | None

Get member by ID.

is_leader

is_leader(player_id: UUID) -> bool

Check if player is leader.

is_member

is_member(player_id: UUID) -> bool

Check if player is in group.

get_next_looter

get_next_looter() -> UUID

Get next player for round-robin loot.

GroupMember

Bases: MAIDBaseModel

A member of a group.

GroupRole

Bases: StrEnum

Roles within a group.

LootMode

Bases: StrEnum

Group loot distribution modes.

MailAttachment

Bases: MAIDBaseModel

Attachment to a mail message.

MailMessage

Bases: MAIDBaseModel

A mail message.

has_attachment property

has_attachment: bool

Check if message has a non-empty attachment.

is_unread property

is_unread: bool

Check if message is unread.

create_with_expiry classmethod

create_with_expiry(
    sender_id: UUID,
    sender_name: str,
    recipient_id: UUID,
    recipient_name: str,
    subject: str,
    body: str,
    expiry_days: int = 30,
    attachment: MailAttachment | None = None,
    reply_to_id: UUID | None = None,
) -> MailMessage

Create a mail message with automatic expiry.

MailStatus

Bases: StrEnum

Mail message status.

SocialList

Bases: MAIDBaseModel

Player's friends and ignore lists.

is_friend

is_friend(player_id: UUID) -> bool

Check if player is a friend.

is_ignored

is_ignored(player_id: UUID) -> bool

Check if player is ignored.

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.

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, and dialogue systems.

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

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.

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.

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

manifest: ContentPackManifest

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.

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

manifest: ContentPackManifest

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

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()

    # 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."""
            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
                char_handler = CharacterHandler(session, account, self.character_manager)
                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
                server.sessions.link_player(session.id, character.id)
                session.state = SessionState.PLAYING

                # 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,
                )
                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,
                        )

                        handled = await engine.command_registry.execute(ctx)
                        if not handled:
                            await session.send_line("I don't understand that.")

                        await session.send_prompt()

                    except (ConnectionError, asyncio.CancelledError):
                        break

            except Exception as e:
                logger.error(f"Error in connection handler: {e}")
                await session.send_line(f"An error occurred: {e}")
                await session.close()

        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")

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_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):
        from maid_engine.commands import (
            CombatStateHook,
            CooldownHook,
            CooldownRecordHook,
            HookPriority,
        )

        # 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_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

    register_classic_rpg_schemas(store)