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:
True— the command was handled (even if the player made an error like wrong arguments)False— the command was not recognized or should fall through to another handler
Send feedback to the player directly via the session rather than through a return value:
Success¶
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 (fall through)¶
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.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) -> 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 len(inventory.items) >= inventory.capacity:
await send_to_session(ctx.session, "Your inventory is full.")
return True
# 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
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(
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¶
@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 — 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