Implementing Command Handlers

Command handlers are async functions that process player commands. This guide covers patterns and best practices for writing effective handlers.

Basic Handler Structure

Minimal Handler

from maid_engine.commands import CommandContext

async def quit_handler(ctx: CommandContext) -> bool:
    """Handle the quit command.

    Returns True if handled, False if not recognized.
    """
    await send_to_session(ctx.session, "Goodbye!")
    return True

With Arguments

async def say_handler(ctx: CommandContext) -> bool:
    """Handle the say command."""
    if not ctx.args:
        await send_to_session(ctx.session, "Say what?")
        return True

    message = " ".join(ctx.args)
    player = ctx.world.get_entity(ctx.player_id)
    name = player.get(NameComponent).name

    # Broadcast to room
    await broadcast_to_room(
        player.get(PositionComponent).room_id,
        f'{name} says, "{message}"',
    )

    await send_to_session(ctx.session, f'You say, "{message}"')
    return True

Command Context

The CommandContext provides all necessary information:

@dataclass
class CommandContext:
    session: Session         # Network session
    player_id: UUID          # Player executing command
    command: str             # Command name (lowercase)
    args: list[str]          # Parsed arguments
    raw_input: str           # Original input string
    world: World             # Game world reference

Access entities through the world:

# Get the player entity
player = ctx.world.get_entity(ctx.player_id)

# Get the player's current room
pos = player.try_get(PositionComponent)
room = ctx.world.get_room(pos.room_id) if pos else None

Using Context

async def look_handler(ctx: CommandContext) -> bool:
    player = ctx.world.get_entity(ctx.player_id)
    if not player:
        await send_to_session(ctx.session, "Player not found.")
        return True

    pos = player.try_get(PositionComponent)
    room = ctx.world.get_room(pos.room_id) if pos else None
    if not room:
        await send_to_session(ctx.session, "You are nowhere.")
        return True

    description = format_room_description(room, player)
    await send_to_session(ctx.session, description)
    return True

Return Value

Command handlers return bool (returning None is treated as True):

• True — the command was handled (even if the player made an error like wrong arguments)
• False — the command was not handled; the dispatcher reports it as an unknown command

False does not "fall through" to another layer. The LayeredCommandRegistry resolves a command to a single highest-priority handler and runs only that one. If it returns False, the network dispatcher simply prints Unknown command: ... — lower-priority handlers from other packs are not tried. To delegate to a lower layer, invoke it explicitly (see below).

Send feedback to the player directly via the session rather than through a return value:

# Send a message to the player
await send_to_session(ctx.session, "You picked up the sword.")

Success

await send_to_session(ctx.session, "You picked up the sword.")
return True

Failure (bad arguments, but command recognized)

await send_to_session(ctx.session, "You can't pick that up.")
return True  # Still return True — the command was recognized

Not handled

return False  # Dispatcher reports "Unknown command"

Explicit delegation to a lower layer

Because False does not fall through, a higher-priority handler that wants to defer to the previous pack's handler must call it directly. Use get_all_layers(name) to reach the underlying definitions (highest priority first):

async def custom_look(ctx: CommandContext) -> bool:
    if not special_case(ctx):
        # Delegate to the next-lower "look" handler (e.g. stdlib's).
        layers = ctx.world.engine.command_registry.get_all_layers("look")
        for definition in layers:
            if definition.handler is not custom_look:
                return await definition.handler(ctx)
        return False
    ...
    return True

Handler Patterns

Target Resolution

Find a target from player input:

async def attack_handler(ctx: CommandContext) -> bool:
    if not ctx.args:
        await send_to_session(ctx.session, "Attack what?")
        return True

    target_name = ctx.args[0].lower()
    player = ctx.world.get_entity(ctx.player_id)
    pos = player.try_get(PositionComponent)
    room = ctx.world.get_room(pos.room_id) if pos else None

    # Search for target in room
    target = find_entity_in_room(
        ctx.world,
        room.id,
        target_name,
        exclude=[ctx.player_id],  # Can't attack self
    )

    if not target:
        await send_to_session(ctx.session, f"You don't see '{ctx.args[0]}' here.")
        return True

    # Perform attack
    await perform_attack(player, target)
    await send_to_session(ctx.session, f"You attack {get_name(target)}!")
    return True


def find_entity_in_room(
    world: World,
    room_id: UUID,
    search: str,
    exclude: list[UUID] = None,
) -> Entity | None:
    """Find entity by name in a room."""
    exclude = exclude or []

    for entity in world.entities_in_room(room_id):
        if entity.id in exclude:
            continue

        name = entity.try_get(NameComponent)
        if name and search in name.name.lower():
            return entity

        # Check keywords
        keywords = entity.try_get(KeywordsComponent)
        if keywords and search in keywords.keywords:
            return entity

    return None

Inventory Operations

async def get_handler(ctx: CommandContext) -> bool:
    """Pick up an item."""
    if not ctx.args:
        await send_to_session(ctx.session, "Get what?")
        return True

    item_name = ctx.args[0].lower()
    player = ctx.world.get_entity(ctx.player_id)
    pos = player.try_get(PositionComponent)
    room = ctx.world.get_room(pos.room_id) if pos else None

    # Find item in room
    item = find_item_in_room(ctx.world, room.id, item_name)
    if not item:
        await send_to_session(ctx.session, f"You don't see '{ctx.args[0]}' here.")
        return True

    # Check if player can carry
    inventory = player.try_get(InventoryComponent)
    if not inventory:
        await send_to_session(ctx.session, "You can't carry items.")
        return True

    if inventory.is_full:
        await send_to_session(ctx.session, "Your inventory is full.")
        return True

    # Transfer item. add_item() enforces capacity/weight/dedup and calls
    # notify_mutation() so the change is tracked for persistence — unlike a raw
    # `inventory.items.append(...)`, which bypasses dirty tracking.
    if not inventory.add_item(item.id):
        await send_to_session(ctx.session, "You can't carry that.")
        return True
    remove_from_room(item.id, room.id)

    # Emit event
    await ctx.world.events.emit(ItemPickedUpEvent(
        entity_id=ctx.player_id,
        item_id=item.id,
        room_id=room.id,
    ))

    item_name = item.get(NameComponent).name
    await send_to_session(ctx.session, f"You pick up {item_name}.")
    return True

Movement

async def move_handler(ctx: CommandContext) -> bool:
    """Move in a direction."""
    direction = ctx.command  # "north", "south", etc.
    player = ctx.world.get_entity(ctx.player_id)
    pos = player.try_get(PositionComponent)
    room = ctx.world.get_room(pos.room_id) if pos else None

    # Get exits
    exits = room.try_get(ExitsComponent)
    if not exits or direction not in exits.exits:
        await send_to_session(ctx.session, "You can't go that way.")
        return True

    # Get destination room
    dest_room_id = exits.exits[direction]
    dest_room = ctx.world.get_room(dest_room_id)
    if not dest_room:
        await send_to_session(ctx.session, "That exit leads nowhere.")
        return True

    # Leave current room
    await ctx.world.events.emit(RoomLeaveEvent(
        entity_id=ctx.player_id,
        room_id=room.id,
        to_room_id=dest_room_id,
    ))

    # Update position
    pos = player.get(PositionComponent)
    old_room_id = pos.room_id
    pos.room_id = dest_room_id

    # Enter new room
    await ctx.world.events.emit(RoomEnterEvent(
        entity_id=ctx.player_id,
        room_id=dest_room_id,
        from_room_id=old_room_id,
    ))

    # Show new room
    description = format_room_description(dest_room, player)
    await send_to_session(ctx.session, description)
    return True

Complex Commands

Handle commands with sub-commands or complex syntax:

async def equipment_handler(ctx: CommandContext) -> bool:
    """Handle equipment commands.

    Usage:
        equipment - Show equipped items
        equipment equip <item> - Equip an item
        equipment remove <slot> - Remove from slot
    """
    if not ctx.args:
        return await show_equipment(ctx)

    sub_command = ctx.args[0].lower()

    if sub_command == "equip":
        if len(ctx.args) < 2:
            await send_to_session(ctx.session, "Equip what?")
            return True
        return await equip_item(ctx, " ".join(ctx.args[1:]))

    elif sub_command == "remove":
        if len(ctx.args) < 2:
            await send_to_session(ctx.session, "Remove from which slot?")
            return True
        return await remove_equipment(ctx, ctx.args[1])

    else:
        await send_to_session(
            ctx.session,
            f"Unknown sub-command: {sub_command}\n"
            "Usage: equipment [equip <item>|remove <slot>]",
        )
        return True

Permission Checks

async def admin_kick_handler(ctx: CommandContext) -> bool:
    """Kick a player (admin only)."""
    player = ctx.world.get_entity(ctx.player_id)

    # Check permissions
    if not player.has_tag("admin"):
        await send_to_session(ctx.session, "You don't have permission to use this command.")
        return True

    if not ctx.args:
        await send_to_session(ctx.session, "Kick who?")
        return True

    target_name = ctx.args[0]
    target = find_player_by_name(ctx.world, target_name)

    if not target:
        await send_to_session(ctx.session, f"Player '{target_name}' not found.")
        return True

    if target.has_tag("admin"):
        await send_to_session(ctx.session, "You can't kick an admin.")
        return True

    await kick_player(target)
    await send_to_session(ctx.session, f"Kicked {target_name}.")
    return True

Handler Classes

For complex commands, use a handler class:

class CombatCommands:
    """Handler class for combat commands."""

    def __init__(self, world: World):
        self._world = world

    async def attack(self, ctx: CommandContext) -> bool:
        """Handle attack command."""
        ...

    async def flee(self, ctx: CommandContext) -> bool:
        """Handle flee command."""
        ...

    async def defend(self, ctx: CommandContext) -> bool:
        """Handle defend command."""
        ...

    def register(self, registry: CommandRegistry) -> None:
        """Register all combat commands."""
        registry.register("attack", self.attack, aliases=["a"])
        registry.register("flee", self.flee, aliases=["run"])
        registry.register("defend", self.defend, aliases=["block"])

Error Handling

Graceful Errors

async def handler(ctx: CommandContext) -> bool:
    try:
        player = ctx.world.get_entity(ctx.player_id)
        if not player:
            await send_to_session(ctx.session, "Error: Player not found.")
            return True

        # Command logic...

        await send_to_session(ctx.session, "Success!")
        return True

    except EntityNotFoundException as e:
        await send_to_session(ctx.session, str(e))
        return True

    except PermissionError as e:
        await send_to_session(ctx.session, f"Permission denied: {e}")
        return True

    except Exception as e:
        logger.exception(f"Error in command: {e}")
        await send_to_session(ctx.session, "An unexpected error occurred. Please try again.")
        return True

Validation Helper

def validate_args(
    ctx: CommandContext,
    min_args: int = 0,
    max_args: int | None = None,
    usage: str = "",
) -> str | None:
    """Validate argument count. Returns error message or None if valid."""
    if len(ctx.args) < min_args:
        return f"Not enough arguments.\nUsage: {usage}" if usage else "Not enough arguments."

    if max_args is not None and len(ctx.args) > max_args:
        return f"Too many arguments.\nUsage: {usage}" if usage else "Too many arguments."

    return None


async def give_handler(ctx: CommandContext) -> bool:
    """Give an item to another player."""
    error = validate_args(ctx, min_args=2, usage="give <item> <player>")
    if error:
        await send_to_session(ctx.session, error)
        return True

    item_name, target_name = ctx.args[0], ctx.args[1]
    # Continue with validated arguments...

Testing Handlers

Unit Tests

import pytest
from unittest.mock import MagicMock, AsyncMock

@pytest.mark.asyncio
async def test_look_handler():
    # Setup mock world
    world = MagicMock()
    player = MagicMock()
    room = MagicMock()

    world.get_entity.return_value = player
    player.get.return_value = MagicMock(room_id=UUID("..."))
    world.get_room.return_value = room
    room.get.return_value = MagicMock(
        name="Test Room",
        description="A test room.",
    )

    # Create context
    ctx = CommandContext(
        session=AsyncMock(),
        player_id=UUID("..."),
        command="look",
        args=[],
        raw_input="look",
        world=world,
    )

    # Execute
    result = await look_handler(ctx)

    # Assert — handler returns True if it handled the command
    assert result is True

Integration Tests

from unittest.mock import AsyncMock

@pytest.mark.asyncio
async def test_movement():
    # Setup real world
    world = create_test_world()
    player = create_test_player(world)
    room_a = create_test_room(world, "Room A")
    room_b = create_test_room(world, "Room B", {"south": room_a.id})
    room_a.get(ExitsComponent).exits["north"] = room_b.id

    # Place player
    place_entity(player, room_a)

    # Execute command
    ctx = CommandContext(
        session=AsyncMock(),
        player_id=player.id,
        command="north",
        args=[],
        raw_input="north",
        world=world,
    )
    result = await move_handler(ctx)

    # Assert — handler returns True if handled
    assert result is True
    assert player.get(PositionComponent).room_id == room_b.id

Best Practices

1. Always Validate Input

async def handler(ctx: CommandContext) -> bool:
    if not ctx.args:
        await send_to_session(ctx.session, "Missing argument.")
        return True
    # Validated, proceed...

2. Provide Helpful Error Messages

# Bad
await send_to_session(ctx.session, "Error")

# Good
await send_to_session(
    ctx.session,
    "You can't attack yourself.\n"
    "Try: attack <target>",
)

3. Use Events for Side Effects

# Don't directly modify other systems
# Instead, emit events
await ctx.world.events.emit(ItemPickedUpEvent(...))

4. Keep Handlers Focused

# Good - focused handler
async def drop_handler(ctx: CommandContext) -> bool:
    # Just handles dropping an item
    ...

# Bad - handler does too much
async def drop_handler(ctx: CommandContext) -> bool:
    # Drops item
    # Updates inventory
    # Checks for quest completion
    # Awards experience
    # Too much!
    ...

5. Document Complex Commands

async def trade_handler(ctx: CommandContext) -> bool:
    """Initiate or manage a trade.

    Usage:
        trade <player>         - Start trade with player
        trade offer <item>     - Offer an item
        trade accept           - Accept current trade
        trade cancel           - Cancel trade

    Examples:
        trade John
        trade offer sword
        trade accept
    """
    ...

Next Steps

• Command Overview - Understanding the command system
• Command Priority and Layering - Override commands properly