Skip to content

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

async def quit_handler(ctx: CommandContext) -> CommandResult:
    """Handle the quit command."""
    return CommandResult(
        success=True,
        message="Goodbye!",
    )

With Arguments

async def say_handler(ctx: CommandContext) -> CommandResult:
    """Handle the say command."""
    if not ctx.args:
        return CommandResult(
            success=False,
            message="Say what?",
        )

    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}"',
    )

    return CommandResult(
        success=True,
        message=f'You say, "{message}"',
    )

Command Context

The CommandContext provides all necessary information:

@dataclass
class CommandContext:
    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
    session: Session | None  # Network session (optional)

    # Helper methods
    def get_player(self) -> Entity | None:
        """Get the player entity."""
        return self.world.get_entity(self.player_id)

    def get_room(self) -> Entity | None:
        """Get the player's current room."""
        player = self.get_player()
        if player:
            pos = player.try_get(PositionComponent)
            if pos:
                return self.world.get_room(pos.room_id)
        return None

Using Context Helpers

async def look_handler(ctx: CommandContext) -> CommandResult:
    player = ctx.get_player()
    if not player:
        return CommandResult(False, "Player not found.")

    room = ctx.get_room()
    if not room:
        return CommandResult(False, "You are nowhere.")

    description = format_room_description(room, player)
    return CommandResult(True, description)

Command Result

The CommandResult describes the outcome:

@dataclass
class CommandResult:
    success: bool                     # Did the command succeed?
    message: str = ""                 # Message to show player
    data: dict[str, Any] | None = None  # Optional additional data

Success Result

return CommandResult(
    success=True,
    message="You picked up the sword.",
    data={"item_id": str(sword.id)},
)

Failure Result

return CommandResult(
    success=False,
    message="You can't pick that up.",
)

Multi-line Messages

lines = [
    "Room: Town Square",
    "",
    "You are in the center of town.",
    "A fountain bubbles nearby.",
    "",
    "Exits: north, south, east",
]
return CommandResult(True, "\n".join(lines))

Handler Patterns

Target Resolution

Find a target from player input:

async def attack_handler(ctx: CommandContext) -> CommandResult:
    if not ctx.args:
        return CommandResult(False, "Attack what?")

    target_name = ctx.args[0].lower()
    player = ctx.get_player()
    room = ctx.get_room()

    # 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:
        return CommandResult(
            False,
            f"You don't see '{ctx.args[0]}' here.",
        )

    # Perform attack
    await perform_attack(player, target)
    return CommandResult(True, f"You attack {get_name(target)}!")


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.get_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) -> CommandResult:
    """Pick up an item."""
    if not ctx.args:
        return CommandResult(False, "Get what?")

    item_name = ctx.args[0].lower()
    player = ctx.get_player()
    room = ctx.get_room()

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

    # Check if player can carry
    inventory = player.try_get(InventoryComponent)
    if not inventory:
        return CommandResult(False, "You can't carry items.")

    if len(inventory.items) >= inventory.max_slots:
        return CommandResult(False, "Your inventory is full.")

    # Transfer item
    inventory.items.append(item.id)
    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
    return CommandResult(True, f"You pick up {item_name}.")

Movement

async def move_handler(ctx: CommandContext) -> CommandResult:
    """Move in a direction."""
    direction = ctx.command  # "north", "south", etc.
    player = ctx.get_player()
    room = ctx.get_room()

    # Get exits
    exits = room.try_get(ExitsComponent)
    if not exits or direction not in exits.exits:
        return CommandResult(False, "You can't go that way.")

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

    # 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)
    return CommandResult(True, description)

Complex Commands

Handle commands with sub-commands or complex syntax:

async def equipment_handler(ctx: CommandContext) -> CommandResult:
    """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:
            return CommandResult(False, "Equip what?")
        return await equip_item(ctx, " ".join(ctx.args[1:]))

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

    else:
        return CommandResult(
            False,
            f"Unknown sub-command: {sub_command}\n"
            "Usage: equipment [equip <item>|remove <slot>]",
        )

Permission Checks

async def admin_kick_handler(ctx: CommandContext) -> CommandResult:
    """Kick a player (admin only)."""
    player = ctx.get_player()

    # Check permissions
    if not player.has_tag("admin"):
        return CommandResult(
            False,
            "You don't have permission to use this command.",
        )

    if not ctx.args:
        return CommandResult(False, "Kick who?")

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

    if not target:
        return CommandResult(False, f"Player '{target_name}' not found.")

    if target.has_tag("admin"):
        return CommandResult(False, "You can't kick an admin.")

    await kick_player(target)
    return CommandResult(True, f"Kicked {target_name}.")

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) -> CommandResult:
        """Handle attack command."""
        ...

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

    async def defend(self, ctx: CommandContext) -> CommandResult:
        """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) -> CommandResult:
    try:
        player = ctx.get_player()
        if not player:
            return CommandResult(False, "Error: Player not found.")

        # Command logic...

        return CommandResult(True, "Success!")

    except EntityNotFoundException as e:
        return CommandResult(False, str(e))

    except PermissionError as e:
        return CommandResult(False, f"Permission denied: {e}")

    except Exception as e:
        logger.exception(f"Error in command: {e}")
        return CommandResult(
            False,
            "An unexpected error occurred. Please try again.",
        )

Validation Helper

def validate_args(
    ctx: CommandContext,
    min_args: int = 0,
    max_args: int | None = None,
    usage: str = "",
) -> CommandResult | None:
    """Validate argument count. Returns error result or None if valid."""
    if len(ctx.args) < min_args:
        return CommandResult(
            False,
            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 CommandResult(
            False,
            f"Too many arguments.\nUsage: {usage}" if usage else "Too many arguments.",
        )

    return None


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

    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(
        player_id=UUID("..."),
        command="look",
        args=[],
        raw_input="look",
        world=world,
    )

    # Execute
    result = await look_handler(ctx)

    # Assert
    assert result.success
    assert "Test Room" in result.message

Integration Tests

@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(
        player_id=player.id,
        command="north",
        args=[],
        raw_input="north",
        world=world,
    )
    result = await move_handler(ctx)

    # Assert
    assert result.success
    assert player.get(PositionComponent).room_id == room_b.id

Best Practices

1. Always Validate Input

async def handler(ctx: CommandContext) -> CommandResult:
    if not ctx.args:
        return CommandResult(False, "Missing argument.")
    # Validated, proceed...

2. Provide Helpful Error Messages

# Bad
return CommandResult(False, "Error")

# Good
return CommandResult(
    False,
    "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) -> CommandResult:
    # Just handles dropping an item
    ...

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

5. Document Complex Commands

async def trade_handler(ctx: CommandContext) -> CommandResult:
    """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