maid-engine

The maid-engine package provides the core infrastructure for building MUD games with MAID. It is the foundation layer that all other MAID packages build upon.

Installation

pip install maid-engine

# With AI providers
pip install maid-engine[openai]
pip install maid-engine[anthropic]
pip install maid-engine[ollama]

Module Overview

Core (maid_engine.core)

The core module contains the fundamental building blocks:

• engine - Main GameEngine class - tick loop, content pack loading
• world - World class - state management, entity tracking
• events - EventBus and core event types
• ecs - Entity Component System implementation

Commands (maid_engine.commands)

The command system provides layered, priority-based command resolution:

• registry - LayeredCommandRegistry for command management
• decorators - @command decorator and helpers
• arguments - Argument parsing utilities
• hooks - Pre/post command hooks
• locks - Permission lock expressions

Plugins (maid_engine.plugins)

The content pack system for extensibility:

• protocol - ContentPack protocol and BaseContentPack
• loader - ContentPackLoader for discovery and loading
• manifest - ContentPackManifest metadata

Storage (maid_engine.storage)

Document-based persistence layer:

• document_store - DocumentStore interface and implementations

Networking (maid_engine.net)

Network layer for player connections:

• server - Base NetworkServer class
• session - Session base class
• telnet - Telnet protocol support
• web - WebSocket support

AI Integration (maid_engine.ai)

LLM provider abstraction:

• providers - LLM provider interface and implementations
• conversation - Conversation history management
• prompts - Prompt building utilities
• rate_limiter - API rate limiting
• safety - Content filtering

Authentication (maid_engine.auth)

Account and session management.

Configuration (maid_engine.config)

Settings and configuration with Pydantic.

Internationalization (maid_engine.i18n)

Multi-language support with Translator and translation utilities.

---

Key Classes

GameEngine

The central game engine that manages the tick loop and content packs.

GameEngine

GameEngine(
    settings: Settings | None = None,
    document_store: DocumentStore | None = None,
)

Main game engine that runs the tick loop.

The GameEngine manages the game world, networking, and the main tick loop that drives all game systems. It is content-agnostic - all game-specific functionality comes from content packs.

Example

engine = GameEngine()

Load content packs

engine.load_content_pack(my_content_pack)

await engine.start()

Engine runs until stopped

In another context:

await engine.stop()

start async

start() -> None

Start the game engine.

stop async

stop() -> None

Stop the game engine.

load_content_pack

load_content_pack(
    pack: ContentPack, use_hot_reload: bool = False
) -> None

Load a content pack.

When the engine is stopped, content packs are loaded synchronously. When the engine is running, set use_hot_reload=True to use the hot reload manager for safe runtime loading.

Parameters:

Name	Type	Description	Default
pack	ContentPack	The content pack to load.	required
use_hot_reload	bool	If True and engine is running, use hot reload. If False and engine is running, raises RuntimeError.	False

Raises:

Type	Description
RuntimeError	If engine is running and use_hot_reload is False.
ValueError	If pack is already loaded or dependencies are missing.

World

Central state management for the game world.

World

World(settings: Settings)

Central game world state manager.

The World holds all game state including entities, rooms, and provides efficient queries for game logic.

Example

world = World(settings) await world.startup()

Create a player entity

player = world.create_entity() player.add(PositionComponent(room_id=starting_room.id)) player.add(HealthComponent(current=100, maximum=100))

Query entities in a room

for entity in world.entities_in_room(room_id): ...

get_entity

get_entity(entity_id: UUID) -> Entity | None

Get an entity by ID.

tick async

tick(delta: float) -> None

Process a single game tick.

Parameters:

Name	Type	Description	Default
delta	float	Time since last tick in seconds	required

EventBus

Event-driven communication system.

EventBus

EventBus()

Publish/subscribe event system.

The EventBus allows systems to communicate through events without direct coupling. Handlers can be async or sync.

Supports wildcard subscriptions via subscribe_all() to receive all events.

Example

bus = EventBus()

async def on_damage(event: DamageDealtEvent): print(f"Damage dealt: {event.damage}")

bus.subscribe(DamageDealtEvent, on_damage)

Subscribe to ALL events (wildcard)

def log_event(event: Event): print(f"Event: {event.event_type}") bus.subscribe_all(log_event)

await bus.emit(DamageDealtEvent( source_id=attacker.id, target_id=defender.id, damage=25, damage_type="physical" ))

subscribe

subscribe(
    event_type: type[E],
    handler: EventHandler | SyncEventHandler,
    priority: EventPriority = NORMAL,
    once: bool = False,
    source_pack: str | None = None,
) -> UUID

Subscribe to an event type.

Parameters:

Name	Type	Description	Default
event_type	type[E]	The event class to subscribe to	required
handler	EventHandler | SyncEventHandler	Async or sync function to handle events	required
priority	EventPriority	Handler priority (higher priority runs first)	NORMAL
once	bool	If True, handler is removed after first invocation	False
source_pack	str | None	Name of the content pack registering this handler (for hot reload)	None

Returns:

Type	Description
UUID	Handler ID for unsubscribing

unsubscribe

unsubscribe(handler_id: UUID) -> bool

Unsubscribe a handler by ID.

If called during event dispatch, the removal is deferred until the dispatch completes to ensure safe iteration.

Returns:

Type	Description
bool	True if handler was found and removed (or queued for removal)

emit async

emit(event: Event) -> None

Emit an event immediately, invoking all handlers.

Handlers are invoked in priority order. If an event is cancelled, remaining handlers are not called.

Both type-specific handlers and wildcard handlers (registered via subscribe_all) are invoked.

Handler removals during dispatch are deferred until dispatch completes.

Entity

Base class for all game entities.

Entity

Entity(
    entity_id: UUID | None = None,
    manager: EntityManager | None = None,
)

Represents a game entity as a collection of components.

Entities are essentially just IDs that group components together. The EntityManager handles the actual component storage and queries.

Example

entity = Entity() entity.add(PositionComponent(room_id=room.id)) entity.add(HealthComponent(current=100, maximum=100))

if entity.has(HealthComponent): health = entity.get(HealthComponent) health.damage(10)

ContentPack Protocol

Protocol defining the content pack interface.

ContentPack

Bases: Protocol

Protocol defining the interface for content packs.

Content packs provide game functionality to the MAID engine. They can include: - ECS systems that process entities - Event types for communication - Commands for player interaction - Document schemas for persistence - Setup/teardown hooks

Important

Content packs that are discovered via entry points or directory scanning MUST have a no-argument constructor (or a constructor where all parameters have default values). This is because the discovery mechanism cannot pass constructor arguments. If your pack requires initialization parameters, consider using the on_load method or environment variables instead.

Example

class MyContentPack: def init(self) -> None: # No required arguments - this is important for auto-discovery! self._initialized = False

@property
def manifest(self) -> ContentPackManifest:
    return ContentPackManifest(
        name="my-pack",
        version="1.0.0",
        display_name="My Pack",
        description="A custom content pack",
    )

def get_dependencies(self) -> list[str]:
    return ["maid-stdlib"]

def get_systems(self, world: World) -> list[System]:
    return [MySystem(world)]

def get_events(self) -> list[type[Event]]:
    return [MyEvent]

def register_commands(self, registry: CommandRegistry) -> None:
    registry.register(...)

def register_document_schemas(self, store: DocumentStore) -> None:
    store.register_schema("my_collection", MyModel)

async def on_load(self, engine: GameEngine) -> None:
    # Initialize resources - use this for configuration that
    # would otherwise require constructor arguments
    self._initialized = True

async def on_unload(self, engine: GameEngine) -> None:
    # Cleanup resources
    pass

manifest property

manifest: ContentPackManifest

Get the content pack manifest.

Returns metadata about this pack including name, version, dependencies, and capabilities.

get_dependencies

get_dependencies() -> list[str]

Get the names of content packs this pack depends on.

Dependencies must be loaded before this pack can be loaded. Returns a list of content pack names (not versions).

get_systems

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

Get ECS systems provided by this pack.

Systems are registered with the engine's SystemManager and are updated each tick.

Parameters:

Name	Type	Description	Default
world	World	The game world to create systems for	required

Returns:

Type	Description
list[System]	List of System instances to register

get_events

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

Get event types defined by this pack.

These event types can be used by the EventBus for publish/subscribe communication.

Returns:

Type	Description
list[type[Event]]	List of Event subclasses

register_commands

register_commands(registry: AnyCommandRegistry) -> None

Register commands provided by this pack.

Commands are registered with the engine's CommandRegistry and can be executed by players.

Parameters:

Name	Type	Description	Default
registry	AnyCommandRegistry	The command registry to register with	required

register_document_schemas

register_document_schemas(store: DocumentStore) -> None

Register document schemas used by this pack.

Schemas define the structure of documents stored in the document store.

Parameters:

Name	Type	Description	Default
store	DocumentStore	The document store to register schemas with	required

on_load async

on_load(engine: GameEngine) -> None

Called when the pack is loaded.

Use this hook to: - Initialize resources - Load data files - Subscribe to events - Set up background tasks

Parameters:

Name	Type	Description	Default
engine	GameEngine	The game engine	required

on_unload async

on_unload(engine: GameEngine) -> None

Called when the pack is unloaded.

Use this hook to: - Clean up resources - Save state - Unsubscribe from events - Cancel background tasks

Parameters:

Name	Type	Description	Default
engine	GameEngine	The game engine	required

LayeredCommandRegistry

Priority-based command resolution.

LayeredCommandRegistry

LayeredCommandRegistry(event_bus: 'EventBus | None' = None)

Command registry with layered resolution for content packs.

Commands can be registered by multiple content packs. When a command is executed, the highest priority registration is used.

The registry now integrates with: - Hook system: Pre/post command hooks for logging, rate limiting, etc. - Lock system: Permission checks using lock expressions

Example

registry = LayeredCommandRegistry()

Set pack priorities (higher = wins)

registry.set_pack_priority("my-game", 100) registry.set_pack_priority("stdlib", 50) registry.set_pack_priority("engine", 0)

Register commands from different packs

registry.register("look", look_handler, pack_name="stdlib") registry.register("look", custom_look_handler, pack_name="my-game")

When "look" is executed, my-game's handler is used (priority 100)

Register hooks

registry.register_pre_hook("rate_limiter", check_rate_limit) registry.register_post_hook("logger", log_command)

Register custom lock functions

registry.register_lock_function("is_owner", check_ownership)

execute async

execute(
    context: CommandContext,
    player_access_level: AccessLevel = PLAYER,
) -> bool

Execute a command using the highest-priority handler with lock checking and hook pipeline.

Execution flow: 1. Find the command definition 2. Check lock expression (if defined) 3. Check access level 4. Run pre-hooks (can cancel execution) 5. Execute command handler 6. Run post-hooks (always runs, even on exception)

Parameters:

Name	Type	Description	Default
context	CommandContext	Command context containing session, player_id, command, args, etc.	required
player_access_level	AccessLevel	The access level of the player executing	PLAYER

Returns:

Type	Description
bool	True if command was handled successfully, False if command not found
bool	or execution was cancelled

Raises:

Type	Description
PermissionError	If player lacks required access level

---

Package API

Full package API documentation:

maid_engine

MAID Engine - Core infrastructure for building MUD games.

This package provides the foundational infrastructure for building Multi-User Dungeon (MUD) games, including:

• Entity Component System (ECS) for game entity management
• Event-driven communication via EventBus
• Network layer (Telnet and WebSocket support)
• Authentication and session management
• Storage abstraction (SQL for auth, document store for game data)
• Plugin/content pack system for extensibility
• AI provider abstraction
• Internationalization (i18n) support

AccessLevel

Bases: IntEnum

Command access levels.

BaseContentPack

Base implementation of ContentPack with sensible defaults.

Extend this class to create a content pack with minimal boilerplate. Override only the methods you need.

Example

class MyPack(BaseContentPack): @property def manifest(self) -> ContentPackManifest: return ContentPackManifest( name="my-pack", version="1.0.0", display_name="My Pack", description="Custom content", )

def get_systems(self, world: World) -> list[System]:
    return [MySystem(world)]

manifest property

manifest: ContentPackManifest

Get the content pack manifest. Must be overridden.

get_dependencies

get_dependencies() -> list[str]

Get dependencies. Default: none.

Source code in packages/maid-engine/src/maid_engine/plugins/protocol.py

def get_dependencies(self) -> list[str]:
    """Get dependencies. Default: none."""
    return []

get_events

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

Get events. Default: none.

Source code in packages/maid-engine/src/maid_engine/plugins/protocol.py

def get_events(self) -> list[type[Event]]:
    """Get events. Default: none."""
    return []

get_systems

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

Get systems. Default: none.

Source code in packages/maid-engine/src/maid_engine/plugins/protocol.py

def get_systems(self, world: World) -> list[System]:  # noqa: ARG002
    """Get systems. Default: none."""
    return []

on_load async

on_load(engine: GameEngine) -> None

On load hook. Default: no-op.

Source code in packages/maid-engine/src/maid_engine/plugins/protocol.py

async def on_load(self, engine: GameEngine) -> None:
    """On load hook. Default: no-op."""
    pass

on_unload async

on_unload(engine: GameEngine) -> None

On unload hook. Default: no-op.

Source code in packages/maid-engine/src/maid_engine/plugins/protocol.py

async def on_unload(self, engine: GameEngine) -> None:
    """On unload hook. Default: no-op."""
    pass

register_commands

register_commands(registry: AnyCommandRegistry) -> None

Register commands. Default: none.

Source code in packages/maid-engine/src/maid_engine/plugins/protocol.py

def register_commands(self, registry: AnyCommandRegistry) -> None:
    """Register commands. Default: none."""
    pass

register_document_schemas

register_document_schemas(store: DocumentStore) -> None

Register schemas. Default: none.

Source code in packages/maid-engine/src/maid_engine/plugins/protocol.py

def register_document_schemas(self, store: DocumentStore) -> None:
    """Register schemas. Default: none."""
    pass

CommandContext dataclass

CommandContext(
    session: Any,
    player_id: UUID,
    command: str,
    args: list[str],
    raw_input: str,
    world: World,
    document_store: DocumentStore | None = None,
    metadata: dict[str, Any] = dict(),
    target: Any | None = None,
)

Context passed to command handlers.

Attributes:

Name	Type	Description
session	Any	The player session
player_id	UUID	UUID of the player entity
command	str	The command name executed
args	list[str]	Command arguments
raw_input	str	Full raw input string
world	World	The game world
document_store	DocumentStore | None	Document store for persistence
metadata	dict[str, Any]	Additional metadata
target	Any | None	Optional target entity for lock evaluation (e.g., for owns()/holds() checks)

CommandDefinition dataclass

CommandDefinition(
    name: str,
    handler: CommandHandler,
    pack_name: str,
    priority: int = 0,
    aliases: list[str] = list(),
    category: str = "general",
    description: str = "",
    usage: str = "",
    access_level: AccessLevel = PLAYER,
    hidden: bool = False,
    locks: str = "",
    metadata: dict[str, Any] = dict(),
)

Definition of a registered command.

Attributes:

Name	Type	Description
name	str	Primary command name
handler	CommandHandler	Async function to handle the command
pack_name	str	Name of the content pack that registered this
priority	int	Priority for layered resolution
aliases	list[str]	Alternative names for this command
category	str	Command category for help
description	str	Short description
usage	str	Usage string
access_level	AccessLevel	Minimum access level required
hidden	bool	Whether to hide from help listings
locks	str	Lock expression for permission checks (e.g., "perm(admin) OR owns()")
metadata	dict[str, Any]	Additional metadata for hooks (e.g., cooldowns, combat state)

CommandRegistry

CommandRegistry()

Simple command registry without layering.

For basic command registration without content pack support.

Source code in packages/maid-engine/src/maid_engine/commands/registry.py

def __init__(self) -> None:
    self._commands: dict[str, CommandDefinition] = {}
    self._aliases: dict[str, str] = {}  # alias -> primary name

all_commands

all_commands() -> list[CommandDefinition]

Get all registered commands.

Source code in packages/maid-engine/src/maid_engine/commands/registry.py

def all_commands(self) -> list[CommandDefinition]:
    """Get all registered commands."""
    return list(self._commands.values())

commands_by_category

commands_by_category() -> dict[
    str, list[CommandDefinition]
]

Get commands grouped by category.

Source code in packages/maid-engine/src/maid_engine/commands/registry.py

def commands_by_category(self) -> dict[str, list[CommandDefinition]]:
    """Get commands grouped by category."""
    by_category: dict[str, list[CommandDefinition]] = defaultdict(list)
    for cmd in self._commands.values():
        if not cmd.hidden:
            by_category[cmd.category].append(cmd)
    return dict(by_category)

execute async

execute(
    context: CommandContext,
    player_access_level: AccessLevel = PLAYER,
) -> bool

Execute a command.

Parameters:

Name	Type	Description	Default
context	CommandContext	Command context	required
player_access_level	AccessLevel	The access level of the player executing	PLAYER

Returns:

Type	Description
bool	True if command was handled successfully

Raises:

Type	Description
PermissionError	If player lacks required access level

Source code in packages/maid-engine/src/maid_engine/commands/registry.py

async def execute(
    self,
    context: CommandContext,
    player_access_level: AccessLevel = AccessLevel.PLAYER,
) -> bool:
    """Execute a command.

    Args:
        context: Command context
        player_access_level: The access level of the player executing

    Returns:
        True if command was handled successfully

    Raises:
        PermissionError: If player lacks required access level
    """
    definition = self.get(context.command)
    if not definition:
        return False

    # Check access level
    if player_access_level < definition.access_level:
        raise PermissionError(
            f"Command '{definition.name}' requires {definition.access_level.name} "
            f"access (you have {player_access_level.name})"
        )

    result = await definition.handler(context)
    # Ensure we always return a bool, even if handler returns None
    return result if result is not None else True

get

get(name: str) -> CommandDefinition | None

Get a command by name or alias.

Source code in packages/maid-engine/src/maid_engine/commands/registry.py

def get(self, name: str) -> CommandDefinition | None:
    """Get a command by name or alias."""
    # Check if it's an alias
    if name in self._aliases:
        name = self._aliases[name]
    return self._commands.get(name)

register

register(
    name: str,
    handler: CommandHandler,
    *,
    aliases: list[str] | None = None,
    category: str = "general",
    description: str = "",
    usage: str = "",
    access_level: AccessLevel = PLAYER,
    hidden: bool = False,
) -> None

Register a command.

Source code in packages/maid-engine/src/maid_engine/commands/registry.py

def register(
    self,
    name: str,
    handler: CommandHandler,
    *,
    aliases: list[str] | None = None,
    category: str = "general",
    description: str = "",
    usage: str = "",
    access_level: AccessLevel = AccessLevel.PLAYER,
    hidden: bool = False,
) -> None:
    """Register a command."""
    definition = CommandDefinition(
        name=name,
        handler=handler,
        pack_name="core",
        aliases=aliases or [],
        category=category,
        description=description,
        usage=usage,
        access_level=access_level,
        hidden=hidden,
    )

    self._commands[name] = definition

    for alias in definition.aliases:
        self._aliases[alias] = name

unregister

unregister(name: str) -> bool

Unregister a command.

Source code in packages/maid-engine/src/maid_engine/commands/registry.py

def unregister(self, name: str) -> bool:
    """Unregister a command."""
    if name not in self._commands:
        return False

    definition = self._commands.pop(name)

    for alias in definition.aliases:
        self._aliases.pop(alias, None)

    return True

Component

Bases: BaseModel

Base class for all ECS components.

Components are pure data containers that can be attached to entities. They contain no logic - all behavior is implemented in Systems.

Example

class PositionComponent(Component): room_id: UUID x: int = 0 y: int = 0

get_type classmethod

get_type() -> str

Get the component type identifier.

Source code in packages/maid-engine/src/maid_engine/core/ecs/component.py

@classmethod
def get_type(cls) -> str:
    """Get the component type identifier."""
    return cls.component_type or cls.__name__

ContentPack

Bases: Protocol

Protocol defining the interface for content packs.

Content packs provide game functionality to the MAID engine. They can include: - ECS systems that process entities - Event types for communication - Commands for player interaction - Document schemas for persistence - Setup/teardown hooks

Important

Content packs that are discovered via entry points or directory scanning MUST have a no-argument constructor (or a constructor where all parameters have default values). This is because the discovery mechanism cannot pass constructor arguments. If your pack requires initialization parameters, consider using the on_load method or environment variables instead.

Example

class MyContentPack: def init(self) -> None: # No required arguments - this is important for auto-discovery! self._initialized = False

@property
def manifest(self) -> ContentPackManifest:
    return ContentPackManifest(
        name="my-pack",
        version="1.0.0",
        display_name="My Pack",
        description="A custom content pack",
    )

def get_dependencies(self) -> list[str]:
    return ["maid-stdlib"]

def get_systems(self, world: World) -> list[System]:
    return [MySystem(world)]

def get_events(self) -> list[type[Event]]:
    return [MyEvent]

def register_commands(self, registry: CommandRegistry) -> None:
    registry.register(...)

def register_document_schemas(self, store: DocumentStore) -> None:
    store.register_schema("my_collection", MyModel)

async def on_load(self, engine: GameEngine) -> None:
    # Initialize resources - use this for configuration that
    # would otherwise require constructor arguments
    self._initialized = True

async def on_unload(self, engine: GameEngine) -> None:
    # Cleanup resources
    pass

manifest property

manifest: ContentPackManifest

Get the content pack manifest.

Returns metadata about this pack including name, version, dependencies, and capabilities.

get_dependencies

get_dependencies() -> list[str]

Get the names of content packs this pack depends on.

Dependencies must be loaded before this pack can be loaded. Returns a list of content pack names (not versions).

Source code in packages/maid-engine/src/maid_engine/plugins/protocol.py

def get_dependencies(self) -> list[str]:
    """Get the names of content packs this pack depends on.

    Dependencies must be loaded before this pack can be loaded.
    Returns a list of content pack names (not versions).
    """
    ...

get_events

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

Get event types defined by this pack.

These event types can be used by the EventBus for publish/subscribe communication.

Returns:

Type	Description
list[type[Event]]	List of Event subclasses

Source code in packages/maid-engine/src/maid_engine/plugins/protocol.py

def get_events(self) -> list[type[Event]]:
    """Get event types defined by this pack.

    These event types can be used by the EventBus for
    publish/subscribe communication.

    Returns:
        List of Event subclasses
    """
    ...

get_systems

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

Get ECS systems provided by this pack.

Systems are registered with the engine's SystemManager and are updated each tick.

Parameters:

Name	Type	Description	Default
world	World	The game world to create systems for	required

Returns:

Type	Description
list[System]	List of System instances to register

Source code in packages/maid-engine/src/maid_engine/plugins/protocol.py

def get_systems(self, world: World) -> list[System]:
    """Get ECS systems provided by this pack.

    Systems are registered with the engine's SystemManager and
    are updated each tick.

    Args:
        world: The game world to create systems for

    Returns:
        List of System instances to register
    """
    ...

on_load async

on_load(engine: GameEngine) -> None

Called when the pack is loaded.

Use this hook to: - Initialize resources - Load data files - Subscribe to events - Set up background tasks

Parameters:

Name	Type	Description	Default
engine	GameEngine	The game engine	required

Source code in packages/maid-engine/src/maid_engine/plugins/protocol.py

async def on_load(self, engine: GameEngine) -> None:
    """Called when the pack is loaded.

    Use this hook to:
    - Initialize resources
    - Load data files
    - Subscribe to events
    - Set up background tasks

    Args:
        engine: The game engine
    """
    ...

on_unload async

on_unload(engine: GameEngine) -> None

Called when the pack is unloaded.

Use this hook to: - Clean up resources - Save state - Unsubscribe from events - Cancel background tasks

Parameters:

Name	Type	Description	Default
engine	GameEngine	The game engine	required

Source code in packages/maid-engine/src/maid_engine/plugins/protocol.py

async def on_unload(self, engine: GameEngine) -> None:
    """Called when the pack is unloaded.

    Use this hook to:
    - Clean up resources
    - Save state
    - Unsubscribe from events
    - Cancel background tasks

    Args:
        engine: The game engine
    """
    ...

register_commands

register_commands(registry: AnyCommandRegistry) -> None

Register commands provided by this pack.

Commands are registered with the engine's CommandRegistry and can be executed by players.

Parameters:

Name	Type	Description	Default
registry	AnyCommandRegistry	The command registry to register with	required

Source code in packages/maid-engine/src/maid_engine/plugins/protocol.py

def register_commands(self, registry: AnyCommandRegistry) -> None:
    """Register commands provided by this pack.

    Commands are registered with the engine's CommandRegistry
    and can be executed by players.

    Args:
        registry: The command registry to register with
    """
    ...

register_document_schemas

register_document_schemas(store: DocumentStore) -> None

Register document schemas used by this pack.

Schemas define the structure of documents stored in the document store.

Parameters:

Name	Type	Description	Default
store	DocumentStore	The document store to register schemas with	required

Source code in packages/maid-engine/src/maid_engine/plugins/protocol.py

def register_document_schemas(self, store: DocumentStore) -> None:
    """Register document schemas used by this pack.

    Schemas define the structure of documents stored in the
    document store.

    Args:
        store: The document store to register schemas with
    """
    ...

ContentPackLoader

ContentPackLoader()

Manages content pack loading with dependency resolution.

Example

loader = ContentPackLoader()

Discover and register available packs

for pack in discover_content_packs(): loader.register(pack)

Load packs in dependency order

ordered_packs = loader.resolve_load_order() for pack in ordered_packs: engine.load_content_pack(pack)

Source code in packages/maid-engine/src/maid_engine/plugins/loader.py

def __init__(self) -> None:
    self._available: dict[str, ContentPack] = {}
    self._loaded: set[str] = set()

available_packs property

available_packs: list[str]

Get names of all available packs.

loaded_packs property

loaded_packs: set[str]

Get names of loaded packs.

can_unload

can_unload(pack_name: str) -> tuple[bool, list[str]]

Check if a pack can be unloaded.

Returns:

Type	Description
tuple[bool, list[str]]	Tuple of (can_unload, list of blocking dependents)

Source code in packages/maid-engine/src/maid_engine/plugins/loader.py

def can_unload(self, pack_name: str) -> tuple[bool, list[str]]:
    """Check if a pack can be unloaded.

    Returns:
        Tuple of (can_unload, list of blocking dependents)
    """
    dependents = self.get_dependents(pack_name)
    loaded_dependents = [d for d in dependents if d in self._loaded]
    return len(loaded_dependents) == 0, loaded_dependents

get

get(pack_name: str) -> ContentPack | None

Get a registered pack by name.

Source code in packages/maid-engine/src/maid_engine/plugins/loader.py

def get(self, pack_name: str) -> ContentPack | None:
    """Get a registered pack by name."""
    return self._available.get(pack_name)

get_dependents

get_dependents(pack_name: str) -> list[str]

Get packs that depend on the given pack.

Source code in packages/maid-engine/src/maid_engine/plugins/loader.py

def get_dependents(self, pack_name: str) -> list[str]:
    """Get packs that depend on the given pack."""
    dependents: list[str] = []
    for name, pack in self._available.items():
        if pack_name in pack.get_dependencies():
            dependents.append(name)
    return dependents

mark_loaded

mark_loaded(pack_name: str) -> None

Mark a pack as loaded.

Source code in packages/maid-engine/src/maid_engine/plugins/loader.py

def mark_loaded(self, pack_name: str) -> None:
    """Mark a pack as loaded."""
    self._loaded.add(pack_name)

mark_unloaded

mark_unloaded(pack_name: str) -> None

Mark a pack as unloaded.

Source code in packages/maid-engine/src/maid_engine/plugins/loader.py

def mark_unloaded(self, pack_name: str) -> None:
    """Mark a pack as unloaded."""
    self._loaded.discard(pack_name)

register

register(pack: ContentPack) -> None

Register an available content pack.

Source code in packages/maid-engine/src/maid_engine/plugins/loader.py

def register(self, pack: ContentPack) -> None:
    """Register an available content pack."""
    self._available[pack.manifest.name] = pack

resolve_load_order

resolve_load_order(
    pack_names: list[str] | None = None,
) -> list[ContentPack]

Resolve the loading order for content packs.

Uses topological sort to order packs by dependencies.

Parameters:

Name	Type	Description	Default
pack_names	list[str] | None	Specific packs to load (default: all available)	None

Returns:

Type	Description
list[ContentPack]	List of packs in dependency order

Raises:

Type	Description
DependencyError	If dependencies cannot be resolved

Source code in packages/maid-engine/src/maid_engine/plugins/loader.py

def resolve_load_order(
    self, pack_names: list[str] | None = None
) -> list[ContentPack]:
    """Resolve the loading order for content packs.

    Uses topological sort to order packs by dependencies.

    Args:
        pack_names: Specific packs to load (default: all available)

    Returns:
        List of packs in dependency order

    Raises:
        DependencyError: If dependencies cannot be resolved
    """
    if pack_names is None:
        pack_names = list(self._available.keys())

    # Build dependency graph - include transitive dependencies
    graph: dict[str, set[str]] = {}
    to_process = list(pack_names)
    processed = set()

    while to_process:
        name = to_process.pop(0)
        if name in processed:
            continue
        processed.add(name)

        pack = self._available.get(name)
        if not pack:
            raise DependencyError(f"Content pack '{name}' not found")

        deps = set(pack.get_dependencies())
        graph[name] = deps

        # Verify all dependencies are available and queue them for processing
        for dep in deps:
            if dep not in self._available:
                raise DependencyError(
                    f"Content pack '{name}' requires '{dep}' which is not available"
                )
            # Add transitive dependency to processing queue
            if dep not in processed and dep not in to_process:
                to_process.append(dep)

    # Topological sort
    ordered: list[str] = []
    visited: set[str] = set()
    visiting: set[str] = set()

    def visit(name: str) -> None:
        if name in visited:
            return
        if name in visiting:
            raise DependencyError(
                f"Circular dependency detected involving '{name}'"
            )

        visiting.add(name)

        for dep in graph.get(name, set()):
            visit(dep)

        visiting.remove(name)
        visited.add(name)
        ordered.append(name)

    # Visit all packs in the graph (including transitive dependencies)
    for name in graph:
        visit(name)

    return [self._available[name] for name in ordered]

unregister

unregister(pack_name: str) -> ContentPack | None

Unregister a content pack.

Source code in packages/maid-engine/src/maid_engine/plugins/loader.py

def unregister(self, pack_name: str) -> ContentPack | None:
    """Unregister a content pack."""
    return self._available.pop(pack_name, None)

ContentPackManifest dataclass

ContentPackManifest(
    name: str,
    version: str,
    display_name: str = "",
    description: str = "",
    dependencies: dict[str, str] = dict(),
    provides: list[str] = list(),
    requires: list[str] = list(),
    authors: list[str] = list(),
    license: str = "",
    homepage: str = "",
    keywords: list[str] = list(),
)

Metadata about a content pack.

Attributes:

Name	Type	Description
name	str	Unique identifier for the pack (e.g., "classic-rpg")
version	str	Semantic version string (e.g., "0.1.0")
display_name	str	Human-readable name (e.g., "Classic RPG")
description	str	Brief description of the pack
dependencies	dict[str, str]	Dict of pack_name -> version_constraint
provides	list[str]	List of capability identifiers this pack provides
requires	list[str]	List of capability identifiers this pack requires
authors	list[str]	List of author names
license	str	SPDX license identifier
homepage	str	URL to pack homepage/repository
keywords	list[str]	List of keywords for discovery

from_dict classmethod

from_dict(data: dict[str, Any]) -> ContentPackManifest

Create manifest from a dictionary.

Parameters:

Name	Type	Description	Default
data	dict[str, Any]	Dictionary with 'pack' key containing manifest data	required

Returns:

Type	Description
ContentPackManifest	ContentPackManifest instance

Source code in packages/maid-engine/src/maid_engine/plugins/manifest.py

@classmethod
def from_dict(cls, data: dict[str, Any]) -> ContentPackManifest:
    """Create manifest from a dictionary.

    Args:
        data: Dictionary with 'pack' key containing manifest data

    Returns:
        ContentPackManifest instance
    """
    pack_data = data.get("pack", data)

    # Extract capabilities section
    capabilities = pack_data.get("capabilities", {})

    return cls(
        name=pack_data["name"],
        version=pack_data["version"],
        display_name=pack_data.get("display_name", ""),
        description=pack_data.get("description", ""),
        dependencies=pack_data.get("dependencies", {}),
        provides=capabilities.get("provides", []),
        requires=capabilities.get("requires", []),
        authors=pack_data.get("authors", []),
        license=pack_data.get("license", ""),
        homepage=pack_data.get("homepage", ""),
        keywords=pack_data.get("keywords", []),
    )

from_toml classmethod

from_toml(path: Path) -> ContentPackManifest

Load manifest from a TOML file.

Expected format

[pack] name = "my-pack" version = "1.0.0" display_name = "My Pack" description = "A custom content pack"

[pack.dependencies] maid-stdlib = ">=0.1.0"

[pack.capabilities] provides = ["combat-system", "magic-system"] requires = ["ecs", "event-bus"]

Source code in packages/maid-engine/src/maid_engine/plugins/manifest.py

@classmethod
def from_toml(cls, path: Path) -> ContentPackManifest:
    """Load manifest from a TOML file.

    Expected format:
        [pack]
        name = "my-pack"
        version = "1.0.0"
        display_name = "My Pack"
        description = "A custom content pack"

        [pack.dependencies]
        maid-stdlib = ">=0.1.0"

        [pack.capabilities]
        provides = ["combat-system", "magic-system"]
        requires = ["ecs", "event-bus"]
    """
    with open(path, "rb") as f:
        data = tomllib.load(f)

    return cls.from_dict(data)

satisfies_dependency

satisfies_dependency(name: str, version: str) -> bool

Check if this pack satisfies a dependency requirement.

Parameters:

Name	Type	Description	Default
name	str	Required pack name	required
version	str	Version constraint (e.g., ">=0.1.0")	required

Returns:

Type	Description
bool	True if this pack satisfies the requirement

Source code in packages/maid-engine/src/maid_engine/plugins/manifest.py

def satisfies_dependency(self, name: str, version: str) -> bool:
    """Check if this pack satisfies a dependency requirement.

    Args:
        name: Required pack name
        version: Version constraint (e.g., ">=0.1.0")

    Returns:
        True if this pack satisfies the requirement
    """
    if self.name != name:
        return False

    # Simple version comparison (for now, just check equality or >=)
    constraint = version.strip()
    if constraint.startswith(">="):
        min_version = constraint[2:].strip()
        return self._compare_versions(self.version, min_version) >= 0
    elif constraint.startswith(">"):
        min_version = constraint[1:].strip()
        return self._compare_versions(self.version, min_version) > 0
    elif constraint.startswith("<="):
        max_version = constraint[2:].strip()
        return self._compare_versions(self.version, max_version) <= 0
    elif constraint.startswith("<"):
        max_version = constraint[1:].strip()
        return self._compare_versions(self.version, max_version) < 0
    elif constraint.startswith("==") or constraint.startswith("="):
        exact_version = constraint.lstrip("=").strip()
        return self.version == exact_version
    else:
        # Assume exact match
        return self.version == constraint

to_dict

to_dict() -> dict[str, Any]

Convert manifest to dictionary.

Source code in packages/maid-engine/src/maid_engine/plugins/manifest.py

def to_dict(self) -> dict[str, Any]:
    """Convert manifest to dictionary."""
    return {
        "pack": {
            "name": self.name,
            "version": self.version,
            "display_name": self.display_name,
            "description": self.description,
            "dependencies": self.dependencies,
            "capabilities": {
                "provides": self.provides,
                "requires": self.requires,
            },
            "authors": self.authors,
            "license": self.license,
            "homepage": self.homepage,
            "keywords": self.keywords,
        }
    }

DocumentCollection

Bases: ABC

Abstract base class for document collections.

A collection holds documents of a specific type, identified by UUID. All operations are async for compatibility with async databases.

Example

Get a collection

characters = store.get_collection("characters")

CRUD operations

doc_id = await characters.create(character_data) character = await characters.get(doc_id) await characters.update(doc_id, updated_data) await characters.delete(doc_id)

Query

results = await characters.query(QueryOptions( filters={"level": 10}, order_by="name", limit=20 ))

name abstractmethod property

name: str

Get the collection name.

count abstractmethod async

count(filters: dict[str, Any] | None = None) -> int

Count documents.

Parameters:

Name	Type	Description	Default
filters	dict[str, Any] | None	Optional filters to apply	None

Returns:

Type	Description
int	Number of matching documents

Source code in packages/maid-engine/src/maid_engine/storage/document_store.py

@abstractmethod
async def count(self, filters: dict[str, Any] | None = None) -> int:
    """Count documents.

    Args:
        filters: Optional filters to apply

    Returns:
        Number of matching documents
    """
    ...

create abstractmethod async

create(document: T, doc_id: UUID | None = None) -> UUID

Create a new document.

Parameters:

Name	Type	Description	Default
document	T	Document data	required
doc_id	UUID | None	Optional specific UUID (generated if not provided)	None

Returns:

Type	Description
UUID	UUID of created document

Source code in packages/maid-engine/src/maid_engine/storage/document_store.py

@abstractmethod
async def create(self, document: T, doc_id: UUID | None = None) -> UUID:
    """Create a new document.

    Args:
        document: Document data
        doc_id: Optional specific UUID (generated if not provided)

    Returns:
        UUID of created document
    """
    ...

delete abstractmethod async

delete(doc_id: UUID) -> bool

Delete a document.

Parameters:

Name	Type	Description	Default
doc_id	UUID	Document UUID	required

Returns:

Type	Description
bool	True if document was deleted, False if not found

Source code in packages/maid-engine/src/maid_engine/storage/document_store.py

@abstractmethod
async def delete(self, doc_id: UUID) -> bool:
    """Delete a document.

    Args:
        doc_id: Document UUID

    Returns:
        True if document was deleted, False if not found
    """
    ...

exists abstractmethod async

exists(doc_id: UUID) -> bool

Check if a document exists.

Parameters:

Name	Type	Description	Default
doc_id	UUID	Document UUID	required

Returns:

Type	Description
bool	True if document exists

Source code in packages/maid-engine/src/maid_engine/storage/document_store.py

@abstractmethod
async def exists(self, doc_id: UUID) -> bool:
    """Check if a document exists.

    Args:
        doc_id: Document UUID

    Returns:
        True if document exists
    """
    ...

get abstractmethod async

get(doc_id: UUID) -> T | None

Get a document by ID.

Parameters:

Name	Type	Description	Default
doc_id	UUID	Document UUID	required

Returns:

Type	Description
T | None	Document if found, None otherwise

Source code in packages/maid-engine/src/maid_engine/storage/document_store.py

@abstractmethod
async def get(self, doc_id: UUID) -> T | None:
    """Get a document by ID.

    Args:
        doc_id: Document UUID

    Returns:
        Document if found, None otherwise
    """
    ...

get_many abstractmethod async

get_many(doc_ids: list[UUID]) -> list[T]

Get multiple documents by ID.

Parameters:

Name	Type	Description	Default
doc_ids	list[UUID]	List of document UUIDs	required

Returns:

Type	Description
list[T]	List of found documents (may be shorter than doc_ids)

Source code in packages/maid-engine/src/maid_engine/storage/document_store.py

@abstractmethod
async def get_many(self, doc_ids: list[UUID]) -> list[T]:
    """Get multiple documents by ID.

    Args:
        doc_ids: List of document UUIDs

    Returns:
        List of found documents (may be shorter than doc_ids)
    """
    ...

query abstractmethod async

query(options: QueryOptions) -> list[T]

Query documents.

Parameters:

Name	Type	Description	Default
options	QueryOptions	Query options (filters, ordering, pagination)	required

Returns:

Type	Description
list[T]	List of matching documents

Source code in packages/maid-engine/src/maid_engine/storage/document_store.py

@abstractmethod
async def query(self, options: QueryOptions) -> list[T]:
    """Query documents.

    Args:
        options: Query options (filters, ordering, pagination)

    Returns:
        List of matching documents
    """
    ...

update abstractmethod async

update(doc_id: UUID, document: T) -> bool

Update an existing document.

Parameters:

Name	Type	Description	Default
doc_id	UUID	Document UUID	required
document	T	New document data	required

Returns:

Type	Description
bool	True if document was updated, False if not found

Source code in packages/maid-engine/src/maid_engine/storage/document_store.py

@abstractmethod
async def update(self, doc_id: UUID, document: T) -> bool:
    """Update an existing document.

    Args:
        doc_id: Document UUID
        document: New document data

    Returns:
        True if document was updated, False if not found
    """
    ...

DocumentStore

Bases: ABC

Abstract base class for the document store.

The document store manages collections of documents. Content packs register their document schemas, then use collections for persistence.

Example

In content pack initialization

store.register_schema("characters", CharacterModel) store.register_schema("items", ItemModel)

Later, in game logic

characters = store.get_collection("characters") character = await characters.get(character_id)

query_collector property

query_collector: QueryCollector | None

Get the query collector for profiling (if enabled).

close abstractmethod async

close() -> None

Close the document store.

Closes database connections and cleans up resources.

Source code in packages/maid-engine/src/maid_engine/storage/document_store.py

@abstractmethod
async def close(self) -> None:
    """Close the document store.

    Closes database connections and cleans up resources.
    """
    ...

collection_names abstractmethod

collection_names() -> list[str]

Get names of all registered collections.

Returns:

Type	Description
list[str]	List of collection names

Source code in packages/maid-engine/src/maid_engine/storage/document_store.py

@abstractmethod
def collection_names(self) -> list[str]:
    """Get names of all registered collections.

    Returns:
        List of collection names
    """
    ...

get_collection abstractmethod

get_collection(name: str) -> DocumentCollection[Any]

Get a document collection.

Parameters:

Name	Type	Description	Default
name	str	Collection name	required

Returns:

Type	Description
DocumentCollection[Any]	DocumentCollection for the given name

Raises:

Type	Description
KeyError	If collection is not registered

Source code in packages/maid-engine/src/maid_engine/storage/document_store.py

@abstractmethod
def get_collection(self, name: str) -> DocumentCollection[Any]:
    """Get a document collection.

    Args:
        name: Collection name

    Returns:
        DocumentCollection for the given name

    Raises:
        KeyError: If collection is not registered
    """
    ...

has_collection abstractmethod

has_collection(name: str) -> bool

Check if a collection is registered.

Parameters:

Name	Type	Description	Default
name	str	Collection name	required

Returns:

Type	Description
bool	True if collection exists

Source code in packages/maid-engine/src/maid_engine/storage/document_store.py

@abstractmethod
def has_collection(self, name: str) -> bool:
    """Check if a collection is registered.

    Args:
        name: Collection name

    Returns:
        True if collection exists
    """
    ...

initialize abstractmethod async

initialize() -> None

Initialize the document store.

Creates necessary database tables/structures. Should be called before using the store.

Source code in packages/maid-engine/src/maid_engine/storage/document_store.py

@abstractmethod
async def initialize(self) -> None:
    """Initialize the document store.

    Creates necessary database tables/structures.
    Should be called before using the store.
    """
    ...

register_schema abstractmethod

register_schema(name: str, schema: type[BaseModel]) -> None

Register a document schema.

Must be called before using a collection. The schema defines the structure of documents in the collection.

Parameters:

Name	Type	Description	Default
name	str	Collection name	required
schema	type[BaseModel]	Pydantic model class for document validation	required

Source code in packages/maid-engine/src/maid_engine/storage/document_store.py

@abstractmethod
def register_schema(self, name: str, schema: type[BaseModel]) -> None:
    """Register a document schema.

    Must be called before using a collection. The schema defines
    the structure of documents in the collection.

    Args:
        name: Collection name
        schema: Pydantic model class for document validation
    """
    ...

set_query_collector

set_query_collector(
    collector: QueryCollector | None,
) -> None

Set the query collector for profiling.

The query collector must be started externally (e.g., via ProfileManager). When set, the store will record query profiling data during operations.

Parameters:

Name	Type	Description	Default
collector	QueryCollector | None	The QueryCollector instance, or None to disable profiling.	required

Source code in packages/maid-engine/src/maid_engine/storage/document_store.py

def set_query_collector(self, collector: QueryCollector | None) -> None:
    """Set the query collector for profiling.

    The query collector must be started externally (e.g., via ProfileManager).
    When set, the store will record query profiling data during operations.

    Args:
        collector: The QueryCollector instance, or None to disable profiling.
    """
    self._query_collector = collector

EngineState

Bases: Enum

Game engine states.

Entity

Entity(
    entity_id: UUID | None = None,
    manager: EntityManager | None = None,
)

Represents a game entity as a collection of components.

Entities are essentially just IDs that group components together. The EntityManager handles the actual component storage and queries.

Example

entity = Entity() entity.add(PositionComponent(room_id=room.id)) entity.add(HealthComponent(current=100, maximum=100))

if entity.has(HealthComponent): health = entity.get(HealthComponent) health.damage(10)

Source code in packages/maid-engine/src/maid_engine/core/ecs/entity.py

def __init__(
    self,
    entity_id: UUID | None = None,
    manager: "EntityManager | None" = None,
) -> None:
    self._id = entity_id or uuid4()
    self._components: dict[type[Component], Component] = {}
    self._tags: set[str] = set()
    self._manager = manager
    self._created_at: datetime = datetime.now(UTC)
    self._updated_at: datetime = self._created_at

components property

components: Iterator[Component]

Iterate over all components.

created_at property

created_at: datetime

Get entity creation timestamp.

id property

id: UUID

Get entity ID.

tags property

tags: set[str]

Get entity tags.

updated_at property

updated_at: datetime

Get entity last updated timestamp.

add

add(component: Component) -> Entity

Add a component to this entity.

Source code in packages/maid-engine/src/maid_engine/core/ecs/entity.py

def add(self, component: "Component") -> "Entity":
    """Add a component to this entity."""
    component_type = type(component)
    self._components[component_type] = component
    self._mark_updated()
    if self._manager:
        self._manager._on_component_added(self, component_type)
    return self

add_tag

add_tag(tag: str) -> Entity

Add a tag to this entity.

Source code in packages/maid-engine/src/maid_engine/core/ecs/entity.py

def add_tag(self, tag: str) -> "Entity":
    """Add a tag to this entity."""
    self._tags.add(tag)
    self._mark_updated()
    if self._manager:
        self._manager._on_tag_added(self, tag)
    return self

get

get(component_type: type[C]) -> C

Get a component by type. Raises KeyError if not found.

Source code in packages/maid-engine/src/maid_engine/core/ecs/entity.py

def get(self, component_type: type[C]) -> C:
    """Get a component by type. Raises KeyError if not found."""
    component = self._components.get(component_type)
    if component is None:
        raise KeyError(f"Entity {self._id} does not have component {component_type.__name__}")
    # Safe cast: we looked up by component_type, so result is an instance of C
    return component  # type: ignore[return-value]

has

has(*component_types: type[Component]) -> bool

Check if entity has all specified component types.

Source code in packages/maid-engine/src/maid_engine/core/ecs/entity.py

def has(self, *component_types: type["Component"]) -> bool:
    """Check if entity has all specified component types."""
    return all(ct in self._components for ct in component_types)

has_any

has_any(*component_types: type[Component]) -> bool

Check if entity has any of the specified component types.

Source code in packages/maid-engine/src/maid_engine/core/ecs/entity.py

def has_any(self, *component_types: type["Component"]) -> bool:
    """Check if entity has any of the specified component types."""
    return any(ct in self._components for ct in component_types)

has_tag

has_tag(tag: str) -> bool

Check if entity has a specific tag.

Source code in packages/maid-engine/src/maid_engine/core/ecs/entity.py

def has_tag(self, tag: str) -> bool:
    """Check if entity has a specific tag."""
    return tag in self._tags

remove

remove(component_type: type[C]) -> C | None

Remove and return a component from this entity.

Source code in packages/maid-engine/src/maid_engine/core/ecs/entity.py

def remove(self, component_type: type[C]) -> C | None:
    """Remove and return a component from this entity."""
    component = self._components.pop(component_type, None)
    if component:
        self._mark_updated()
        if self._manager:
            self._manager._on_component_removed(self, component_type)
    # Safe cast: component is None or matches component_type since we used it as dict key
    return component if component is None else component  # type: ignore[return-value]

remove_tag

remove_tag(tag: str) -> bool

Remove a tag from this entity. Returns True if tag was present.

Source code in packages/maid-engine/src/maid_engine/core/ecs/entity.py

def remove_tag(self, tag: str) -> bool:
    """Remove a tag from this entity. Returns True if tag was present."""
    if tag in self._tags:
        self._tags.discard(tag)
        self._mark_updated()
        if self._manager:
            self._manager._on_tag_removed(self, tag)
        return True
    return False

to_dict

to_dict() -> dict[str, Any]

Serialize entity to dictionary.

Source code in packages/maid-engine/src/maid_engine/core/ecs/entity.py

def to_dict(self) -> dict[str, Any]:
    """Serialize entity to dictionary."""
    return {
        "id": str(self._id),
        "tags": list(self._tags),
        "components": {
            comp_type.__name__: comp.model_dump()
            for comp_type, comp in self._components.items()
        },
        "created_at": self._created_at.isoformat(),
        "updated_at": self._updated_at.isoformat(),
    }

try_get

try_get(component_type: type[C]) -> C | None

Get a component by type, or None if not found.

Source code in packages/maid-engine/src/maid_engine/core/ecs/entity.py

def try_get(self, component_type: type[C]) -> C | None:
    """Get a component by type, or None if not found."""
    # Safe cast: we looked up by component_type, so result is C | None
    return self._components.get(component_type)  # type: ignore[return-value]

EntityManager

EntityManager()

Manages all entities and provides efficient component queries.

The EntityManager maintains indexes for fast component-based queries and handles entity lifecycle events.

Example

manager = EntityManager()

Create entities

player = manager.create() player.add(PositionComponent(room_id=room_id)) player.add(HealthComponent(current=100, maximum=100))

Query entities with specific components

for entity in manager.with_components(PositionComponent, HealthComponent): pos = entity.get(PositionComponent) health = entity.get(HealthComponent)

Source code in packages/maid-engine/src/maid_engine/core/ecs/entity.py

def __init__(self) -> None:
    self._entities: dict[UUID, Entity] = {}
    self._by_component: dict[type[Component], set[UUID]] = defaultdict(set)
    self._by_tag: dict[str, set[UUID]] = defaultdict(set)

all

all() -> Iterator[Entity]

Iterate over all entities.

Source code in packages/maid-engine/src/maid_engine/core/ecs/entity.py

def all(self) -> Iterator[Entity]:
    """Iterate over all entities."""
    return iter(self._entities.values())

clear

clear() -> None

Remove all entities.

Source code in packages/maid-engine/src/maid_engine/core/ecs/entity.py

def clear(self) -> None:
    """Remove all entities."""
    for entity in list(self._entities.values()):
        entity._manager = None
    self._entities.clear()
    self._by_component.clear()
    self._by_tag.clear()

count

count() -> int

Get total number of entities.

Source code in packages/maid-engine/src/maid_engine/core/ecs/entity.py

def count(self) -> int:
    """Get total number of entities."""
    return len(self._entities)

create

create(entity_id: UUID | None = None) -> Entity

Create a new entity.

Source code in packages/maid-engine/src/maid_engine/core/ecs/entity.py

def create(self, entity_id: UUID | None = None) -> Entity:
    """Create a new entity."""
    entity = Entity(entity_id=entity_id, manager=self)
    self._entities[entity.id] = entity
    return entity

destroy

destroy(entity_id: UUID) -> bool

Destroy an entity. Returns True if entity existed.

Source code in packages/maid-engine/src/maid_engine/core/ecs/entity.py

def destroy(self, entity_id: UUID) -> bool:
    """Destroy an entity. Returns True if entity existed."""
    entity = self._entities.pop(entity_id, None)
    if entity is None:
        return False

    # Clean up indexes
    for component_type in list(entity._components.keys()):
        self._by_component[component_type].discard(entity_id)

    for tag in list(entity._tags):
        self._by_tag[tag].discard(entity_id)

    entity._manager = None
    return True

get

get(entity_id: UUID) -> Entity | None

Get an entity by ID.

Source code in packages/maid-engine/src/maid_engine/core/ecs/entity.py

def get(self, entity_id: UUID) -> Entity | None:
    """Get an entity by ID."""
    return self._entities.get(entity_id)

get_or_raise

get_or_raise(entity_id: UUID) -> Entity

Get an entity by ID, raising KeyError if not found.

Source code in packages/maid-engine/src/maid_engine/core/ecs/entity.py

def get_or_raise(self, entity_id: UUID) -> Entity:
    """Get an entity by ID, raising KeyError if not found."""
    entity = self._entities.get(entity_id)
    if entity is None:
        raise KeyError(f"Entity {entity_id} not found")
    return entity

with_components

with_components(
    *component_types: type[Component],
) -> Iterator[Entity]

Get all entities that have all specified component types.

Source code in packages/maid-engine/src/maid_engine/core/ecs/entity.py

def with_components(self, *component_types: type["Component"]) -> Iterator[Entity]:
    """Get all entities that have all specified component types."""
    if not component_types:
        yield from self._entities.values()
        return

    # Start with the smallest set for efficiency
    sets = [self._by_component.get(ct, set()) for ct in component_types]
    if not all(sets):
        return

    smallest = min(sets, key=len)
    candidates = smallest.intersection(*sets)

    for entity_id in candidates:
        entity = self._entities.get(entity_id)
        if entity:
            yield entity

with_tag

with_tag(tag: str) -> Iterator[Entity]

Get all entities with a specific tag.

Source code in packages/maid-engine/src/maid_engine/core/ecs/entity.py

def with_tag(self, tag: str) -> Iterator[Entity]:
    """Get all entities with a specific tag."""
    for entity_id in self._by_tag.get(tag, set()):
        entity = self._entities.get(entity_id)
        if entity:
            yield entity

with_tags

with_tags(*tags: str) -> Iterator[Entity]

Get all entities that have all specified tags.

Source code in packages/maid-engine/src/maid_engine/core/ecs/entity.py

def with_tags(self, *tags: str) -> Iterator[Entity]:
    """Get all entities that have all specified tags."""
    if not tags:
        yield from self._entities.values()
        return

    sets = [self._by_tag.get(tag, set()) for tag in tags]
    if not all(sets):
        return

    smallest = min(sets, key=len)
    candidates = smallest.intersection(*sets)

    for entity_id in candidates:
        entity = self._entities.get(entity_id)
        if entity:
            yield entity

Event dataclass

Event()

Base class for all events.

Events are messages that can be emitted and handled by subscribers. They carry data about something that happened in the game.

Example

@dataclass class PlayerDamagedEvent(Event): player_id: UUID damage: int source_id: UUID | None = None

cancel

cancel() -> None

Cancel this event to prevent further processing.

Source code in packages/maid-engine/src/maid_engine/core/events.py

def cancel(self) -> None:
    """Cancel this event to prevent further processing."""
    self.cancelled = True

EventBus

EventBus()

Publish/subscribe event system.

The EventBus allows systems to communicate through events without direct coupling. Handlers can be async or sync.

Supports wildcard subscriptions via subscribe_all() to receive all events.

Example

bus = EventBus()

async def on_damage(event: DamageDealtEvent): print(f"Damage dealt: {event.damage}")

bus.subscribe(DamageDealtEvent, on_damage)

Subscribe to ALL events (wildcard)

def log_event(event: Event): print(f"Event: {event.event_type}") bus.subscribe_all(log_event)

await bus.emit(DamageDealtEvent( source_id=attacker.id, target_id=defender.id, damage=25, damage_type="physical" ))

Source code in packages/maid-engine/src/maid_engine/core/events.py

def __init__(self) -> None:
    self._handlers: dict[str, list[HandlerRegistration]] = defaultdict(list)
    self._pending_events: asyncio.Queue[Event] = asyncio.Queue()
    # Track handler->pack mapping for hot reload support
    self._handler_sources: dict[UUID, str] = {}  # handler_id -> pack_name
    # Track handlers by pack for efficient bulk unsubscription
    self._handlers_by_pack: dict[str, set[UUID]] = defaultdict(set)
    # Lock for safe handler removal during dispatch
    self._dispatch_lock = RLock()
    self._dispatch_depth = 0  # Track nested dispatches
    # Queue for deferred removals during dispatch
    self._pending_removals: set[UUID] = set()
    self._pending_pack_removals: set[str] = set()
    # Track total number of events processed for metrics
    self._processed_count: int = 0

processed_count property

processed_count: int

Get the total number of events that have been processed.

This count includes all events emitted via emit() since the EventBus was created or last cleared. It does not include events queued via emit_sync() that haven't been processed yet.

Returns:

Type	Description
int	Total number of events processed

clear

clear() -> None

Remove all handlers, pending events, and reset counters.

Source code in packages/maid-engine/src/maid_engine/core/events.py

def clear(self) -> None:
    """Remove all handlers, pending events, and reset counters."""
    self._handlers.clear()
    self._handler_sources.clear()
    self._handlers_by_pack.clear()
    self._pending_removals.clear()
    self._pending_pack_removals.clear()
    self._processed_count = 0
    # Clear pending events queue
    while not self._pending_events.empty():
        try:
            self._pending_events.get_nowait()
        except asyncio.QueueEmpty:
            break

emit async

emit(event: Event) -> None

Emit an event immediately, invoking all handlers.

Handlers are invoked in priority order. If an event is cancelled, remaining handlers are not called.

Both type-specific handlers and wildcard handlers (registered via subscribe_all) are invoked.

Handler removals during dispatch are deferred until dispatch completes.

Source code in packages/maid-engine/src/maid_engine/core/events.py

async def emit(self, event: Event) -> None:
    """Emit an event immediately, invoking all handlers.

    Handlers are invoked in priority order. If an event is
    cancelled, remaining handlers are not called.

    Both type-specific handlers and wildcard handlers (registered via
    subscribe_all) are invoked.

    Handler removals during dispatch are deferred until dispatch completes.
    """
    type_name = event.event_type

    # Enter dispatch mode
    with self._dispatch_lock:
        self._dispatch_depth += 1

    try:
        # Copy handlers list to avoid concurrent modification during iteration
        # Include both type-specific handlers and wildcard handlers
        type_handlers = list(self._handlers.get(type_name, []))
        wildcard_handlers = list(self._handlers.get(self.WILDCARD_KEY, []))

        # Combine and sort all handlers by priority
        all_handlers = type_handlers + wildcard_handlers
        all_handlers.sort(key=lambda h: h.priority.value)

        to_remove: list[UUID] = []

        for registration in all_handlers:
            # Skip handlers that are pending removal
            if registration.handler_id in self._pending_removals:
                continue

            # Skip handlers from packs pending removal
            if (
                registration.source_pack
                and registration.source_pack in self._pending_pack_removals
            ):
                continue

            if event.cancelled:
                break

            try:
                await registration.handler(event)
            except Exception:
                # Log but don't stop other handlers
                # In production, this should use proper logging
                import traceback

                traceback.print_exc()

            if registration.once:
                to_remove.append(registration.handler_id)

        # Queue one-time handlers for removal (will be processed after dispatch)
        for handler_id in to_remove:
            self._pending_removals.add(handler_id)

        # Increment processed count for successfully emitted events
        self._processed_count += 1

    finally:
        # Exit dispatch mode
        with self._dispatch_lock:
            self._dispatch_depth -= 1

            # Process deferred removals if we're back to top level
            if self._dispatch_depth == 0:
                self._process_pending_removals()

emit_sync

emit_sync(event: Event) -> None

Queue an event for later processing.

Use this when you need to emit from a sync context. Call process_pending() to handle queued events.

Source code in packages/maid-engine/src/maid_engine/core/events.py

def emit_sync(self, event: Event) -> None:
    """Queue an event for later processing.

    Use this when you need to emit from a sync context.
    Call process_pending() to handle queued events.
    """
    self._pending_events.put_nowait(event)

get_handler_ids_for_pack

get_handler_ids_for_pack(pack_name: str) -> set[UUID]

Get all handler IDs registered by a content pack.

Parameters:

Name	Type	Description	Default
pack_name	str	Name of the content pack	required

Returns:

Type	Description
set[UUID]	Set of handler UUIDs registered by the pack

Source code in packages/maid-engine/src/maid_engine/core/events.py

def get_handler_ids_for_pack(self, pack_name: str) -> set[UUID]:
    """Get all handler IDs registered by a content pack.

    Args:
        pack_name: Name of the content pack

    Returns:
        Set of handler UUIDs registered by the pack
    """
    return self._handlers_by_pack.get(pack_name, set()).copy()

get_handler_source

get_handler_source(handler_id: UUID) -> str | None

Get the source pack for a handler.

Parameters:

Name	Type	Description	Default
handler_id	UUID	The handler's UUID	required

Returns:

Type	Description
str | None	The pack name that registered this handler, or None if unknown

Source code in packages/maid-engine/src/maid_engine/core/events.py

def get_handler_source(self, handler_id: UUID) -> str | None:
    """Get the source pack for a handler.

    Args:
        handler_id: The handler's UUID

    Returns:
        The pack name that registered this handler, or None if unknown
    """
    return self._handler_sources.get(handler_id)

get_handlers_for_pack

get_handlers_for_pack(
    pack_name: str,
) -> dict[str, list[EventHandler]]

Get all handlers registered by a content pack.

Parameters:

Name	Type	Description	Default
pack_name	str	Name of the content pack	required

Returns:

Type	Description
dict[str, list[EventHandler]]	Dictionary mapping event type names (strings) to lists of handlers

Source code in packages/maid-engine/src/maid_engine/core/events.py

def get_handlers_for_pack(
    self, pack_name: str
) -> dict[str, list[EventHandler]]:
    """Get all handlers registered by a content pack.

    Args:
        pack_name: Name of the content pack

    Returns:
        Dictionary mapping event type names (strings) to lists of handlers
    """
    handler_ids = self._handlers_by_pack.get(pack_name, set())
    if not handler_ids:
        return {}

    result: dict[str, list[EventHandler]] = {}

    for type_name, handlers in self._handlers.items():
        pack_handlers = [
            reg.handler for reg in handlers if reg.handler_id in handler_ids
        ]
        if pack_handlers:
            result[type_name] = pack_handlers

    return result

handler_count

handler_count(event_type: type[Event] | None = None) -> int

Get the number of handlers, optionally filtered by event type.

Source code in packages/maid-engine/src/maid_engine/core/events.py

def handler_count(self, event_type: type[Event] | None = None) -> int:
    """Get the number of handlers, optionally filtered by event type."""
    if event_type:
        type_name = event_type.__name__
        return len(self._handlers.get(type_name, []))
    return sum(len(handlers) for handlers in self._handlers.values())

has_handlers

has_handlers(event_type: type[Event]) -> bool

Check if any handlers are registered for an event type.

Source code in packages/maid-engine/src/maid_engine/core/events.py

def has_handlers(self, event_type: type[Event]) -> bool:
    """Check if any handlers are registered for an event type."""
    type_name = event_type.__name__
    return bool(self._handlers.get(type_name))

is_dispatching

is_dispatching() -> bool

Check if currently dispatching events.

Returns:

Type	Description
bool	True if in the middle of an event dispatch

Source code in packages/maid-engine/src/maid_engine/core/events.py

def is_dispatching(self) -> bool:
    """Check if currently dispatching events.

    Returns:
        True if in the middle of an event dispatch
    """
    with self._dispatch_lock:
        return self._dispatch_depth > 0

process_pending async

process_pending() -> int

Process all pending events.

Returns:

Type	Description
int	Number of events processed

Source code in packages/maid-engine/src/maid_engine/core/events.py

async def process_pending(self) -> int:
    """Process all pending events.

    Returns:
        Number of events processed
    """
    count = 0
    while not self._pending_events.empty():
        event = await self._pending_events.get()
        await self.emit(event)
        count += 1
    return count

subscribe

subscribe(
    event_type: type[E],
    handler: EventHandler | SyncEventHandler,
    priority: EventPriority = NORMAL,
    once: bool = False,
    source_pack: str | None = None,
) -> UUID

Subscribe to an event type.

Parameters:

Name	Type	Description	Default
event_type	type[E]	The event class to subscribe to	required
handler	EventHandler | SyncEventHandler	Async or sync function to handle events	required
priority	EventPriority	Handler priority (higher priority runs first)	NORMAL
once	bool	If True, handler is removed after first invocation	False
source_pack	str | None	Name of the content pack registering this handler (for hot reload)	None

Returns:

Type	Description
UUID	Handler ID for unsubscribing

Source code in packages/maid-engine/src/maid_engine/core/events.py

def subscribe(
    self,
    event_type: type[E],
    handler: EventHandler | SyncEventHandler,
    priority: EventPriority = EventPriority.NORMAL,
    once: bool = False,
    source_pack: str | None = None,
) -> UUID:
    """Subscribe to an event type.

    Args:
        event_type: The event class to subscribe to
        handler: Async or sync function to handle events
        priority: Handler priority (higher priority runs first)
        once: If True, handler is removed after first invocation
        source_pack: Name of the content pack registering this handler (for hot reload)

    Returns:
        Handler ID for unsubscribing
    """
    type_name = event_type.__name__

    # Wrap sync handlers
    async_handler: EventHandler
    if asyncio.iscoroutinefunction(handler):
        async_handler = handler
    else:
        # Use default argument to capture handler by value, not by reference
        # This prevents closure bugs when registering handlers in loops
        # We've verified it's not a coroutine, so it must be a sync handler
        # Store in a typed variable for the closure
        captured_handler = handler

        async def _wrapper(
            event: Event,
            _handler: EventHandler | SyncEventHandler = captured_handler,
        ) -> None:
            # At runtime, we know this is a SyncEventHandler
            result = _handler(event)
            if asyncio.iscoroutine(result):
                await result

        async_handler = _wrapper

    registration = HandlerRegistration(
        handler_id=uuid4(),
        event_type=type_name,
        handler=async_handler,
        priority=priority,
        once=once,
        source_pack=source_pack,
    )

    handlers = self._handlers[type_name]
    handlers.append(registration)

    # Sort by priority (HIGHEST first)
    handlers.sort(key=lambda h: h.priority.value)

    # Track handler source for hot reload support
    if source_pack is not None:
        self._handler_sources[registration.handler_id] = source_pack
        self._handlers_by_pack[source_pack].add(registration.handler_id)

    return registration.handler_id

subscribe_all

subscribe_all(
    handler: EventHandler | SyncEventHandler,
    priority: EventPriority = NORMAL,
    once: bool = False,
    source_pack: str | None = None,
) -> UUID

Subscribe to ALL events (wildcard subscription).

This handler will be invoked for every event emitted, regardless of type. Useful for logging, metrics collection, or debugging.

Parameters:

Name	Type	Description	Default
handler	EventHandler | SyncEventHandler	Async or sync function to handle events	required
priority	EventPriority	Handler priority (higher priority runs first)	NORMAL
once	bool	If True, handler is removed after first invocation	False
source_pack	str | None	Name of the content pack registering this handler (for hot reload)	None

Returns:

Type	Description
UUID	Handler ID for unsubscribing

Source code in packages/maid-engine/src/maid_engine/core/events.py

def subscribe_all(
    self,
    handler: EventHandler | SyncEventHandler,
    priority: EventPriority = EventPriority.NORMAL,
    once: bool = False,
    source_pack: str | None = None,
) -> UUID:
    """Subscribe to ALL events (wildcard subscription).

    This handler will be invoked for every event emitted, regardless of type.
    Useful for logging, metrics collection, or debugging.

    Args:
        handler: Async or sync function to handle events
        priority: Handler priority (higher priority runs first)
        once: If True, handler is removed after first invocation
        source_pack: Name of the content pack registering this handler (for hot reload)

    Returns:
        Handler ID for unsubscribing
    """
    # Wrap sync handlers
    async_handler: EventHandler
    if asyncio.iscoroutinefunction(handler):
        async_handler = handler
    else:
        # Use default argument to capture handler by value, not by reference
        # This prevents closure bugs when registering handlers in loops
        # Store in a typed variable for the closure
        captured_handler = handler

        async def _wrapper(
            event: Event,
            _handler: EventHandler | SyncEventHandler = captured_handler,
        ) -> None:
            # At runtime, we know this is a SyncEventHandler
            result = _handler(event)
            if asyncio.iscoroutine(result):
                await result

        async_handler = _wrapper

    registration = HandlerRegistration(
        handler_id=uuid4(),
        event_type=self.WILDCARD_KEY,
        handler=async_handler,
        priority=priority,
        once=once,
        source_pack=source_pack,
    )

    handlers = self._handlers[self.WILDCARD_KEY]
    handlers.append(registration)

    # Sort by priority (HIGHEST first)
    handlers.sort(key=lambda h: h.priority.value)

    # Track handler source for hot reload support
    if source_pack is not None:
        self._handler_sources[registration.handler_id] = source_pack
        self._handlers_by_pack[source_pack].add(registration.handler_id)

    return registration.handler_id

unsubscribe

unsubscribe(handler_id: UUID) -> bool

Unsubscribe a handler by ID.

If called during event dispatch, the removal is deferred until the dispatch completes to ensure safe iteration.

Returns:

Type	Description
bool	True if handler was found and removed (or queued for removal)

Source code in packages/maid-engine/src/maid_engine/core/events.py

def unsubscribe(self, handler_id: UUID) -> bool:
    """Unsubscribe a handler by ID.

    If called during event dispatch, the removal is deferred until
    the dispatch completes to ensure safe iteration.

    Returns:
        True if handler was found and removed (or queued for removal)
    """
    with self._dispatch_lock:
        # If we're in a dispatch, defer the removal
        if self._dispatch_depth > 0:
            # Check if handler exists first
            for handlers in self._handlers.values():
                for reg in handlers:
                    if reg.handler_id == handler_id:
                        self._pending_removals.add(handler_id)
                        return True
            return False

        # Safe to remove immediately
        return self._do_unsubscribe(handler_id)

unsubscribe_all

unsubscribe_all(event_type: type[Event]) -> int

Unsubscribe all handlers for an event type.

Returns:

Type	Description
int	Number of handlers removed

Source code in packages/maid-engine/src/maid_engine/core/events.py

def unsubscribe_all(self, event_type: type[Event]) -> int:
    """Unsubscribe all handlers for an event type.

    Returns:
        Number of handlers removed
    """
    type_name = event_type.__name__
    handlers = self._handlers.get(type_name, [])
    count = len(handlers)

    # Clean up source tracking for all handlers being removed
    for reg in handlers:
        pack_name = self._handler_sources.pop(reg.handler_id, None)
        if pack_name and reg.handler_id in self._handlers_by_pack.get(pack_name, set()):
            self._handlers_by_pack[pack_name].discard(reg.handler_id)

    self._handlers.pop(type_name, None)
    return count

unsubscribe_pack

unsubscribe_pack(pack_name: str) -> int

Unsubscribe all handlers registered by a content pack.

If called during event dispatch, the removal is deferred until the dispatch completes to ensure safe iteration.

Parameters:

Name	Type	Description	Default
pack_name	str	Name of the content pack whose handlers should be removed	required

Returns:

Type	Description
int	Number of handlers removed (or queued for removal)

Source code in packages/maid-engine/src/maid_engine/core/events.py

def unsubscribe_pack(self, pack_name: str) -> int:
    """Unsubscribe all handlers registered by a content pack.

    If called during event dispatch, the removal is deferred until
    the dispatch completes to ensure safe iteration.

    Args:
        pack_name: Name of the content pack whose handlers should be removed

    Returns:
        Number of handlers removed (or queued for removal)
    """
    with self._dispatch_lock:
        # If we're in a dispatch, defer the removal
        if self._dispatch_depth > 0:
            handler_ids = self._handlers_by_pack.get(pack_name, set())
            if handler_ids:
                self._pending_pack_removals.add(pack_name)
                return len(handler_ids)
            return 0

        # Safe to remove immediately
        return self._do_unsubscribe_pack(pack_name)

GameEngine

GameEngine(
    settings: Settings | None = None,
    document_store: DocumentStore | None = None,
)

Main game engine that runs the tick loop.

The GameEngine manages the game world, networking, and the main tick loop that drives all game systems. It is content-agnostic - all game-specific functionality comes from content packs.

Example

engine = GameEngine()

Load content packs

engine.load_content_pack(my_content_pack)

await engine.start()

Engine runs until stopped

In another context:

await engine.stop()

Source code in packages/maid-engine/src/maid_engine/core/engine.py

def __init__(
    self,
    settings: Settings | None = None,
    document_store: DocumentStore | None = None,
) -> None:
    self._settings = settings or get_settings()
    self._world = World(self._settings)
    self._world._engine = self  # Set back-reference for systems to access engine
    self._server: MAIDServer | None = None
    self._state = EngineState.STOPPED
    self._tick_task: asyncio.Task[None] | None = None
    self._stop_event = asyncio.Event()
    self._shutdown_callbacks: list[Callable[[], None]] = []

    # Command registry for layered command resolution
    # Pass the world's event bus so PlayerCommandEvent can be emitted
    self._command_registry = LayeredCommandRegistry(event_bus=self._world._events)

    # Document store for persistence (default to in-memory for testing)
    self._document_store = document_store or InMemoryDocumentStore()

    # Content packs
    self._content_packs: dict[str, ContentPack] = {}
    self._content_pack_order: list[str] = []

    # Timing
    self._tick_rate = self._settings.game.tick_rate
    self._tick_interval = 1.0 / self._tick_rate
    self._last_tick_time = 0.0
    self._tick_count = 0

    # Statistics
    self._start_time = 0.0
    self._total_tick_time = 0.0
    self._max_tick_time = 0.0
    self._tick_overruns = 0

    # Profiling (opt-in)
    self._tick_collector: TickCollector | None = None

    # Bridge manager for external service integration
    self._bridge_manager = BridgeManager()
    self._rss_feed_manager: RSSFeedManager | None = None

    # Hot reload support
    self._hot_reload_manager: HotReloadManager | None = None
    self._hot_reload_pause = asyncio.Event()

    # Initialize LLM provider registry from settings
    self._provider_registry = create_registry_from_settings(self._settings)
    # Also set as the global registry so other code can access it
    set_registry(self._provider_registry)

    # Initialize bridges from settings
    self._setup_bridges()

average_tick_time property

average_tick_time: float

Get average tick processing time.

bridge_manager property

bridge_manager: BridgeManager

Get the bridge manager for external service integration.

characters property

characters: Any

Get character manager.

Note: This is a placeholder. Content packs should attach their managers to the engine or world.

command_registry property

command_registry: LayeredCommandRegistry

Get the command registry.

content_packs property

content_packs: dict[str, ContentPack]

Get loaded content packs.

document_store property

document_store: DocumentStore

Get the document store.

hot_reload property

hot_reload: HotReloadManager

Get the hot reload manager (lazy-initialized).

The hot reload manager is created on first access. This allows hot reload operations to be performed on the engine at runtime, including loading, unloading, and reloading content packs while the engine is running.

Returns:

Type	Description
HotReloadManager	The HotReloadManager instance for this engine.

is_running property

is_running: bool

Check if engine is running.

items property

items: Any

Get item manager.

provider_registry property

provider_registry: LLMProviderRegistry

Get the LLM provider registry.

The provider registry manages AI/LLM providers (Anthropic, OpenAI, Ollama, etc.) for use by AI-powered features like NPC dialogue. Providers are initialized from settings during engine construction.

Returns:

Type	Description
LLMProviderRegistry	The LLMProviderRegistry instance for this engine.

quest_manager property

quest_manager: Any

Get quest manager.

Note: This is set by content packs that provide quest functionality.

reward_distributor property

reward_distributor: Any

Get reward distributor.

Note: This is set by content packs that provide quest functionality.

rooms property

rooms: Any

Get room manager.

rss_feed_manager property

rss_feed_manager: RSSFeedManager | None

Get the RSS feed manager (if enabled).

server property

server: MAIDServer | None

Get the network server.

settings property

settings: Settings

Get engine settings.

state property

state: EngineState

Get current engine state.

tick_collector property

tick_collector: TickCollector | None

Get the tick collector for profiling (if enabled).

tick_count property

tick_count: int

Get total tick count.

uptime property

uptime: float

Get engine uptime in seconds.

world property

world: World

Get the game world.

add_shutdown_callback

add_shutdown_callback(callback: Callable[[], None]) -> None

Add a callback to run on shutdown.

Source code in packages/maid-engine/src/maid_engine/core/engine.py

def add_shutdown_callback(self, callback: Callable[[], None]) -> None:
    """Add a callback to run on shutdown."""
    self._shutdown_callbacks.append(callback)

get_stats

get_stats() -> dict[str, float | int]

Get engine statistics.

Source code in packages/maid-engine/src/maid_engine/core/engine.py

def get_stats(self) -> dict[str, float | int]:
    """Get engine statistics."""
    return {
        "uptime": self.uptime,
        "tick_count": self._tick_count,
        "tick_rate": self._tick_rate,
        "average_tick_time": self.average_tick_time,
        "max_tick_time": self._max_tick_time,
        "tick_overruns": self._tick_overruns,
        "entity_count": len(self._world.entities),
        "room_count": self._world.room_count(),
        "content_packs": len(self._content_packs),
    }

load_content_pack

load_content_pack(
    pack: ContentPack, use_hot_reload: bool = False
) -> None

Load a content pack.

When the engine is stopped, content packs are loaded synchronously. When the engine is running, set use_hot_reload=True to use the hot reload manager for safe runtime loading.

Parameters:

Name	Type	Description	Default
pack	ContentPack	The content pack to load.	required
use_hot_reload	bool	If True and engine is running, use hot reload. If False and engine is running, raises RuntimeError.	False

Raises:

Type	Description
RuntimeError	If engine is running and use_hot_reload is False.
ValueError	If pack is already loaded or dependencies are missing.

Source code in packages/maid-engine/src/maid_engine/core/engine.py

def load_content_pack(self, pack: "ContentPack", use_hot_reload: bool = False) -> None:
    """Load a content pack.

    When the engine is stopped, content packs are loaded synchronously.
    When the engine is running, set use_hot_reload=True to use the
    hot reload manager for safe runtime loading.

    Args:
        pack: The content pack to load.
        use_hot_reload: If True and engine is running, use hot reload.
                       If False and engine is running, raises RuntimeError.

    Raises:
        RuntimeError: If engine is running and use_hot_reload is False.
        ValueError: If pack is already loaded or dependencies are missing.
    """
    if self._state != EngineState.STOPPED:
        if use_hot_reload:
            # Delegate to hot reload for runtime loading
            # Note: This is a synchronous method, but hot reload is async.
            # The caller should use hot_reload.load_pack() directly for
            # async operation with full result feedback.
            raise RuntimeError(
                "Cannot use synchronous load_content_pack() with use_hot_reload=True. "
                "Use engine.hot_reload.load_pack() for runtime loading."
            )
        raise RuntimeError("Cannot load content packs while engine is running")

    pack_name = pack.manifest.name
    if pack_name in self._content_packs:
        raise ValueError(f"Content pack '{pack_name}' is already loaded")

    # Verify dependencies are loaded
    for dep in pack.get_dependencies():
        if dep not in self._content_packs:
            raise ValueError(
                f"Content pack '{pack_name}' depends on '{dep}' which is not loaded"
            )

    self._content_packs[pack_name] = pack
    self._content_pack_order.append(pack_name)

relay_to_external async

relay_to_external(
    game_channel: str, sender_name: str, message: str
) -> None

Relay a game message to mapped external channels.

This is a convenience method that wraps bridge_manager.relay_to_external(). Content packs can call this to send game channel messages to external services like Discord and IRC.

Parameters:

Name	Type	Description	Default
game_channel	str	The game channel name (e.g., "ooc", "chat", "newbie")	required
sender_name	str	The name of the sender in the game	required
message	str	The message content	required

Source code in packages/maid-engine/src/maid_engine/core/engine.py

async def relay_to_external(
    self, game_channel: str, sender_name: str, message: str
) -> None:
    """Relay a game message to mapped external channels.

    This is a convenience method that wraps bridge_manager.relay_to_external().
    Content packs can call this to send game channel messages to external
    services like Discord and IRC.

    Args:
        game_channel: The game channel name (e.g., "ooc", "chat", "newbie")
        sender_name: The name of the sender in the game
        message: The message content
    """
    await self._bridge_manager.relay_to_external(game_channel, sender_name, message)

run async

run() -> None

Start and run until stopped.

This is a convenience method that starts the engine and waits for it to be stopped.

Source code in packages/maid-engine/src/maid_engine/core/engine.py

async def run(self) -> None:
    """Start and run until stopped.

    This is a convenience method that starts the engine and
    waits for it to be stopped.
    """
    await self.start()
    await self._stop_event.wait()
    await self.stop()

set_server

set_server(server: MAIDServer) -> None

Set the network server.

Source code in packages/maid-engine/src/maid_engine/core/engine.py

def set_server(self, server: "MAIDServer") -> None:
    """Set the network server."""
    self._server = server

set_tick_collector

set_tick_collector(collector: TickCollector | None) -> None

Set the tick collector for profiling.

The tick collector must be started externally (e.g., via ProfileManager). When set, the engine will record tick profiling data during each tick.

Parameters:

Name	Type	Description	Default
collector	TickCollector | None	The TickCollector instance, or None to disable profiling.	required

Source code in packages/maid-engine/src/maid_engine/core/engine.py

def set_tick_collector(self, collector: "TickCollector | None") -> None:
    """Set the tick collector for profiling.

    The tick collector must be started externally (e.g., via ProfileManager).
    When set, the engine will record tick profiling data during each tick.

    Args:
        collector: The TickCollector instance, or None to disable profiling.
    """
    self._tick_collector = collector

start async

start() -> None

Start the game engine.

Source code in packages/maid-engine/src/maid_engine/core/engine.py

async def start(self) -> None:
    """Start the game engine."""
    if self._state != EngineState.STOPPED:
        return

    self._state = EngineState.STARTING
    self._stop_event.clear()

    # Set up signal handlers
    self._setup_signals()

    # Initialize document store
    await self._document_store.initialize()

    # Initialize content packs (in dependency order)
    # Set pack priorities based on load order (later = higher priority)
    for priority, pack_name in enumerate(self._content_pack_order):
        self._command_registry.set_pack_priority(pack_name, priority * 10)

    for pack_name in self._content_pack_order:
        pack = self._content_packs[pack_name]

        # Register document schemas from content pack
        pack.register_document_schemas(self._document_store)

        # Register events from content pack
        # (Events are used for documentation/discovery, emitting is dynamic)
        _events = pack.get_events()  # noqa: F841 - events for future use

        # Register commands from content pack
        pack.register_commands(self._command_registry)

        # Register systems from content pack
        for system in pack.get_systems(self._world):
            self._world.systems.register(system)

            # Auto-inject storage into systems that implement StorageAware
            if isinstance(system, StorageAware):
                system.set_storage(self._document_store)

        # Call pack's on_load hook
        await pack.on_load(self)

    # Register world persistence schemas and load saved state
    # This is done here after content packs have loaded so that
    # grid/wilderness managers are set up on the world
    register_world_persistence_schemas(self._document_store)
    loaded = await load_world_systems_state(
        self._document_store,
        self._world,
        world_id="default",
    )
    if loaded:
        _logger.info("Loaded world systems state from storage")
    else:
        _logger.debug("No saved world systems state found")

    # Start the world
    await self._world.startup()

    # Start the network server if configured
    if self._server:
        await self._server.start()

        # Register RSS router with web server if enabled
        if self._rss_feed_manager and self._settings.bridges.rss.enabled:
            self._register_rss_router()

        # Wire up events WebSocket to game events
        from maid_engine.api.v1.events_ws import setup_event_hooks

        setup_event_hooks(self)
        _logger.info("Events WebSocket hooks configured")

        # Wire up admin WebSocket to game events
        from maid_engine.api.admin.websocket import setup_admin_event_hooks

        setup_admin_event_hooks(self)
        _logger.info("Admin WebSocket hooks configured")

    # Start external bridges
    await self._bridge_manager.start_all()

    # Wire up bridge message handler for external -> game messages
    self._setup_bridge_message_handler()

    # Start the tick loop
    self._start_time = time.monotonic()
    self._last_tick_time = time.monotonic()
    self._state = EngineState.RUNNING
    self._tick_task = asyncio.create_task(self._tick_loop())

stop async

stop() -> None

Stop the game engine.

Source code in packages/maid-engine/src/maid_engine/core/engine.py

async def stop(self) -> None:
    """Stop the game engine."""
    if self._state not in (EngineState.RUNNING, EngineState.STARTING):
        return

    self._state = EngineState.STOPPING
    self._stop_event.set()

    # Wait for tick loop to finish
    if self._tick_task:
        try:
            await asyncio.wait_for(self._tick_task, timeout=5.0)
        except TimeoutError:
            self._tick_task.cancel()
            with contextlib.suppress(asyncio.CancelledError):
                await self._tick_task

    # Stop external bridges
    await self._bridge_manager.stop_all()

    # Stop the network server
    if self._server:
        await self._server.stop()

    # Save world systems state before shutdown
    # This preserves grid and wilderness state across server restarts
    try:
        await save_world_systems_state(
            self._document_store,
            self._world,
            world_id="default",
        )
        _logger.info("Saved world systems state to storage")
    except Exception as e:
        _logger.error("Failed to save world systems state: %s", e)

    # Shut down the world
    await self._world.shutdown()

    # Unload content packs (in reverse order)
    for pack_name in reversed(self._content_pack_order):
        pack = self._content_packs[pack_name]
        # Unregister commands from this pack
        self._command_registry.unregister_pack(pack_name)
        await pack.on_unload(self)

    # Close document store
    await self._document_store.close()

    # Run shutdown callbacks
    for callback in self._shutdown_callbacks:
        try:
            callback()
        except Exception as e:
            # Log but don't let callbacks prevent shutdown
            _logger.warning("Shutdown callback %s raised exception: %s", callback, e)

    self._state = EngineState.STOPPED

unload_content_pack

unload_content_pack(pack_name: str) -> bool

Unload a content pack.

Can only be done when engine is stopped. Returns True if pack was unloaded.

Source code in packages/maid-engine/src/maid_engine/core/engine.py

def unload_content_pack(self, pack_name: str) -> bool:
    """Unload a content pack.

    Can only be done when engine is stopped.
    Returns True if pack was unloaded.
    """
    if self._state != EngineState.STOPPED:
        raise RuntimeError("Cannot unload content packs while engine is running")

    if pack_name not in self._content_packs:
        return False

    # Check if other packs depend on this one
    for other_name, other_pack in self._content_packs.items():
        if other_name != pack_name and pack_name in other_pack.get_dependencies():
            raise ValueError(
                f"Cannot unload '{pack_name}': "
                f"'{other_name}' depends on it"
            )

    del self._content_packs[pack_name]
    self._content_pack_order.remove(pack_name)
    return True

InMemoryDocumentStore

InMemoryDocumentStore()

Bases: DocumentStore

In-memory document store for testing.

Uses dictionaries. Not suitable for production.

Source code in packages/maid-engine/src/maid_engine/storage/document_store.py

def __init__(self) -> None:
    self._schemas: dict[str, type[BaseModel]] = {}
    self._collections: dict[str, InMemoryDocumentCollection[Any]] = {}
    self._query_collector: QueryCollector | None = None

LayeredCommandRegistry

LayeredCommandRegistry(event_bus: 'EventBus | None' = None)

Command registry with layered resolution for content packs.

Commands can be registered by multiple content packs. When a command is executed, the highest priority registration is used.

The registry now integrates with: - Hook system: Pre/post command hooks for logging, rate limiting, etc. - Lock system: Permission checks using lock expressions

Example

registry = LayeredCommandRegistry()

Set pack priorities (higher = wins)

registry.set_pack_priority("my-game", 100) registry.set_pack_priority("stdlib", 50) registry.set_pack_priority("engine", 0)

Register commands from different packs

registry.register("look", look_handler, pack_name="stdlib") registry.register("look", custom_look_handler, pack_name="my-game")

When "look" is executed, my-game's handler is used (priority 100)

Register hooks

registry.register_pre_hook("rate_limiter", check_rate_limit) registry.register_post_hook("logger", log_command)

Register custom lock functions

registry.register_lock_function("is_owner", check_ownership)

Source code in packages/maid-engine/src/maid_engine/commands/registry.py

def __init__(self, event_bus: "EventBus | None" = None) -> None:
    # command_name -> [(priority, pack_name, definition)]
    self._layers: dict[str, list[tuple[int, str, CommandDefinition]]] = defaultdict(list)
    self._pack_priorities: dict[str, int] = {}
    self._aliases: dict[str, str] = {}  # alias -> primary name

    # Event bus for emitting PlayerCommandEvent
    self._event_bus = event_bus

    # Hook system
    self._hook_registry = HookRegistry()
    self._command_executor = CommandExecutor(self._hook_registry)

    # Lock system
    self._lock_evaluator = LockEvaluator()

event_bus property writable

event_bus: 'EventBus | None'

Get the event bus for PlayerCommandEvent emission.

hook_registry property

hook_registry: HookRegistry

Get the hook registry for advanced hook management.

lock_evaluator property

lock_evaluator: LockEvaluator

Get the lock evaluator for advanced lock management.

all_commands

all_commands() -> list[CommandDefinition]

Get highest-priority definitions for all commands.

Source code in packages/maid-engine/src/maid_engine/commands/registry.py

def all_commands(self) -> list[CommandDefinition]:
    """Get highest-priority definitions for all commands."""
    commands = []
    for name in self._layers:
        defn = self.get(name)
        if defn:
            commands.append(defn)
    return commands

commands_by_category

commands_by_category() -> dict[
    str, list[CommandDefinition]
]

Get commands grouped by category.

Source code in packages/maid-engine/src/maid_engine/commands/registry.py

def commands_by_category(self) -> dict[str, list[CommandDefinition]]:
    """Get commands grouped by category."""
    by_category: dict[str, list[CommandDefinition]] = defaultdict(list)
    for cmd in self.all_commands():
        if not cmd.hidden:
            by_category[cmd.category].append(cmd)
    return dict(by_category)

execute async

execute(
    context: CommandContext,
    player_access_level: AccessLevel = PLAYER,
) -> bool

Execute a command using the highest-priority handler with lock checking and hook pipeline.

Execution flow: 1. Find the command definition 2. Check lock expression (if defined) 3. Check access level 4. Run pre-hooks (can cancel execution) 5. Execute command handler 6. Run post-hooks (always runs, even on exception)

Parameters:

Name	Type	Description	Default
context	CommandContext	Command context containing session, player_id, command, args, etc.	required
player_access_level	AccessLevel	The access level of the player executing	PLAYER

Returns:

Type	Description
bool	True if command was handled successfully, False if command not found
bool	or execution was cancelled

Raises:

Type	Description
PermissionError	If player lacks required access level

Source code in packages/maid-engine/src/maid_engine/commands/registry.py

async def execute(
    self,
    context: CommandContext,
    player_access_level: AccessLevel = AccessLevel.PLAYER,
) -> bool:
    """Execute a command using the highest-priority handler with lock checking and hook pipeline.

    Execution flow:
    1. Find the command definition
    2. Check lock expression (if defined)
    3. Check access level
    4. Run pre-hooks (can cancel execution)
    5. Execute command handler
    6. Run post-hooks (always runs, even on exception)

    Args:
        context: Command context containing session, player_id, command, args, etc.
        player_access_level: The access level of the player executing

    Returns:
        True if command was handled successfully, False if command not found
        or execution was cancelled

    Raises:
        PermissionError: If player lacks required access level
    """
    definition = self.get(context.command)
    if not definition:
        return False

    # Check locks BEFORE hooks and access level
    if definition.locks:
        # Determine target for lock evaluation
        # Use explicit target if provided, otherwise try to resolve from first argument
        target = context.target
        if target is None and context.args:
            target = self._resolve_target_for_lock(
                context.args[0], context.world, context.player_id
            )

        lock_ctx = LockContext(
            player_entity_id=context.player_id,
            world=context.world,
            command_name=context.command,
            session=context.session,
            target=target,
        )
        if not self._lock_evaluator.check(definition.locks, lock_ctx):
            await self._send(context.session, "You don't have permission to do that.")
            return False

    # Check access level
    if player_access_level < definition.access_level:
        raise PermissionError(
            f"Command '{definition.name}' requires {definition.access_level.name} "
            f"access (you have {player_access_level.name})"
        )

    # Create handler wrapper that calls the actual command
    # Returns True if handler returns None (implicit success)
    async def handler_wrapper() -> bool:
        result = await definition.handler(context)
        # Convert None to True (handler succeeded but didn't return a value)
        return result if result is not None else True

    # Execute with hook pipeline
    # Note: CommandExecutor returns None when pre-hooks cancel execution
    success = False
    try:
        result = await self._command_executor.execute(
            handler=handler_wrapper,
            command_name=definition.name,
            category=definition.category,
            args=context.args,
            raw_input=context.raw_input,
            session=context.session,
            player_entity_id=context.player_id,
            world=context.world,
            metadata=definition.metadata,
        )
        # If result is None, pre-hooks cancelled execution
        # If result is True/False, that's the handler result
        if result is None:
            success = False  # Cancelled by pre-hook
        else:
            success = bool(result)
        return success
    except Exception as e:
        # Log error, send message
        await self._send(context.session, f"Error executing command: {e}")
        success = False
        return False
    finally:
        # Emit PlayerCommandEvent after execution (success or failure)
        # Import locally to avoid circular imports
        if self._event_bus:
            from maid_engine.core.events import PlayerCommandEvent

            event = PlayerCommandEvent(
                player_id=context.player_id,
                command=definition.name,
                args=context.args,
                success=success,
                raw_input=context.raw_input,
            )
            await self._event_bus.emit(event)

get

get(name: str) -> CommandDefinition | None

Get the highest-priority command definition.

Source code in packages/maid-engine/src/maid_engine/commands/registry.py

def get(self, name: str) -> CommandDefinition | None:
    """Get the highest-priority command definition."""
    # Check if it's an alias
    if name in self._aliases:
        name = self._aliases[name]

    layers = self._layers.get(name, [])
    if not layers:
        return None

    # Return highest priority
    return layers[0][2]

get_all_layers

get_all_layers(name: str) -> list[CommandDefinition]

Get all definitions for a command across all packs.

Source code in packages/maid-engine/src/maid_engine/commands/registry.py

def get_all_layers(self, name: str) -> list[CommandDefinition]:
    """Get all definitions for a command across all packs."""
    if name in self._aliases:
        name = self._aliases[name]

    return [layer[2] for layer in self._layers.get(name, [])]

get_commands_for_pack

get_commands_for_pack(pack_name: str) -> list[str]

Get all command names registered by a content pack.

This is useful for hot reload inspection to see what commands a pack has registered.

Parameters:

Name	Type	Description	Default
pack_name	str	Name of the content pack	required

Returns:

Type	Description
list[str]	List of command names registered by the pack

Source code in packages/maid-engine/src/maid_engine/commands/registry.py

def get_commands_for_pack(self, pack_name: str) -> list[str]:
    """Get all command names registered by a content pack.

    This is useful for hot reload inspection to see what commands
    a pack has registered.

    Args:
        pack_name: Name of the content pack

    Returns:
        List of command names registered by the pack
    """
    commands: list[str] = []
    for name, layers in self._layers.items():
        for _, pname, _ in layers:
            if pname == pack_name:
                commands.append(name)
                break  # Only count each command once per pack
    return commands

get_pack_command_definitions

get_pack_command_definitions(
    pack_name: str,
) -> list[CommandDefinition]

Get all command definitions registered by a content pack.

Parameters:

Name	Type	Description	Default
pack_name	str	Name of the content pack	required

Returns:

Type	Description
list[CommandDefinition]	List of CommandDefinition objects registered by the pack

Source code in packages/maid-engine/src/maid_engine/commands/registry.py

def get_pack_command_definitions(
    self, pack_name: str
) -> list[CommandDefinition]:
    """Get all command definitions registered by a content pack.

    Args:
        pack_name: Name of the content pack

    Returns:
        List of CommandDefinition objects registered by the pack
    """
    definitions: list[CommandDefinition] = []
    for layers in self._layers.values():
        for _, pname, definition in layers:
            if pname == pack_name:
                definitions.append(definition)
    return definitions

get_pack_priority

get_pack_priority(pack_name: str) -> int

Get the priority for a content pack.

Source code in packages/maid-engine/src/maid_engine/commands/registry.py

def get_pack_priority(self, pack_name: str) -> int:
    """Get the priority for a content pack."""
    return self._pack_priorities.get(pack_name, 0)

register

register(
    name: str,
    handler: CommandHandler,
    pack_name: str,
    *,
    priority: int | None = None,
    aliases: list[str] | None = None,
    category: str = "general",
    description: str = "",
    usage: str = "",
    access_level: AccessLevel = PLAYER,
    hidden: bool = False,
    locks: str = "",
    metadata: dict[str, Any] | None = None,
) -> None

Register a command from a content pack.

Parameters:

Name	Type	Description	Default
name	str	Primary command name	required
handler	CommandHandler	Async function to handle the command	required
pack_name	str	Name of the content pack registering this command	required
priority	int | None	Priority for layered resolution (defaults to pack priority)	None
aliases	list[str] | None	Alternative names for this command	None
category	str	Command category for help	'general'
description	str	Short description	''
usage	str	Usage string	''
access_level	AccessLevel	Minimum access level required	PLAYER
hidden	bool	Whether to hide from help listings	False
locks	str	Lock expression for permission checks (e.g., "perm(admin) OR owns()")	''
metadata	dict[str, Any] | None	Additional metadata for hooks (e.g., cooldowns, combat state)	None

Source code in packages/maid-engine/src/maid_engine/commands/registry.py

def register(
    self,
    name: str,
    handler: CommandHandler,
    pack_name: str,
    *,
    priority: int | None = None,
    aliases: list[str] | None = None,
    category: str = "general",
    description: str = "",
    usage: str = "",
    access_level: AccessLevel = AccessLevel.PLAYER,
    hidden: bool = False,
    locks: str = "",
    metadata: dict[str, Any] | None = None,
) -> None:
    """Register a command from a content pack.

    Args:
        name: Primary command name
        handler: Async function to handle the command
        pack_name: Name of the content pack registering this command
        priority: Priority for layered resolution (defaults to pack priority)
        aliases: Alternative names for this command
        category: Command category for help
        description: Short description
        usage: Usage string
        access_level: Minimum access level required
        hidden: Whether to hide from help listings
        locks: Lock expression for permission checks (e.g., "perm(admin) OR owns()")
        metadata: Additional metadata for hooks (e.g., cooldowns, combat state)
    """
    # Use pack priority if not explicitly specified
    if priority is None:
        priority = self.get_pack_priority(pack_name)

    definition = CommandDefinition(
        name=name,
        handler=handler,
        pack_name=pack_name,
        priority=priority,
        aliases=aliases or [],
        category=category,
        description=description,
        usage=usage,
        access_level=access_level,
        hidden=hidden,
        locks=locks,
        metadata=metadata or {},
    )

    # Add to layers
    layers = self._layers[name]
    layers.append((priority, pack_name, definition))
    # Keep sorted by priority (highest first)
    layers.sort(key=lambda x: x[0], reverse=True)

    # Register aliases
    for alias in definition.aliases:
        self._aliases[alias] = name

register_lock_function

register_lock_function(
    name: str,
    func: Callable[[LockContext, list[str]], bool],
) -> None

Register a custom lock function.

Lock functions are named predicates that can be used in lock expressions. Each function receives a LockContext and a list of string arguments, and returns a boolean.

Parameters:

Name	Type	Description	Default
name	str	The function name used in expressions (e.g., "is_owner")	required
func	Callable[[LockContext, list[str]], bool]	The function to call, receives (context, args) and returns bool	required

Example

def check_gold(ctx: LockContext, args: list[str]) -> bool: player = ctx.get_player() if not player: return False required = int(args[0]) if args else 0 return player.gold >= required

registry.register_lock_function("has_gold", check_gold)

Now you can use "has_gold(100)" in lock expressions

Source code in packages/maid-engine/src/maid_engine/commands/registry.py

def register_lock_function(
    self, name: str, func: Callable[[LockContext, list[str]], bool]
) -> None:
    """Register a custom lock function.

    Lock functions are named predicates that can be used in lock expressions.
    Each function receives a LockContext and a list of string arguments,
    and returns a boolean.

    Args:
        name: The function name used in expressions (e.g., "is_owner")
        func: The function to call, receives (context, args) and returns bool

    Example:
        def check_gold(ctx: LockContext, args: list[str]) -> bool:
            player = ctx.get_player()
            if not player:
                return False
            required = int(args[0]) if args else 0
            return player.gold >= required

        registry.register_lock_function("has_gold", check_gold)
        # Now you can use "has_gold(100)" in lock expressions
    """
    self._lock_evaluator.register_function(name, func)

register_post_hook

register_post_hook(
    name: str,
    handler: Callable[..., Any],
    priority: HookPriority = NORMAL,
    commands: list[str] | None = None,
    categories: list[str] | None = None,
) -> None

Register a post-command hook.

Post-hooks run after the command handler completes.

Parameters:

Name	Type	Description	Default
name	str	Unique identifier for the hook	required
handler	Callable[..., Any]	Async or sync function accepting PostHookContext	required
priority	HookPriority	Execution priority (lower runs first)	NORMAL
commands	list[str] | None	Filter to specific command names (None = all)	None
categories	list[str] | None	Filter to specific categories (None = all)	None

Source code in packages/maid-engine/src/maid_engine/commands/registry.py

def register_post_hook(
    self,
    name: str,
    handler: Callable[..., Any],
    priority: HookPriority = HookPriority.NORMAL,
    commands: list[str] | None = None,
    categories: list[str] | None = None,
) -> None:
    """Register a post-command hook.

    Post-hooks run after the command handler completes.

    Args:
        name: Unique identifier for the hook
        handler: Async or sync function accepting PostHookContext
        priority: Execution priority (lower runs first)
        commands: Filter to specific command names (None = all)
        categories: Filter to specific categories (None = all)
    """
    self._hook_registry.register_post_hook(name, handler, priority, commands, categories)

register_pre_hook

register_pre_hook(
    name: str,
    handler: Callable[..., Any],
    priority: HookPriority = NORMAL,
    commands: list[str] | None = None,
    categories: list[str] | None = None,
) -> None

Register a pre-command hook.

Pre-hooks run before the command handler and can cancel execution.

Parameters:

Name	Type	Description	Default
name	str	Unique identifier for the hook	required
handler	Callable[..., Any]	Async or sync function accepting PreHookContext	required
priority	HookPriority	Execution priority (lower runs first)	NORMAL
commands	list[str] | None	Filter to specific command names (None = all)	None
categories	list[str] | None	Filter to specific categories (None = all)	None

Source code in packages/maid-engine/src/maid_engine/commands/registry.py

def register_pre_hook(
    self,
    name: str,
    handler: Callable[..., Any],
    priority: HookPriority = HookPriority.NORMAL,
    commands: list[str] | None = None,
    categories: list[str] | None = None,
) -> None:
    """Register a pre-command hook.

    Pre-hooks run before the command handler and can cancel execution.

    Args:
        name: Unique identifier for the hook
        handler: Async or sync function accepting PreHookContext
        priority: Execution priority (lower runs first)
        commands: Filter to specific command names (None = all)
        categories: Filter to specific categories (None = all)
    """
    self._hook_registry.register_pre_hook(name, handler, priority, commands, categories)

set_pack_priority

set_pack_priority(pack_name: str, priority: int) -> None

Set the priority for a content pack.

Higher priority packs override lower priority ones.

Source code in packages/maid-engine/src/maid_engine/commands/registry.py

def set_pack_priority(self, pack_name: str, priority: int) -> None:
    """Set the priority for a content pack.

    Higher priority packs override lower priority ones.
    """
    self._pack_priorities[pack_name] = priority

unregister

unregister(name: str, pack_name: str) -> bool

Unregister a command from a specific pack.

Source code in packages/maid-engine/src/maid_engine/commands/registry.py

def unregister(self, name: str, pack_name: str) -> bool:
    """Unregister a command from a specific pack."""
    if name not in self._layers:
        return False

    layers = self._layers[name]
    original_len = len(layers)

    # Collect aliases that the removed pack defined
    removed_aliases: set[str] = set()
    for _, pname, definition in layers:
        if pname == pack_name:
            removed_aliases.update(definition.aliases)

    # Remove entries for this pack
    self._layers[name] = [
        layer for layer in layers if layer[1] != pack_name
    ]

    # Collect aliases still defined by remaining layers
    remaining_aliases: set[str] = set()
    for _, _, definition in self._layers.get(name, []):
        remaining_aliases.update(definition.aliases)

    # Only delete aliases that are no longer defined by any layer
    for alias in removed_aliases:
        if alias not in remaining_aliases and self._aliases.get(alias) == name:
            del self._aliases[alias]

    if len(self._layers[name]) == 0:
        del self._layers[name]

    return len(self._layers.get(name, [])) < original_len

unregister_hook

unregister_hook(name: str) -> bool

Unregister a hook by name.

Parameters:

Name	Type	Description	Default
name	str	The unique identifier of the hook to remove	required

Returns:

Type	Description
bool	True if a hook was found and removed, False otherwise

Source code in packages/maid-engine/src/maid_engine/commands/registry.py

def unregister_hook(self, name: str) -> bool:
    """Unregister a hook by name.

    Args:
        name: The unique identifier of the hook to remove

    Returns:
        True if a hook was found and removed, False otherwise
    """
    return self._hook_registry.unregister(name)

unregister_pack

unregister_pack(pack_name: str) -> int

Unregister all commands from a pack.

Returns:

Type	Description
int	Number of commands unregistered

Source code in packages/maid-engine/src/maid_engine/commands/registry.py

def unregister_pack(self, pack_name: str) -> int:
    """Unregister all commands from a pack.

    Returns:
        Number of commands unregistered
    """
    count = 0
    for name in list(self._layers.keys()):
        if self.unregister(name, pack_name):
            count += 1
    return count

QueryOptions dataclass

QueryOptions(
    filters: dict[str, Any] = dict(),
    order_by: str | None = None,
    order: SortOrder = ASC,
    limit: int | None = None,
    offset: int = 0,
)

Options for querying documents.

Attributes:

Name	Type	Description
filters	dict[str, Any]	Dict of field_name -> value for exact matches
order_by	str | None	Field name to sort by
order	SortOrder	Sort direction
limit	int | None	Maximum number of results
offset	int	Number of results to skip

RoomEnterEvent dataclass

RoomEnterEvent(
    entity_id: UUID,
    room_id: UUID,
    from_room_id: UUID | None = None,
)

Bases: Event

Emitted when an entity enters a room.

RoomIndex

RoomIndex()

Index for efficient room-based entity lookups.

Source code in packages/maid-engine/src/maid_engine/core/world.py

def __init__(self) -> None:
    self._entities_by_room: dict[UUID, set[UUID]] = defaultdict(set)
    self._room_by_entity: dict[UUID, UUID] = {}

add

add(entity_id: UUID, room_id: UUID) -> None

Add entity to room index.

Source code in packages/maid-engine/src/maid_engine/core/world.py

def add(self, entity_id: UUID, room_id: UUID) -> None:
    """Add entity to room index."""
    # Remove from previous room if present
    if entity_id in self._room_by_entity:
        old_room = self._room_by_entity[entity_id]
        self._entities_by_room[old_room].discard(entity_id)

    self._entities_by_room[room_id].add(entity_id)
    self._room_by_entity[entity_id] = room_id

clear

clear() -> None

Clear all indexes.

Source code in packages/maid-engine/src/maid_engine/core/world.py

def clear(self) -> None:
    """Clear all indexes."""
    self._entities_by_room.clear()
    self._room_by_entity.clear()

count_in_room

count_in_room(room_id: UUID) -> int

Get count of entities in a room.

Source code in packages/maid-engine/src/maid_engine/core/world.py

def count_in_room(self, room_id: UUID) -> int:
    """Get count of entities in a room."""
    return len(self._entities_by_room.get(room_id, set()))

get_entities

get_entities(room_id: UUID) -> set[UUID]

Get all entities in a room.

Source code in packages/maid-engine/src/maid_engine/core/world.py

def get_entities(self, room_id: UUID) -> set[UUID]:
    """Get all entities in a room."""
    return self._entities_by_room.get(room_id, set()).copy()

get_room

get_room(entity_id: UUID) -> UUID | None

Get the room an entity is in.

Source code in packages/maid-engine/src/maid_engine/core/world.py

def get_room(self, entity_id: UUID) -> UUID | None:
    """Get the room an entity is in."""
    return self._room_by_entity.get(entity_id)

move

move(entity_id: UUID, new_room_id: UUID) -> UUID | None

Move entity to new room. Returns previous room ID.

Source code in packages/maid-engine/src/maid_engine/core/world.py

def move(self, entity_id: UUID, new_room_id: UUID) -> UUID | None:
    """Move entity to new room. Returns previous room ID."""
    old_room = self._room_by_entity.get(entity_id)
    if old_room == new_room_id:
        return old_room  # Already in this room

    if old_room:
        self._entities_by_room[old_room].discard(entity_id)

    self._entities_by_room[new_room_id].add(entity_id)
    self._room_by_entity[entity_id] = new_room_id
    return old_room

remove

remove(entity_id: UUID) -> UUID | None

Remove entity from index. Returns previous room ID.

Source code in packages/maid-engine/src/maid_engine/core/world.py

def remove(self, entity_id: UUID) -> UUID | None:
    """Remove entity from index. Returns previous room ID."""
    room_id = self._room_by_entity.pop(entity_id, None)
    if room_id:
        self._entities_by_room[room_id].discard(entity_id)
    return room_id

RoomLeaveEvent dataclass

RoomLeaveEvent(
    entity_id: UUID,
    room_id: UUID,
    to_room_id: UUID | None = None,
)

Bases: Event

Emitted when an entity leaves a room.

SortOrder

Bases: Enum

Sort order for queries.

StorageAware

Bases: Protocol

Protocol for systems that need document store access.

Systems implementing this protocol will automatically receive the document store during registration with the engine.

Example

class GuildSystem(System, StorageAware): def init(self, world: World) -> None: super().init(world) self._storage: DocumentStore | None = None

def set_storage(self, store: DocumentStore) -> None:
    self._storage = store

async def startup(self) -> None:
    if self._storage:
        # Load from database
        ...

set_storage

set_storage(store: DocumentStore) -> None

Receive the document store for persistence.

Called automatically by the engine when the system is registered.

Parameters:

Name	Type	Description	Default
store	DocumentStore	The document store instance	required

Source code in packages/maid-engine/src/maid_engine/storage/protocols.py

def set_storage(self, store: "DocumentStore") -> None:
    """Receive the document store for persistence.

    Called automatically by the engine when the system is registered.

    Args:
        store: The document store instance
    """
    ...

System

System(world: 'World')

Bases: ABC

Base class for all ECS systems.

Systems contain the logic that operates on entities with specific components. They are executed each tick by the SystemManager.

Example

class MovementSystem(System): priority = 10 # Lower = runs earlier

async def update(self, delta: float) -> None:
    for entity in self.entities.with_components(
        PositionComponent, MovementComponent
    ):
        pos = entity.get(PositionComponent)
        mov = entity.get(MovementComponent)
        # Process movement...

Source code in packages/maid-engine/src/maid_engine/core/ecs/system.py

def __init__(self, world: "World") -> None:
    self._world = world

entities property

entities: 'EntityManager'

Get the entity manager.

events property

events: 'EventBus'

Get the event bus.

world property

world: 'World'

Get the world instance.

shutdown async

shutdown() -> None

Called when the system shuts down. Override in subclasses.

Source code in packages/maid-engine/src/maid_engine/core/ecs/system.py

async def shutdown(self) -> None:  # noqa: B027
    """Called when the system shuts down. Override in subclasses."""
    pass

startup async

startup() -> None

Called when the system starts up. Override in subclasses.

Source code in packages/maid-engine/src/maid_engine/core/ecs/system.py

async def startup(self) -> None:  # noqa: B027
    """Called when the system starts up. Override in subclasses."""
    pass

update abstractmethod async

update(delta: float) -> None

Update the system.

Parameters:

Name	Type	Description	Default
delta	float	Time since last update in seconds	required

Source code in packages/maid-engine/src/maid_engine/core/ecs/system.py

@abstractmethod
async def update(self, delta: float) -> None:
    """Update the system.

    Args:
        delta: Time since last update in seconds
    """
    pass

SystemManager

SystemManager(world: 'World')

Manages and runs all ECS systems.

The SystemManager handles system lifecycle and executes systems in priority order each tick. It also tracks which content pack registered each system to support hot reloading.

Example

manager = SystemManager(world) manager.register(MovementSystem(world), source_pack="maid-stdlib") manager.register(CombatSystem(world), source_pack="maid-classic-rpg")

In game loop

await manager.update(delta_time)

Hot reload: unregister all systems from a pack

removed = await manager.unregister_systems_for_pack("maid-classic-rpg")

Source code in packages/maid-engine/src/maid_engine/core/ecs/system.py

def __init__(self, world: "World") -> None:
    self._world = world
    self._systems: list[System] = []
    self._systems_by_type: dict[type[System], System] = {}
    self._started = False

    # Track which content pack registered each system (system class name -> pack name)
    self._system_sources: dict[str, str] = {}

    # Track systems by pack for efficient lookup (pack name -> set of system class names)
    self._systems_by_pack: dict[str, set[str]] = {}

    # Track system dependencies (system class name -> set of dependency class names)
    self._system_dependencies: dict[str, set[str]] = {}

    # Tick pause control for safe system removal
    self._removal_pause_event: asyncio.Event | None = None
    self._tick_in_progress = False
    self._tick_complete_event: asyncio.Event = asyncio.Event()
    self._tick_complete_event.set()  # Initially not in a tick

is_tick_in_progress property

is_tick_in_progress: bool

Check if a tick is currently in progress.

is_tick_paused property

is_tick_paused: bool

Check if ticks are currently paused for removal.

systems property

systems: list[System]

Get all registered systems (sorted by priority).

disable

disable(system_type: type[System]) -> bool

Disable a system. Returns True if system exists.

Source code in packages/maid-engine/src/maid_engine/core/ecs/system.py

def disable(self, system_type: type[System]) -> bool:
    """Disable a system. Returns True if system exists."""
    system = self._systems_by_type.get(system_type)
    if system:
        system.enabled = False
        return True
    return False

enable

enable(system_type: type[System]) -> bool

Enable a system. Returns True if system exists.

Source code in packages/maid-engine/src/maid_engine/core/ecs/system.py

def enable(self, system_type: type[System]) -> bool:
    """Enable a system. Returns True if system exists."""
    system = self._systems_by_type.get(system_type)
    if system:
        system.enabled = True
        return True
    return False

get

get(system_type: type[System]) -> System | None

Get a registered system by type.

Source code in packages/maid-engine/src/maid_engine/core/ecs/system.py

def get(self, system_type: type[System]) -> System | None:
    """Get a registered system by type."""
    return self._systems_by_type.get(system_type)

get_or_raise

get_or_raise(system_type: type[System]) -> System

Get a registered system by type, raising KeyError if not found.

Source code in packages/maid-engine/src/maid_engine/core/ecs/system.py

def get_or_raise(self, system_type: type[System]) -> System:
    """Get a registered system by type, raising KeyError if not found."""
    system = self._systems_by_type.get(system_type)
    if system is None:
        raise KeyError(f"System {system_type.__name__} not registered")
    return system

get_source_pack

get_source_pack(system_type: type[System]) -> str | None

Get the source pack name for a system.

Parameters:

Name	Type	Description	Default
system_type	type[System]	The type of system to look up	required

Returns:

Type	Description
str | None	The pack name or None if no source pack was specified

Source code in packages/maid-engine/src/maid_engine/core/ecs/system.py

def get_source_pack(self, system_type: type[System]) -> str | None:
    """Get the source pack name for a system.

    Args:
        system_type: The type of system to look up

    Returns:
        The pack name or None if no source pack was specified
    """
    return self._system_sources.get(system_type.__name__)

get_system_dependents

get_system_dependents(
    system_type: type[System],
) -> list[type[System]]

Get systems that depend on the given system type.

Parameters:

Name	Type	Description	Default
system_type	type[System]	The system type to check	required

Returns:

Type	Description
list[type[System]]	List of system types that depend on this system

Source code in packages/maid-engine/src/maid_engine/core/ecs/system.py

def get_system_dependents(
    self, system_type: type[System]
) -> list[type[System]]:
    """Get systems that depend on the given system type.

    Args:
        system_type: The system type to check

    Returns:
        List of system types that depend on this system
    """
    system_name = system_type.__name__
    dependents: list[type[System]] = []

    for other_type in self._systems_by_type:
        other_name = other_type.__name__
        deps = self._system_dependencies.get(other_name, set())
        if system_name in deps:
            dependents.append(other_type)

    return dependents

get_system_types_for_pack

get_system_types_for_pack(
    pack_name: str,
) -> list[type[System]]

Get all system types registered by a specific content pack.

Parameters:

Name	Type	Description	Default
pack_name	str	The name of the content pack	required

Returns:

Type	Description
list[type[System]]	List of system types registered by this pack

Source code in packages/maid-engine/src/maid_engine/core/ecs/system.py

def get_system_types_for_pack(self, pack_name: str) -> list[type[System]]:
    """Get all system types registered by a specific content pack.

    Args:
        pack_name: The name of the content pack

    Returns:
        List of system types registered by this pack
    """
    system_names = self._systems_by_pack.get(pack_name, set())
    return [
        system_type
        for system_type in self._systems_by_type
        if system_type.__name__ in system_names
    ]

get_systems_for_pack

get_systems_for_pack(pack_name: str) -> list[System]

Get all systems registered by a specific content pack.

Parameters:

Name	Type	Description	Default
pack_name	str	The name of the content pack	required

Returns:

Type	Description
list[System]	List of systems registered by this pack

Source code in packages/maid-engine/src/maid_engine/core/ecs/system.py

def get_systems_for_pack(self, pack_name: str) -> list[System]:
    """Get all systems registered by a specific content pack.

    Args:
        pack_name: The name of the content pack

    Returns:
        List of systems registered by this pack
    """
    system_names = self._systems_by_pack.get(pack_name, set())
    result: list[System] = []
    for system in self._systems:
        if type(system).__name__ in system_names:
            result.append(system)
    return result

pause_for_removal async

pause_for_removal() -> None

Pause tick processing for safe system removal.

This method waits for any in-progress tick to complete and prevents new ticks from starting until resume_after_removal is called.

Source code in packages/maid-engine/src/maid_engine/core/ecs/system.py

async def pause_for_removal(self) -> None:
    """Pause tick processing for safe system removal.

    This method waits for any in-progress tick to complete and
    prevents new ticks from starting until resume_after_removal is called.
    """
    # Wait for current tick to complete if one is in progress
    if self._tick_in_progress:
        _logger.debug("Waiting for current tick to complete...")
        await self._tick_complete_event.wait()

    # Set pause event to prevent new ticks
    self._removal_pause_event = asyncio.Event()
    _logger.debug("System removal pause activated")

register

register(
    system: System,
    source_pack: str | None = None,
    dependencies: list[type[System]] | None = None,
) -> None

Register a system.

Parameters:

Name	Type	Description	Default
system	System	The system instance to register	required
source_pack	str | None	Optional name of the content pack that registered this system	None
dependencies	list[type[System]] | None	Optional list of system types this system depends on	None

Source code in packages/maid-engine/src/maid_engine/core/ecs/system.py

def register(
    self,
    system: System,
    source_pack: str | None = None,
    dependencies: list[type[System]] | None = None,
) -> None:
    """Register a system.

    Args:
        system: The system instance to register
        source_pack: Optional name of the content pack that registered this system
        dependencies: Optional list of system types this system depends on
    """
    system_type = type(system)
    system_name = system_type.__name__

    if system_type in self._systems_by_type:
        raise ValueError(f"System {system_name} already registered")

    self._systems.append(system)
    self._systems_by_type[system_type] = system

    # Track source pack
    if source_pack is not None:
        self._system_sources[system_name] = source_pack
        if source_pack not in self._systems_by_pack:
            self._systems_by_pack[source_pack] = set()
        self._systems_by_pack[source_pack].add(system_name)

    # Track dependencies
    if dependencies:
        self._system_dependencies[system_name] = {
            dep.__name__ for dep in dependencies
        }

    # Keep systems sorted by priority
    self._systems.sort(key=lambda s: s.priority)

    _logger.debug(
        f"Registered system {system_name}"
        + (f" from pack {source_pack}" if source_pack else "")
    )

remove_system_safely async

remove_system_safely(
    system_type: type[System],
    check_dependencies: bool = True,
) -> System | None

Remove a single system safely with tick pause.

This is a convenience method that handles pausing ticks, removing the system, and resuming ticks.

Parameters:

Name	Type	Description	Default
system_type	type[System]	The type of system to remove	required
check_dependencies	bool	Whether to check for dependent systems	True

Returns:

Type	Description
System | None	The removed system or None if not found

Raises:

Type	Description
SystemDependencyError	If other systems depend on this one

Source code in packages/maid-engine/src/maid_engine/core/ecs/system.py

async def remove_system_safely(
    self,
    system_type: type[System],
    check_dependencies: bool = True,
) -> System | None:
    """Remove a single system safely with tick pause.

    This is a convenience method that handles pausing ticks,
    removing the system, and resuming ticks.

    Args:
        system_type: The type of system to remove
        check_dependencies: Whether to check for dependent systems

    Returns:
        The removed system or None if not found

    Raises:
        SystemDependencyError: If other systems depend on this one
    """
    async with self.safe_removal_context() as ctx:
        return await ctx.remove_system(
            system_type, check_dependencies=check_dependencies
        )

resume_after_removal async

resume_after_removal() -> None

Resume tick processing after system removal.

Source code in packages/maid-engine/src/maid_engine/core/ecs/system.py

async def resume_after_removal(self) -> None:
    """Resume tick processing after system removal."""
    if self._removal_pause_event is not None:
        self._removal_pause_event.set()
        self._removal_pause_event = None
        _logger.debug("System removal pause deactivated")

safe_removal_context async

safe_removal_context() -> (
    "AsyncIterator[SystemRemovalContext]"
)

Context manager for safe system removal.

This provides a convenient way to safely remove systems while ensuring ticks are paused and properly resumed.

Example

async with manager.safe_removal_context() as ctx: await ctx.remove_system(MovementSystem) await ctx.remove_system(CombatSystem)

Yields:

Type	Description
'AsyncIterator[SystemRemovalContext]'	SystemRemovalContext for performing removals

Source code in packages/maid-engine/src/maid_engine/core/ecs/system.py

@asynccontextmanager
async def safe_removal_context(self) -> "AsyncIterator[SystemRemovalContext]":
    """Context manager for safe system removal.

    This provides a convenient way to safely remove systems while
    ensuring ticks are paused and properly resumed.

    Example:
        async with manager.safe_removal_context() as ctx:
            await ctx.remove_system(MovementSystem)
            await ctx.remove_system(CombatSystem)

    Yields:
        SystemRemovalContext for performing removals
    """
    ctx = SystemRemovalContext(self)
    async with ctx:
        yield ctx

shutdown async

shutdown() -> None

Shut down all systems.

Source code in packages/maid-engine/src/maid_engine/core/ecs/system.py

async def shutdown(self) -> None:
    """Shut down all systems."""
    if not self._started:
        return

    # Copy list to avoid concurrent modification if systems unregister during shutdown
    systems = list(self._systems)

    # Shut down in reverse order
    for system in reversed(systems):
        await system.shutdown()

    self._started = False

startup async

startup() -> None

Start all systems.

Source code in packages/maid-engine/src/maid_engine/core/ecs/system.py

async def startup(self) -> None:
    """Start all systems."""
    if self._started:
        return

    for system in self._systems:
        await system.startup()

    self._started = True

unregister

unregister(system_type: type[System]) -> System | None

Unregister a system by type.

Parameters:

Name	Type	Description	Default
system_type	type[System]	The type of system to unregister	required

Returns:

Type	Description
System | None	The unregistered system or None if not found

Source code in packages/maid-engine/src/maid_engine/core/ecs/system.py

def unregister(self, system_type: type[System]) -> System | None:
    """Unregister a system by type.

    Args:
        system_type: The type of system to unregister

    Returns:
        The unregistered system or None if not found
    """
    system = self._systems_by_type.pop(system_type, None)
    if system:
        system_name = system_type.__name__
        self._systems.remove(system)

        # Clean up source pack tracking
        if system_name in self._system_sources:
            pack_name = self._system_sources.pop(system_name)
            if pack_name in self._systems_by_pack:
                self._systems_by_pack[pack_name].discard(system_name)
                if not self._systems_by_pack[pack_name]:
                    del self._systems_by_pack[pack_name]

        # Clean up dependencies
        self._system_dependencies.pop(system_name, None)

        _logger.debug(f"Unregistered system {system_name}")

    return system

unregister_systems_for_pack async

unregister_systems_for_pack(
    pack_name: str, check_dependencies: bool = True
) -> list[System]

Unregister all systems registered by a content pack.

This method safely removes all systems from a pack, calling shutdown on each system and respecting dependency ordering.

Parameters:

Name	Type	Description	Default
pack_name	str	The name of the content pack	required
check_dependencies	bool	Whether to check for and warn about dependencies	True

Returns:

Type	Description
list[System]	List of systems that were unregistered

Raises:

Type	Description
SystemDependencyError	If check_dependencies is True and systems from other packs depend on systems being removed

Source code in packages/maid-engine/src/maid_engine/core/ecs/system.py

async def unregister_systems_for_pack(
    self,
    pack_name: str,
    check_dependencies: bool = True,
) -> list[System]:
    """Unregister all systems registered by a content pack.

    This method safely removes all systems from a pack, calling
    shutdown on each system and respecting dependency ordering.

    Args:
        pack_name: The name of the content pack
        check_dependencies: Whether to check for and warn about dependencies

    Returns:
        List of systems that were unregistered

    Raises:
        SystemDependencyError: If check_dependencies is True and systems
            from other packs depend on systems being removed
    """
    systems_to_remove = self.get_system_types_for_pack(pack_name)
    if not systems_to_remove:
        return []

    # Check dependencies from other packs
    if check_dependencies:
        for system_type in systems_to_remove:
            dependents = self.get_system_dependents(system_type)
            # Filter to only dependents from other packs
            external_dependents = [
                dep
                for dep in dependents
                if self.get_source_pack(dep) != pack_name
            ]
            if external_dependents:
                raise SystemDependencyError(
                    f"Cannot remove systems from pack '{pack_name}': "
                    f"{system_type.__name__} has dependents from other packs",
                    system_type=system_type,
                    dependents=external_dependents,
                )

    removed: list[System] = []

    # Remove in reverse priority order (highest priority first)
    systems_to_remove.sort(
        key=lambda t: self._systems_by_type.get(t, System).priority,
        reverse=True,
    )

    for system_type in systems_to_remove:
        system = self.unregister(system_type)
        if system:
            # Call shutdown on the removed system
            if self._started:
                try:
                    await system.shutdown()
                except Exception as e:
                    _logger.warning(
                        f"Error during shutdown of {system_type.__name__}: {e}"
                    )
            removed.append(system)

    _logger.info(
        f"Unregistered {len(removed)} systems from pack '{pack_name}'"
    )
    return removed

update async

update(delta: float) -> None

Update all enabled systems.

Parameters:

Name	Type	Description	Default
delta	float	Time since last update in seconds	required

Source code in packages/maid-engine/src/maid_engine/core/ecs/system.py

async def update(self, delta: float) -> None:
    """Update all enabled systems.

    Args:
        delta: Time since last update in seconds
    """
    # Check if paused for removal
    if self._removal_pause_event is not None:
        _logger.debug("Tick paused for system removal")
        await self._removal_pause_event.wait()

    # Mark tick as in progress
    self._tick_in_progress = True
    self._tick_complete_event.clear()

    try:
        # Copy systems list to avoid concurrent modification during iteration
        systems = list(self._systems)
        for system in systems:
            if system.enabled:
                await system.update(delta)
    finally:
        # Mark tick as complete
        self._tick_in_progress = False
        self._tick_complete_event.set()

TickEvent dataclass

TickEvent(tick_number: int, delta: float)

Bases: Event

Emitted at the start of each game tick.

TranslationCatalog

TranslationCatalog(locale: str)

A catalog of translations for a single locale.

The catalog stores translations indexed by their msgid (and optional context). It supports loading from PO and MO files, plural forms, and context-based lookups.

Attributes:

Name	Type	Description
locale		The locale code (e.g., 'en_US', 'de_DE').
entries	dict[str, TranslationEntry]	Dictionary mapping keys to TranslationEntry objects.

Parameters:

Name	Type	Description	Default
locale	str	The locale code for this catalog.	required

Source code in packages/maid-engine/src/maid_engine/i18n/catalog.py

def __init__(self, locale: str) -> None:
    """Initialize a translation catalog.

    Args:
        locale: The locale code for this catalog.
    """
    self.locale = locale
    self.entries: dict[str, TranslationEntry] = {}
    self._nplurals: int = 2
    self._plural_expr: str = "(n != 1)"
    self._metadata: dict[str, str] = {}

metadata property

metadata: dict[str, str]

Return the catalog metadata.

nplurals property

nplurals: int

Return the number of plural forms for this catalog.

plural_expr property

plural_expr: str

Return the plural expression for this catalog.

get_entry

get_entry(
    msgid: str, context: str | None = None
) -> TranslationEntry | None

Get the translation entry for a msgid.

Parameters:

Name	Type	Description	Default
msgid	str	The message to look up.	required
context	str | None	Optional context for disambiguation.	None

Returns:

Type	Description
TranslationEntry | None	The TranslationEntry, or None if not found.

Source code in packages/maid-engine/src/maid_engine/i18n/catalog.py

def get_entry(self, msgid: str, context: str | None = None) -> TranslationEntry | None:
    """Get the translation entry for a msgid.

    Args:
        msgid: The message to look up.
        context: Optional context for disambiguation.

    Returns:
        The TranslationEntry, or None if not found.
    """
    key = self._make_key(msgid, context)
    return self.entries.get(key)

has_translation

has_translation(
    msgid: str, context: str | None = None
) -> bool

Check if a translation exists for the given msgid.

Parameters:

Name	Type	Description	Default
msgid	str	The message to check.	required
context	str | None	Optional context for disambiguation.	None

Returns:

Type	Description
bool	True if a non-fuzzy translation exists.

Source code in packages/maid-engine/src/maid_engine/i18n/catalog.py

def has_translation(self, msgid: str, context: str | None = None) -> bool:
    """Check if a translation exists for the given msgid.

    Args:
        msgid: The message to check.
        context: Optional context for disambiguation.

    Returns:
        True if a non-fuzzy translation exists.
    """
    key = self._make_key(msgid, context)
    entry = self.entries.get(key)
    return entry is not None and entry.is_translated and not entry.fuzzy

load classmethod

load(
    path: str | Path, locale: str | None = None
) -> TranslationCatalog

Load a translation catalog from a file.

Supports both PO and MO file formats (detected by extension).

Parameters:

Name	Type	Description	Default
path	str | Path	Path to the translation file.	required
locale	str | None	Optional locale override (defaults to filename).	None

Returns:

Type	Description
TranslationCatalog	A TranslationCatalog instance.

Raises:

Type	Description
FileNotFoundError	If the file doesn't exist.
ValueError	If the file format is not supported.

Source code in packages/maid-engine/src/maid_engine/i18n/catalog.py

@classmethod
def load(cls, path: str | Path, locale: str | None = None) -> TranslationCatalog:
    """Load a translation catalog from a file.

    Supports both PO and MO file formats (detected by extension).

    Args:
        path: Path to the translation file.
        locale: Optional locale override (defaults to filename).

    Returns:
        A TranslationCatalog instance.

    Raises:
        FileNotFoundError: If the file doesn't exist.
        ValueError: If the file format is not supported.
    """
    path = Path(path)

    if not path.exists():
        raise FileNotFoundError(f"Translation file not found: {path}")

    # Determine locale from filename if not provided
    if locale is None:
        locale = path.stem

    catalog = cls(locale)

    suffix = path.suffix.lower()
    if suffix == ".po":
        catalog._load_po(path)
    elif suffix == ".mo":
        catalog._load_mo(path)
    else:
        raise ValueError(f"Unsupported translation file format: {suffix}")

    return catalog

translate

translate(
    msgid: str,
    context: str | None = None,
    n: int | None = None,
    default: str | None = None,
) -> str

Translate a message.

Parameters:

Name	Type	Description	Default
msgid	str	The message to translate.	required
context	str | None	Optional context for disambiguation.	None
n	int | None	The number for plural selection (None for singular).	None
default	str | None	Default value if translation not found (defaults to msgid).	None

Returns:

Type	Description
str	The translated string, or default/msgid if not found.

Source code in packages/maid-engine/src/maid_engine/i18n/catalog.py

def translate(
    self,
    msgid: str,
    context: str | None = None,
    n: int | None = None,
    default: str | None = None,
) -> str:
    """Translate a message.

    Args:
        msgid: The message to translate.
        context: Optional context for disambiguation.
        n: The number for plural selection (None for singular).
        default: Default value if translation not found (defaults to msgid).

    Returns:
        The translated string, or default/msgid if not found.
    """
    key = self._make_key(msgid, context)
    entry = self.entries.get(key)

    if entry is None or not entry.is_translated:
        return default if default is not None else msgid

    # Skip fuzzy translations
    if entry.fuzzy:
        return default if default is not None else msgid

    return entry.get_translation(n, self._plural_expr)

TranslationEntry dataclass

TranslationEntry(
    msgid: str,
    msgstr: str = "",
    msgstr_plural: list[str] = list(),
    msgid_plural: str | None = None,
    context: str | None = None,
    fuzzy: bool = False,
)

A single translation entry.

Attributes:

Name	Type	Description
msgid	str	The original message (source string).
msgstr	str	The translated message (single string for singular).
msgstr_plural	list[str]	List of plural forms (index 0 is singular).
msgid_plural	str | None	The plural form of the source string.
context	str | None	Optional context for disambiguation.
fuzzy	bool	Whether this translation is marked as fuzzy/uncertain.

is_plural property

is_plural: bool

Return True if this is a plural entry.

is_translated property

is_translated: bool

Return True if this entry has a translation.

get_translation

get_translation(
    n: int | None = None, plural_expr: str = "(n != 1)"
) -> str

Get the translation for this entry.

Parameters:

Name	Type	Description	Default
n	int | None	The number for plural selection (None for singular).	None
plural_expr	str	Expression to evaluate plural form index.	'(n != 1)'

Returns:

Type	Description
str	The translated string, or empty string if not translated.

Source code in packages/maid-engine/src/maid_engine/i18n/catalog.py

def get_translation(self, n: int | None = None, plural_expr: str = "(n != 1)") -> str:
    """Get the translation for this entry.

    Args:
        n: The number for plural selection (None for singular).
        plural_expr: Expression to evaluate plural form index.

    Returns:
        The translated string, or empty string if not translated.
    """
    if n is not None and self.is_plural and self.msgstr_plural:
        # Plural form requested
        index = evaluate_plural(n, plural_expr)
        index = min(index, len(self.msgstr_plural) - 1)
        return self.msgstr_plural[index] if index < len(self.msgstr_plural) else ""
    return self.msgstr

Translator

Translator(
    default_locale: str = "en",
    fallback_locale: str | None = None,
    log_missing: bool = True,
    missing_handler: Callable[[str, str | None, str], None]
    | None = None,
)

Translation service that manages catalogs and provides translation functions.

The Translator loads translation catalogs for multiple locales and provides methods for translating messages with support for: - Locale fallback chains - Plural forms - Context-based disambiguation - Missing translation logging

Example

translator = Translator(default_locale="en", fallback_locale="en") translator.load_directory("/path/to/locales") translator.translate("Hello, world!") 'Hallo, Welt!' # if current_locale is 'de'

Parameters:

Name	Type	Description	Default
default_locale	str	The default locale to use when none is set.	'en'
fallback_locale	str | None	Locale to fall back to when translation is missing.	None
log_missing	bool	Whether to log missing translations.	True
missing_handler	Callable[[str, str | None, str], None] | None	Optional callback for missing translations. Called with (msgid, context, locale).	None

Source code in packages/maid-engine/src/maid_engine/i18n/translator.py

def __init__(
    self,
    default_locale: str = "en",
    fallback_locale: str | None = None,
    log_missing: bool = True,
    missing_handler: Callable[[str, str | None, str], None] | None = None,
) -> None:
    """Initialize the translator.

    Args:
        default_locale: The default locale to use when none is set.
        fallback_locale: Locale to fall back to when translation is missing.
        log_missing: Whether to log missing translations.
        missing_handler: Optional callback for missing translations.
                        Called with (msgid, context, locale).
    """
    self.default_locale = default_locale
    self.fallback_locale = fallback_locale or default_locale
    self.log_missing = log_missing
    self.missing_handler = missing_handler

    self._catalogs: dict[str, TranslationCatalog] = {}
    self._missing_logged: set[tuple[str, str | None, str]] = set()

available_locales property

available_locales: list[str]

Return list of available locale codes.

get_catalog

get_catalog(locale: str) -> TranslationCatalog | None

Get the catalog for a specific locale.

Parameters:

Name	Type	Description	Default
locale	str	The locale code.	required

Returns:

Type	Description
TranslationCatalog | None	The TranslationCatalog, or None if not loaded.

Source code in packages/maid-engine/src/maid_engine/i18n/translator.py

def get_catalog(self, locale: str) -> TranslationCatalog | None:
    """Get the catalog for a specific locale.

    Args:
        locale: The locale code.

    Returns:
        The TranslationCatalog, or None if not loaded.
    """
    return self._catalogs.get(locale)

get_current_locale

get_current_locale() -> str

Get the current locale from context.

Returns:

Type	Description
str	The current locale code.

Source code in packages/maid-engine/src/maid_engine/i18n/translator.py

def get_current_locale(self) -> str:
    """Get the current locale from context.

    Returns:
        The current locale code.
    """
    return current_locale.get()

gettext

gettext(msgid: str) -> str

Translate a message (GNU gettext compatible).

Parameters:

Name	Type	Description	Default
msgid	str	The message to translate.	required

Returns:

Type	Description
str	The translated string.

Source code in packages/maid-engine/src/maid_engine/i18n/translator.py

def gettext(self, msgid: str) -> str:
    """Translate a message (GNU gettext compatible).

    Args:
        msgid: The message to translate.

    Returns:
        The translated string.
    """
    return self.translate(msgid)

load_catalog

load_catalog(
    path: str | Path, locale: str | None = None
) -> None

Load a translation catalog from a file.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to the PO or MO file.	required
locale	str | None	Optional locale override (defaults to filename).	None

Source code in packages/maid-engine/src/maid_engine/i18n/translator.py

def load_catalog(self, path: str | Path, locale: str | None = None) -> None:
    """Load a translation catalog from a file.

    Args:
        path: Path to the PO or MO file.
        locale: Optional locale override (defaults to filename).
    """
    catalog = TranslationCatalog.load(path, locale)
    self._catalogs[catalog.locale] = catalog
    logger.info(f"Loaded translation catalog: {catalog.locale} ({len(catalog)} entries)")

load_directory

load_directory(
    directory: str | Path,
    pattern: str = "*.po",
    recursive: bool = False,
) -> None

Load all translation catalogs from a directory.

Parameters:

Name	Type	Description	Default
directory	str | Path	Path to the directory containing translation files.	required
pattern	str	Glob pattern for translation files.	'*.po'
recursive	bool	Whether to search recursively.	False

Source code in packages/maid-engine/src/maid_engine/i18n/translator.py

def load_directory(
    self,
    directory: str | Path,
    pattern: str = "*.po",
    recursive: bool = False,
) -> None:
    """Load all translation catalogs from a directory.

    Args:
        directory: Path to the directory containing translation files.
        pattern: Glob pattern for translation files.
        recursive: Whether to search recursively.
    """
    directory = Path(directory)
    if not directory.exists():
        logger.warning(f"Locale directory does not exist: {directory}")
        return

    glob_method = directory.rglob if recursive else directory.glob

    for path in glob_method(pattern):
        try:
            self.load_catalog(path)
        except Exception as e:
            logger.error(f"Failed to load translation file {path}: {e}")

ngettext

ngettext(msgid: str, msgid_plural: str, n: int) -> str

Translate a message with plural forms.

Parameters:

Name	Type	Description	Default
msgid	str	The singular form of the message.	required
msgid_plural	str	The plural form of the message.	required
n	int	The number determining which form to use.	required

Returns:

Type	Description
str	The translated string.

Source code in packages/maid-engine/src/maid_engine/i18n/translator.py

def ngettext(self, msgid: str, msgid_plural: str, n: int) -> str:
    """Translate a message with plural forms.

    Args:
        msgid: The singular form of the message.
        msgid_plural: The plural form of the message.
        n: The number determining which form to use.

    Returns:
        The translated string.
    """
    return self.translate(msgid, n=n, msgid_plural=msgid_plural)

npgettext

npgettext(
    context: str, msgid: str, msgid_plural: str, n: int
) -> str

Translate a message with context and plural forms.

Parameters:

Name	Type	Description	Default
context	str	The context for disambiguation.	required
msgid	str	The singular form of the message.	required
msgid_plural	str	The plural form of the message.	required
n	int	The number determining which form to use.	required

Returns:

Type	Description
str	The translated string.

Source code in packages/maid-engine/src/maid_engine/i18n/translator.py

def npgettext(self, context: str, msgid: str, msgid_plural: str, n: int) -> str:
    """Translate a message with context and plural forms.

    Args:
        context: The context for disambiguation.
        msgid: The singular form of the message.
        msgid_plural: The plural form of the message.
        n: The number determining which form to use.

    Returns:
        The translated string.
    """
    return self.translate(msgid, context=context, n=n, msgid_plural=msgid_plural)

pgettext

pgettext(context: str, msgid: str) -> str

Translate a message with context.

Parameters:

Name	Type	Description	Default
context	str	The context for disambiguation.	required
msgid	str	The message to translate.	required

Returns:

Type	Description
str	The translated string.

Source code in packages/maid-engine/src/maid_engine/i18n/translator.py

def pgettext(self, context: str, msgid: str) -> str:
    """Translate a message with context.

    Args:
        context: The context for disambiguation.
        msgid: The message to translate.

    Returns:
        The translated string.
    """
    return self.translate(msgid, context=context)

set_current_locale

set_current_locale(locale: str) -> None

Set the current locale in context.

Parameters:

Name	Type	Description	Default
locale	str	The locale code to set.	required

Source code in packages/maid-engine/src/maid_engine/i18n/translator.py

def set_current_locale(self, locale: str) -> None:
    """Set the current locale in context.

    Args:
        locale: The locale code to set.
    """
    current_locale.set(locale)

translate

translate(
    msgid: str,
    context: str | None = None,
    n: int | None = None,
    msgid_plural: str | None = None,
    locale: str | None = None,
    default: str | None = None,
) -> str

Translate a message.

Parameters:

Name	Type	Description	Default
msgid	str	The message to translate.	required
context	str | None	Optional context for disambiguation.	None
n	int | None	The number for plural selection.	None
msgid_plural	str | None	The plural form of the message (for fallback).	None
locale	str | None	Optional locale override (uses current_locale if not provided).	None
default	str | None	Default value if translation not found.	None

Returns:

Type	Description
str	The translated string, or the original msgid if not found.

Source code in packages/maid-engine/src/maid_engine/i18n/translator.py

def translate(
    self,
    msgid: str,
    context: str | None = None,
    n: int | None = None,
    msgid_plural: str | None = None,
    locale: str | None = None,
    default: str | None = None,
) -> str:
    """Translate a message.

    Args:
        msgid: The message to translate.
        context: Optional context for disambiguation.
        n: The number for plural selection.
        msgid_plural: The plural form of the message (for fallback).
        locale: Optional locale override (uses current_locale if not provided).
        default: Default value if translation not found.

    Returns:
        The translated string, or the original msgid if not found.
    """
    locale = locale or self.get_current_locale()

    # Try each locale in the fallback chain
    for try_locale in self._get_fallback_chain(locale):
        catalog = self._catalogs.get(try_locale)
        if catalog is None:
            continue

        if catalog.has_translation(msgid, context):
            return catalog.translate(msgid, context, n, default)

    # Translation not found
    self._handle_missing(msgid, context, locale)

    # Return appropriate form
    if default is not None:
        return default
    if n is not None and msgid_plural is not None:
        return msgid if n == 1 else msgid_plural
    return msgid

World

World(settings: Settings)

Central game world state manager.

The World holds all game state including entities, rooms, and provides efficient queries for game logic.

Example

world = World(settings) await world.startup()

Create a player entity

player = world.create_entity() player.add(PositionComponent(room_id=starting_room.id)) player.add(HealthComponent(current=100, maximum=100))

Query entities in a room

for entity in world.entities_in_room(room_id): ...

Source code in packages/maid-engine/src/maid_engine/core/world.py

def __init__(self, settings: "Settings") -> None:
    self._settings = settings
    self._entities = EntityManager()
    self._events = EventBus()
    self._systems = SystemManager(self)
    self._room_index = RoomIndex()
    self._rooms: dict[UUID, Any] = {}  # Room data storage
    self._areas: dict[UUID, Any] = {}  # Area data storage
    self._custom_data: dict[str, Any] = {}  # Initialize in __init__ to avoid race
    self._tick_count = 0
    self._running = False
    self._engine: Any = None  # Back-reference to GameEngine, set by engine
    self._grid = GridManager(world=self)  # Grid coordinate system

engine property

engine: Any

Get the game engine reference.

This is set by the GameEngine when it creates the World. Systems can use this to access engine-level resources like character managers, item managers, etc.

entities property

entities: EntityManager

Get the entity manager.

events property

events: EventBus

Get the event bus.

grid property

grid: GridManager

Get the grid manager for coordinate-based room operations.

The GridManager provides spatial indexing and queries for rooms placed on a 3D coordinate grid. It supports: - O(1) coordinate lookups - Radius and rectangular region queries - A* pathfinding between rooms - Automatic exit creation between adjacent rooms

Example

Register a room at a coordinate

world.grid.register_room(room_id, GridCoord(5, 10))

Or use the convenience method

world.register_room_at_coord(room_id, GridCoord(5, 10))

Find a path between rooms

result = world.grid.find_path(from_room_id, to_room_id) if result.found: print(result.directions) # ["n", "e", "e"]

is_running property

is_running: bool

Check if world is running.

settings property

settings: Settings

Get settings.

systems property

systems: SystemManager

Get the system manager.

tick_count property

tick_count: int

Get current tick count.

all_areas

all_areas() -> Iterator[tuple[UUID, Any]]

Iterate over all registered areas.

Source code in packages/maid-engine/src/maid_engine/core/world.py

def all_areas(self) -> Iterator[tuple[UUID, Any]]:
    """Iterate over all registered areas."""
    yield from self._areas.items()

all_rooms

all_rooms() -> Iterator[tuple[UUID, Any]]

Iterate over all registered rooms.

Source code in packages/maid-engine/src/maid_engine/core/world.py

def all_rooms(self) -> Iterator[tuple[UUID, Any]]:
    """Iterate over all registered rooms."""
    yield from self._rooms.items()

clear

clear() -> None

Clear all world state.

Source code in packages/maid-engine/src/maid_engine/core/world.py

def clear(self) -> None:
    """Clear all world state."""
    self._entities.clear()
    self._room_index.clear()
    self._rooms.clear()
    self._areas.clear()
    self._events.clear()
    self._custom_data.clear()
    self._grid.clear()
    self._tick_count = 0

count_in_room

count_in_room(room_id: UUID) -> int

Get count of entities in a room.

Source code in packages/maid-engine/src/maid_engine/core/world.py

def count_in_room(self, room_id: UUID) -> int:
    """Get count of entities in a room."""
    return self._room_index.count_in_room(room_id)

create_entity

create_entity(entity_id: UUID | None = None) -> Entity

Create a new entity.

Source code in packages/maid-engine/src/maid_engine/core/world.py

def create_entity(self, entity_id: UUID | None = None) -> Entity:
    """Create a new entity."""
    return self._entities.create(entity_id)

delete_data

delete_data(key: str) -> Any | None

Delete arbitrary data from the world.

Source code in packages/maid-engine/src/maid_engine/core/world.py

def delete_data(self, key: str) -> Any | None:
    """Delete arbitrary data from the world."""
    return self._custom_data.pop(key, None)

destroy_entity

destroy_entity(entity_id: UUID) -> bool

Destroy an entity.

Source code in packages/maid-engine/src/maid_engine/core/world.py

def destroy_entity(self, entity_id: UUID) -> bool:
    """Destroy an entity."""
    # Clean up room index
    self._room_index.remove(entity_id)
    return self._entities.destroy(entity_id)

entities_in_room

entities_in_room(room_id: UUID) -> Iterator[Entity]

Get all entities in a specific room.

Source code in packages/maid-engine/src/maid_engine/core/world.py

def entities_in_room(self, room_id: UUID) -> Iterator[Entity]:
    """Get all entities in a specific room."""
    for entity_id in self._room_index.get_entities(room_id):
        entity = self._entities.get(entity_id)
        if entity:
            yield entity

get_area

get_area(area_id: UUID) -> Any | None

Get area data by ID.

Source code in packages/maid-engine/src/maid_engine/core/world.py

def get_area(self, area_id: UUID) -> Any | None:
    """Get area data by ID."""
    return self._areas.get(area_id)

get_data

get_data(key: str) -> Any | None

Get arbitrary data from the world.

Source code in packages/maid-engine/src/maid_engine/core/world.py

def get_data(self, key: str) -> Any | None:
    """Get arbitrary data from the world."""
    return self._custom_data.get(key)

get_entity

get_entity(entity_id: UUID) -> Entity | None

Get an entity by ID.

Source code in packages/maid-engine/src/maid_engine/core/world.py

def get_entity(self, entity_id: UUID) -> Entity | None:
    """Get an entity by ID."""
    return self._entities.get(entity_id)

get_entity_room

get_entity_room(entity_id: UUID) -> UUID | None

Get the room ID an entity is in.

Source code in packages/maid-engine/src/maid_engine/core/world.py

def get_entity_room(self, entity_id: UUID) -> UUID | None:
    """Get the room ID an entity is in."""
    return self._room_index.get_room(entity_id)

get_room

get_room(room_id: UUID) -> Any | None

Get room data by ID.

Source code in packages/maid-engine/src/maid_engine/core/world.py

def get_room(self, room_id: UUID) -> Any | None:
    """Get room data by ID."""
    return self._rooms.get(room_id)

increment_tick

increment_tick() -> int

Increment and return tick count.

Source code in packages/maid-engine/src/maid_engine/core/world.py

def increment_tick(self) -> int:
    """Increment and return tick count."""
    self._tick_count += 1
    return self._tick_count

move_entity

move_entity(entity_id: UUID, to_room_id: UUID) -> bool

Move an entity to a new room.

Updates both the room index and any PositionComponent on the entity. Emits RoomLeaveEvent and RoomEnterEvent.

Returns:

Type	Description
bool	True if move was successful

Source code in packages/maid-engine/src/maid_engine/core/world.py

def move_entity(self, entity_id: UUID, to_room_id: UUID) -> bool:
    """Move an entity to a new room.

    Updates both the room index and any PositionComponent on the entity.
    Emits RoomLeaveEvent and RoomEnterEvent.

    Returns:
        True if move was successful
    """
    from maid_engine.core.events import RoomEnterEvent, RoomLeaveEvent

    entity = self._entities.get(entity_id)
    if not entity:
        return False

    # Get current room before moving
    from_room_id = self._room_index.get_room(entity_id)

    # Emit leave event if entity was in a room
    if from_room_id:
        self._events.emit_sync(
            RoomLeaveEvent(
                entity_id=entity_id,
                room_id=from_room_id,
                to_room_id=to_room_id,
            )
        )

    # Update room index
    self._room_index.move(entity_id, to_room_id)

    # Update PositionComponent if entity has one
    for comp in entity.components:
        # Check for room_id attribute and component_type to identify position components
        if hasattr(comp, "room_id") and getattr(comp.__class__, "component_type", "") in ("PositionComponent", "Position"):
            comp.room_id = to_room_id
            break

    # Emit enter event
    self._events.emit_sync(
        RoomEnterEvent(
            entity_id=entity_id,
            room_id=to_room_id,
            from_room_id=from_room_id,
        )
    )

    return True

place_entity_in_room

place_entity_in_room(
    entity_id: UUID, room_id: UUID
) -> None

Place an entity in a room (for initial placement).

Source code in packages/maid-engine/src/maid_engine/core/world.py

def place_entity_in_room(self, entity_id: UUID, room_id: UUID) -> None:
    """Place an entity in a room (for initial placement)."""
    self._room_index.add(entity_id, room_id)

    # Update PositionComponent if entity has one
    entity = self._entities.get(entity_id)
    if entity:
        for comp in entity.components:
            # Check for room_id attribute and component_type to identify position components
            if hasattr(comp, "room_id") and getattr(comp.__class__, "component_type", "") in ("PositionComponent", "Position"):
                comp.room_id = room_id
                break

register_area

register_area(area_id: UUID, area_data: Any) -> None

Register an area in the world.

Source code in packages/maid-engine/src/maid_engine/core/world.py

def register_area(self, area_id: UUID, area_data: Any) -> None:
    """Register an area in the world."""
    self._areas[area_id] = area_data

register_room

register_room(room_id: UUID, room_data: Any) -> None

Register a room in the world.

Source code in packages/maid-engine/src/maid_engine/core/world.py

def register_room(self, room_id: UUID, room_data: Any) -> None:
    """Register a room in the world."""
    self._rooms[room_id] = room_data

register_room_at_coord

register_room_at_coord(
    room_id: UUID,
    coord: GridCoord,
    movement_cost: float = 1.0,
    terrain_type: str | None = None,
    metadata: dict[str, Any] | None = None,
) -> GridRoom

Register a room at a specific grid coordinate.

This is a convenience method that delegates to grid.register_room(). If the GridManager has auto_create_exits enabled, this will automatically create bidirectional exits to adjacent registered rooms.

Parameters:

Name	Type	Description	Default
room_id	UUID	The UUID of the room entity.	required
coord	GridCoord	The grid coordinate for the room.	required
movement_cost	float	Cost to enter this room for pathfinding (default 1.0). Higher values make the room less desirable for pathfinding.	1.0
terrain_type	str | None	Optional terrain type string (e.g., "forest", "water").	None
metadata	dict[str, Any] | None	Optional dictionary of additional room metadata.	None

Returns:

Type	Description
GridRoom	The created GridRoom object.

Raises:

Type	Description
CoordinateOccupiedError	If a room already exists at the coordinate.
CoordinateBlockedError	If the coordinate is blocked.

Example

from maid_engine.core.grid import GridCoord

Register the starting room

starting_room = world.create_entity() world.register_room(starting_room.id, {"name": "Town Square"}) world.register_room_at_coord( starting_room.id, GridCoord(0, 0), terrain_type="cobblestone", )

Register adjacent rooms - exits are created automatically

inn = world.create_entity() world.register_room(inn.id, {"name": "The Rusty Lantern Inn"}) world.register_room_at_coord(inn.id, GridCoord(0, 1))

Source code in packages/maid-engine/src/maid_engine/core/world.py

def register_room_at_coord(
    self,
    room_id: UUID,
    coord: GridCoord,
    movement_cost: float = 1.0,
    terrain_type: str | None = None,
    metadata: dict[str, Any] | None = None,
) -> GridRoom:
    """Register a room at a specific grid coordinate.

    This is a convenience method that delegates to grid.register_room().
    If the GridManager has auto_create_exits enabled, this will automatically
    create bidirectional exits to adjacent registered rooms.

    Args:
        room_id: The UUID of the room entity.
        coord: The grid coordinate for the room.
        movement_cost: Cost to enter this room for pathfinding (default 1.0).
                      Higher values make the room less desirable for pathfinding.
        terrain_type: Optional terrain type string (e.g., "forest", "water").
        metadata: Optional dictionary of additional room metadata.

    Returns:
        The created GridRoom object.

    Raises:
        CoordinateOccupiedError: If a room already exists at the coordinate.
        CoordinateBlockedError: If the coordinate is blocked.

    Example:
        from maid_engine.core.grid import GridCoord

        # Register the starting room
        starting_room = world.create_entity()
        world.register_room(starting_room.id, {"name": "Town Square"})
        world.register_room_at_coord(
            starting_room.id,
            GridCoord(0, 0),
            terrain_type="cobblestone",
        )

        # Register adjacent rooms - exits are created automatically
        inn = world.create_entity()
        world.register_room(inn.id, {"name": "The Rusty Lantern Inn"})
        world.register_room_at_coord(inn.id, GridCoord(0, 1))
    """
    return self._grid.register_room(
        room_id,
        coord,
        movement_cost=movement_cost,
        terrain_type=terrain_type,
        metadata=metadata,
    )

remove_entity_from_room

remove_entity_from_room(entity_id: UUID) -> UUID | None

Remove an entity from its current room without destroying it.

This is used when an item is picked up and moves to inventory.

Returns:

Type	Description
UUID | None	The room ID the entity was in, or None if not in a room

Source code in packages/maid-engine/src/maid_engine/core/world.py

def remove_entity_from_room(self, entity_id: UUID) -> UUID | None:
    """Remove an entity from its current room without destroying it.

    This is used when an item is picked up and moves to inventory.

    Returns:
        The room ID the entity was in, or None if not in a room
    """
    return self._room_index.remove(entity_id)

room_count

room_count() -> int

Get count of registered rooms.

Source code in packages/maid-engine/src/maid_engine/core/world.py

def room_count(self) -> int:
    """Get count of registered rooms."""
    return len(self._rooms)

set_data

set_data(key: str, value: Any) -> None

Store arbitrary data in the world.

Source code in packages/maid-engine/src/maid_engine/core/world.py

def set_data(self, key: str, value: Any) -> None:
    """Store arbitrary data in the world."""
    self._custom_data[key] = value

shutdown async

shutdown() -> None

Shut down the world.

Source code in packages/maid-engine/src/maid_engine/core/world.py

async def shutdown(self) -> None:
    """Shut down the world."""
    if not self._running:
        return

    await self._systems.shutdown()
    self._running = False

startup async

startup() -> None

Start up the world.

Source code in packages/maid-engine/src/maid_engine/core/world.py

async def startup(self) -> None:
    """Start up the world."""
    if self._running:
        return

    await self._systems.startup()
    self._running = True

tick async

tick(delta: float) -> None

Process a single game tick.

Parameters:

Name	Type	Description	Default
delta	float	Time since last tick in seconds	required

Source code in packages/maid-engine/src/maid_engine/core/world.py

async def tick(self, delta: float) -> None:
    """Process a single game tick.

    Args:
        delta: Time since last tick in seconds
    """
    self.increment_tick()

    # Process any pending events from sync emitters
    await self._events.process_pending()

    # Update all systems
    await self._systems.update(delta)

    # Process any events generated during system updates
    await self._events.process_pending()

unregister_area

unregister_area(area_id: UUID) -> Any | None

Unregister an area from the world.

Source code in packages/maid-engine/src/maid_engine/core/world.py

def unregister_area(self, area_id: UUID) -> Any | None:
    """Unregister an area from the world."""
    return self._areas.pop(area_id, None)

unregister_room

unregister_room(room_id: UUID) -> Any | None

Unregister a room from the world.

Source code in packages/maid-engine/src/maid_engine/core/world.py

def unregister_room(self, room_id: UUID) -> Any | None:
    """Unregister a room from the world."""
    return self._rooms.pop(room_id, None)

discover_content_packs

discover_content_packs(
    search_paths: list[Path] | None = None,
    use_entry_points: bool = True,
) -> list[ContentPack]

Discover available content packs.

Content packs are discovered from: 1. Python entry points (maid.content_packs) - installed packages 2. Local directories containing manifest.toml files

Parameters:

Name	Type	Description	Default
search_paths	list[Path] | None	Additional directories to search for packs	None
use_entry_points	bool	Whether to search entry points	True

Returns:

Type	Description
list[ContentPack]	List of discovered ContentPack instances

Source code in packages/maid-engine/src/maid_engine/plugins/loader.py

def discover_content_packs(
    search_paths: list[Path] | None = None,
    use_entry_points: bool = True,
) -> list[ContentPack]:
    """Discover available content packs.

    Content packs are discovered from:
    1. Python entry points (maid.content_packs) - installed packages
    2. Local directories containing manifest.toml files

    Args:
        search_paths: Additional directories to search for packs
        use_entry_points: Whether to search entry points

    Returns:
        List of discovered ContentPack instances
    """
    packs: list[ContentPack] = []

    # Discover from entry points
    if use_entry_points:
        packs.extend(_discover_from_entry_points())

    # Discover from search paths
    if search_paths:
        for path in search_paths:
            if path.exists() and path.is_dir():
                packs.extend(_discover_from_directory(path))

    return packs

get_translator

get_translator() -> Translator | None

Get the global translator instance.

Returns:

Type	Description
Translator | None	The global Translator for the current context, or None if not set.

Source code in packages/maid-engine/src/maid_engine/i18n/translator.py

def get_translator() -> Translator | None:
    """Get the global translator instance.

    Returns:
        The global Translator for the current context, or None if not set.
    """
    return _translator.get()

set_translator

set_translator(translator: Translator | None) -> None

Set the global translator instance.

Parameters:

Name	Type	Description	Default
translator	Translator | None	The Translator to use globally, or None to clear.	required

Source code in packages/maid-engine/src/maid_engine/i18n/translator.py

def set_translator(translator: Translator | None) -> None:
    """Set the global translator instance.

    Args:
        translator: The Translator to use globally, or None to clear.
    """
    _translator.set(translator)