Skip to content

Multi-World Support

MAID supports running multiple game worlds simultaneously, connected via portals. This enables instanced dungeons, parallel dimensions, or separate game areas.

Overview

The WorldManager coordinates multiple world instances:

from maid_engine.core.multiworld import WorldManager

manager = WorldManager()

# Create multiple worlds
main_world = manager.create_world("main", settings)
dungeon_world = manager.create_world("dungeon-1", settings)

# Connect with a portal
manager.create_portal(
    source_world="main",
    source_room=portal_room_id,
    target_world="dungeon-1",
    target_room=dungeon_entrance_id,
)

# Transition a player
await manager.transition_player(
    player_id,
    from_world="main",
    to_world="dungeon-1",
    to_room=dungeon_entrance_id,
)

Important limitations (read before building on this):

  • max_players is not enforced. WorldManager.transition_player() never checks capacity — the field is advisory. If you want limits, count players yourself and reject transitions before calling transition_player() (see Validate Transitions).
  • The engine does not track per-world player counts. Helpers such as count_players(world) and find_player_world(manager, player_id) used in the examples below are functions you implement (for example by scanning a world's player-tagged entities or keeping your own registry). They are not WorldManager methods.
  • Transitions do not re-route a live session. transition_player() moves the player entity (and its inventory) between World objects. It does not repoint a connected player's network session or command routing at the new world, and there is no engine-level session.current_world that commands dispatch on. session.current_world in the examples is your own bookkeeping; wiring a connected player's input/output to the target world is your responsibility.

Creating Worlds

Basic World Creation

from maid_engine.config import get_settings

manager = WorldManager()
settings = get_settings()

# Create a world with default settings
world = manager.create_world("main", settings)

# Create with custom properties
dungeon = manager.create_world(
    "dungeon-dragon-lair",
    settings,
    display_name="Dragon's Lair",
    description="A dangerous dungeon filled with treasure and dragons.",
    max_players=10,  # Advisory only — NOT enforced by transition_player()
    is_public=False,  # Hide from world list
)

World Instance Properties

Each world is wrapped in a WorldInstance:

@dataclass
class WorldInstance:
    id: str                    # Unique identifier
    world: World               # The actual World object
    display_name: str          # Human-readable name
    description: str           # World description
    max_players: int           # 0 = unlimited
    is_public: bool            # Listed in world browser
    metadata: dict[str, Any]   # Custom data

Accessing Worlds

# Get a world by ID
world = manager.get_world("main")

# Get the full instance (with metadata)
instance = manager.get_instance("dungeon-1")

# Get default world
main = manager.default_world

# List all public worlds
public_worlds = manager.list_worlds(include_private=False)

# List all worlds
all_worlds = manager.list_worlds(include_private=True)

Portals

Creating Portals

Portals connect rooms in different worlds:

portal = manager.create_portal(
    source_world="main",
    source_room=town_portal_id,
    target_world="dungeon-1",
    target_room=dungeon_entrance_id,
    bidirectional=True,  # Can travel both ways
    name="Dungeon Portal",
    description="A swirling vortex of darkness.",
)

Portal Properties

@dataclass
class PortalConnection:
    id: UUID
    source_world_id: str
    source_room_id: UUID
    target_world_id: str
    target_room_id: UUID
    bidirectional: bool  # Two-way travel
    name: str
    description: str

One-Way Portals

# One-way portal (no return)
exit_portal = manager.create_portal(
    source_world="dungeon-1",
    source_room=boss_room_id,
    target_world="main",
    target_room=graveyard_id,
    bidirectional=False,  # Can only go forward
    name="Escape Rift",
    description="A tear in reality leading back to the mortal world.",
)

Finding Portals

# Get portals in a room
portals = manager.get_portals_in_room("main", portal_room_id)

for portal in portals:
    print(f"Portal to {portal.target_world_id}: {portal.name}")

Removing Portals

# Remove a specific portal
manager.remove_portal(portal.id)

# Removing a world removes its portals automatically. In real cleanup, call
# `await manager.get_world("dungeon-1").shutdown()` first — remove_world() does
# not shut the world down (see World Lifecycle below).
manager.remove_world("dungeon-1")

Player Transitions

Basic Transition

success = await manager.transition_player(
    player_id=player.id,
    from_world="main",
    to_world="dungeon-1",
    to_room=entrance_room_id,
)

if success:
    print("Player transitioned successfully")
else:
    print("Transition failed")

What Happens During Transition

  1. Player entity is copied from source world
  2. All components are deep-copied
  3. Inventory items are transferred with the player
  4. Player is removed from source world
  5. Player is created in target world with same ID
  6. Components and inventory are restored
  7. Player is placed in target room

Handling Transitions in Commands

def _current_world_id(manager: WorldManager, world: World) -> str | None:
    """Resolve the manager ID for a World object.

    `World` has no `world_id` attribute, so a command finds its current world
    by matching identity against the manager's registered instances.
    """
    for instance in manager.list_worlds(include_private=True):
        if instance.world is world:
            return instance.id
    return None


async def enter_portal_handler(ctx: CommandContext) -> bool:
    """Enter a portal.

    Command handlers receive a ``CommandContext`` and return ``bool``; there is
    no ``CommandResult`` type. Output is sent via ``ctx.session.send()``.
    """
    current_world_id = _current_world_id(world_manager, ctx.world)
    if current_world_id is None:
        await ctx.session.send("You are adrift between worlds.\n")
        return True

    room_id = ctx.world.get_entity_room(ctx.player_id)
    if room_id is None:
        await ctx.session.send("There is no portal here.\n")
        return True

    # Find a portal in the current room
    portals = world_manager.get_portals_in_room(current_world_id, room_id)
    if not portals:
        await ctx.session.send("There is no portal here.\n")
        return True

    portal = portals[0]  # Use first portal

    # Validate the destination BEFORE moving. transition_player() checks that
    # both worlds and the player entity exist, but it does NOT verify that
    # to_room exists in the target world — it just calls
    # place_entity_in_room(to_room). A stale or mis-configured portal would
    # therefore strand the player in a nonexistent room (look/movement break).
    # Guard against that here.
    target_world = world_manager.get_world(portal.target_world_id)
    if target_world is None or target_world.get_room(portal.target_room_id) is None:
        await ctx.session.send("The portal leads nowhere.\n")
        return True

    # Transition the player between worlds
    success = await world_manager.transition_player(
        player_id=ctx.player_id,
        from_world=current_world_id,
        to_world=portal.target_world_id,
        to_room=portal.target_room_id,
    )

    if success:
        await ctx.session.send(f"You step through {portal.name or 'the portal'}.\n")
    else:
        await ctx.session.send("The portal flickers but nothing happens.\n")
    return True

World Lifecycle

Starting All Worlds

# Start up all worlds
await manager.startup_all()

# Or start individually
for instance in manager.list_worlds(include_private=True):
    await instance.world.startup()

Ticking All Worlds

# Tick all worlds
await manager.tick_all(delta)

# Or manually iterate
for instance in manager.list_worlds(include_private=True):
    await instance.world.tick(delta)

Shutting Down

# Shutdown all worlds
await manager.shutdown_all()

Use Cases

Instanced Dungeons

Create private instances per party:

async def create_dungeon_instance(party_leader_id: UUID) -> str:
    """Create a private dungeon instance for a party."""
    instance_id = f"dungeon-{party_leader_id}"

    # Create the instance
    world = manager.create_world(
        instance_id,
        settings,
        display_name="Private Dungeon",
        max_players=5,
        is_public=False,
    )

    # Set up dungeon content
    await populate_dungeon(world)

    # Boot the freshly created world. create_world does NOT start it — only
    # startup_all() starts the worlds that exist at boot — so a world created
    # dynamically at runtime must be started explicitly after it is populated,
    # or its systems' startup() hooks (event subscriptions, timers, etc.) never
    # run and it will not tick correctly.
    await world.startup()

    # Create entry portal
    main = manager.get_world("main")
    manager.create_portal(
        source_world="main",
        source_room=dungeon_entrance_room,
        target_world=instance_id,
        target_room=dungeon_start_room,
        bidirectional=True,
    )

    return instance_id

async def cleanup_dungeon_instance(instance_id: str) -> None:
    """Clean up a dungeon instance when party leaves."""
    # Check if empty
    world = manager.get_world(instance_id)
    if world and not get_players_in_world(world):
        # remove_world() only drops the world and its portals from the manager;
        # it does NOT shut the world down. Shut it down first so its systems can
        # release resources (background tasks, timers, pending saves) cleanly.
        await world.shutdown()
        manager.remove_world(instance_id)

Player Housing

Each player gets their own world:

async def create_player_house(player_id: UUID) -> str:
    """Create a personal house world for a player."""
    house_id = f"house-{player_id}"

    world = manager.create_world(
        house_id,
        settings,
        display_name=f"Player's House",
        max_players=20,  # Allow guests
        is_public=False,
    )

    # Create house rooms
    await setup_house_rooms(world)

    # Start the dynamically created world so its systems boot (see note above).
    await world.startup()

    # Create door from main world at the player's current room
    house_location = main_world.get_entity_room(player_id)

    manager.create_portal(
        source_world="main",
        source_room=house_location,
        target_world=house_id,
        target_room=house_entrance_room,
        bidirectional=True,
        name="House Door",
    )

    return house_id

PvP Arenas

Isolated combat zones:

async def start_arena_match(player_a: UUID, player_b: UUID) -> str:
    """Create an arena instance for a PvP match."""
    arena_id = f"arena-{uuid4()}"

    world = manager.create_world(
        arena_id,
        settings,
        display_name="PvP Arena",
        max_players=2,
        is_public=False,
    )
    # create_world has no `metadata` parameter; set it on the WorldInstance.
    instance = manager.get_instance(arena_id)
    if instance is not None:
        instance.metadata.update(
            {
                "type": "pvp",
                "participants": [str(player_a), str(player_b)],
            }
        )

    await setup_arena(world)

    # Start the arena world before moving players in, so its systems are
    # running when the combatants arrive (see the dungeon-instance note above).
    await world.startup()

    # Teleport both players
    await manager.transition_player(player_a, "main", arena_id, spawn_a)
    await manager.transition_player(player_b, "main", arena_id, spawn_b)

    return arena_id

Events

WorldManager.transition_player() performs the move but does not emit a built-in event — there is no WorldTransitionEvent in the engine. If you want transition notifications, define your own event and emit it from a thin wrapper around transition_player():

from dataclasses import dataclass
from uuid import UUID

from maid_engine.core.events import Event


@dataclass
class WorldTransitionEvent(Event):
    """Custom event you define in your own content pack."""

    player_id: UUID
    from_world_id: str
    to_world_id: str


async def transition_and_notify(
    manager: WorldManager,
    world: World,
    player_id: UUID,
    from_world: str,
    to_world: str,
    to_room: UUID,
) -> bool:
    """Transition a player and emit a custom event on success."""
    ok = await manager.transition_player(player_id, from_world, to_world, to_room)
    if ok:
        # Emit on whichever world's EventBus you want handlers to observe.
        await world.events.emit(
            WorldTransitionEvent(
                player_id=player_id,
                from_world_id=from_world,
                to_world_id=to_world,
            )
        )
    return ok

Subscribe with the EventBus.subscribe(event_type, handler) method (it is a method call that returns a handler ID, not a decorator):

async def on_world_transition(event: WorldTransitionEvent) -> None:
    """Handle player moving between worlds."""
    print(f"Player {event.player_id} transitioned:")
    print(f"  From: {event.from_world_id}")
    print(f"  To: {event.to_world_id}")

    # Maybe notify other players
    await broadcast_departure(event.from_world_id, event.player_id)
    await broadcast_arrival(event.to_world_id, event.player_id)


world.events.subscribe(WorldTransitionEvent, on_world_transition)

Best Practices

1. Clean Up Unused Instances

async def cleanup_empty_instances(manager: WorldManager) -> int:
    """Remove instances with no players."""
    removed = 0

    for instance in manager.list_worlds(include_private=True):
        # Skip main world
        if instance.id == "main":
            continue

        # Check if empty and not recently created
        player_count = count_players(instance.world)
        if player_count == 0:
            # Shut the world down before removing it — remove_world() does not
            # call World.shutdown(), so skipping this leaks the world's systems.
            await instance.world.shutdown()
            manager.remove_world(instance.id)
            removed += 1

    return removed

2. Validate Transitions

async def safe_transition(
    manager: WorldManager,
    player_id: UUID,
    target_world_id: str,
    target_room_id: UUID,
) -> tuple[bool, str]:
    """Transition with validation."""
    # Check target world exists
    target = manager.get_world(target_world_id)
    if not target:
        return False, "Target world does not exist."

    # Check player limit
    instance = manager.get_instance(target_world_id)
    if instance.max_players > 0:
        current = count_players(target)
        if current >= instance.max_players:
            return False, "Target world is full."

    # Check target room exists
    if not target.get_room(target_room_id):
        return False, "Target room does not exist."

    # Find current world
    from_world_id = find_player_world(manager, player_id)
    if not from_world_id:
        return False, "Player is not in any world."

    # Perform transition
    success = await manager.transition_player(
        player_id,
        from_world_id,
        target_world_id,
        target_room_id,
    )

    return success, "Transitioned successfully." if success else "Transition failed."

3. Preserve Player State

Components are automatically deep-copied by transition_player(), but ensure external state is handled. Register this handler via world.events.subscribe(WorldTransitionEvent, on_transition) using the custom WorldTransitionEvent defined in the Events section (the engine has no built-in transition event):

async def on_transition(event: WorldTransitionEvent) -> None:
    """Handle external state during transition."""
    # Clear combat state
    combat_manager.remove_from_combat(event.player_id)

    # Save any pending data
    await save_player_data(event.player_id)

    # Update session world reference
    session = get_session(event.player_id)
    if session:
        session.current_world = event.to_world_id

4. Use Metadata

# Store instance-specific data. create_world has no `metadata` parameter, so
# set it on the WorldInstance after creation.
world = manager.create_world("event-dungeon", settings)
instance = manager.get_instance("event-dungeon")
if instance is not None:
    instance.metadata.update(
        {
            "event_type": "winter_festival",
            "difficulty": "hard",
            "rewards_claimed": [],
            "created_at": datetime.now().isoformat(),
        }
    )

# Access later
instance = manager.get_instance("event-dungeon")
if instance.metadata.get("event_type") == "winter_festival":
    apply_winter_theme(instance.world)

Next Steps