Skip to content

Advanced Command System Guide

This guide covers MAID's advanced command system features for building robust, secure, and well-documented commands. The command system provides:

  • Declarative argument parsing with type validation
  • Pattern-based argument matching for complex command formats
  • Pre/post command hooks for cross-cutting concerns
  • Expression-based permission locks for fine-grained access control

Table of Contents


Argument Parsing

The argument parsing system allows you to declare expected arguments for commands and have them automatically parsed, validated, and converted to appropriate types.

Basic Arguments

Use the @arguments decorator for positional argument parsing. Arguments are parsed in order from the input string.

from maid_engine.commands import (
    arguments,
    command,
    ArgumentSpec,
    ArgumentType,
    ParsedArguments,
    CommandContext,
)

@command(
    name="attack",
    aliases=["att", "a"],
    category="combat",
    help_text="Attack a target multiple times",
)
@arguments(
    ArgumentSpec("target", ArgumentType.ENTITY, description="The entity to attack"),
    ArgumentSpec(
        "count",
        ArgumentType.INTEGER,
        required=False,
        default=1,
        min_value=1,
        max_value=10,
        description="Number of attacks",
    ),
)
async def cmd_attack(ctx: CommandContext, args: ParsedArguments) -> bool:
    """Attack a target multiple times."""
    target = args["target"]  # EntityReference
    count = args["count"]    # int (1-10)

    if not target.found:
        await ctx.session.send("You don't see that here.")
        return False

    for _ in range(count):
        # Perform attack logic
        pass

    await ctx.session.send(f"You attack {target.entity} {count} time(s)!")
    return True

The @arguments decorator: - Parses ctx.args using the argument specs - Passes ParsedArguments as the second argument to your handler - Automatically sends error messages to the player if parsing fails - Returns False if parsing fails, preventing command execution

Pattern-Based Arguments

Use the @pattern decorator for complex argument formats with named placeholders. This is useful for commands like "give X to Y" or "put X in Y".

from maid_engine.commands import (
    pattern,
    command,
    ArgumentSpec,
    ArgumentType,
    SearchScope,
    ParsedArguments,
    CommandContext,
)

@command(name="give", category="items", help_text="Give an item to someone")
@pattern(
    "<item> to <target>",
    item=ArgumentSpec(
        "item",
        ArgumentType.ENTITY,
        search_scope=SearchScope.INVENTORY,
        description="Item from your inventory",
    ),
    target=ArgumentSpec(
        "target",
        ArgumentType.ENTITY,
        search_scope=SearchScope.ROOM,
        description="Entity to give item to",
    ),
)
async def cmd_give(ctx: CommandContext, args: ParsedArguments) -> bool:
    """Give an item to someone."""
    item = args["item"]
    target = args["target"]

    if not item.found:
        await ctx.session.send("You don't have that item.")
        return False

    if not target.found:
        await ctx.session.send("You don't see them here.")
        return False

    # Transfer item logic
    await ctx.session.send(f"You give {item.entity} to {target.entity}.")
    return True

Pattern syntax: - <name> placeholders map to argument specs - Literal text between placeholders must match exactly - Whitespace is normalized (multiple spaces treated as one)

More pattern examples:

# "put <item> in <container>"
@pattern(
    "<item> in <container>",
    item=ArgumentSpec("item", ArgumentType.ENTITY, search_scope=SearchScope.INVENTORY),
    container=ArgumentSpec("container", ArgumentType.ENTITY, search_scope=SearchScope.ROOM),
)

# "say <message> to <target>"
@pattern(
    "<message> to <target>",
    message=ArgumentSpec("message", ArgumentType.REST),
    target=ArgumentSpec("target", ArgumentType.PLAYER),
)

# "cast <spell> on <target>"
@pattern(
    "<spell> on <target>",
    spell=ArgumentSpec("spell", ArgumentType.STRING, choices=["fireball", "heal", "shield"]),
    target=ArgumentSpec("target", ArgumentType.ENTITY),
)

Argument Types

The ArgumentType enum defines available argument types:

Type Description Parsed Value Validation Options
STRING Raw string value str choices, pattern (regex)
INTEGER Integer number int min_value, max_value
FLOAT Decimal number float min_value, max_value
BOOLEAN Boolean flag bool Accepts: true/false, yes/no, on/off, 1/0
ENTITY Single entity reference EntityReference search_scope
ENTITY_LIST Multiple entities EntityReference search_scope (for "all.sword")
DIRECTION Cardinal direction str (normalized) Accepts: n, north, ne, up, etc.
PLAYER Online player Entity Searches online players
REST Remaining input str Consumes all remaining tokens

ArgumentSpec Options

ArgumentSpec(
    name="target",           # Argument name (required)
    type=ArgumentType.ENTITY,  # Argument type (default: STRING)
    required=True,           # Whether argument is required (default: True)
    default=None,            # Default value if not provided
    description="",          # Help text description

    # STRING-specific
    choices=["a", "b", "c"],  # Valid choices
    pattern=r"^\w+$",         # Regex pattern to match

    # INTEGER/FLOAT-specific
    min_value=1,             # Minimum value
    max_value=100,           # Maximum value

    # ENTITY-specific
    search_scope=SearchScope.ROOM_AND_INVENTORY,  # Where to search
)

Entity Resolution

Entity resolution supports several special syntaxes:

Syntax Description Example
me Player's own entity examine me
here Current room look here
<keyword> First matching entity get sword
2.<keyword> Nth matching entity get 2.sword
all.<keyword> All matching entities drop all.sword

Search scopes control where entities are searched:

class SearchScope(Enum):
    ROOM = "room"                      # Only in current room
    INVENTORY = "inventory"            # Only in player's inventory
    EQUIPMENT = "equipment"            # Only equipped items
    ROOM_AND_INVENTORY = "room_and_inventory"  # Both (default)
    ALL = "all"                        # Room, inventory, and equipment

Using EntityReference:

target = args["target"]  # EntityReference object

# Check if anything was found
if not target.found:
    await ctx.session.send("Not found.")
    return False

# Get single entity (first match)
entity = target.entity

# Iterate all matches (for ENTITY_LIST or "all.X")
for entity in target:
    print(entity.id)

# Get count
count = len(target)

# Access original keyword
print(f"Searched for: {target.keyword}")

Hook System

Hooks allow you to intercept command execution before and after the handler runs. This enables cross-cutting concerns like:

  • Logging and auditing
  • Rate limiting and cooldowns
  • Permission checks beyond locks
  • Metrics collection
  • Event broadcasting

Pre-Command Hooks

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

from maid_engine.commands import (
    PreHookContext,
    HookResult,
    HookPriority,
    LayeredCommandRegistry,
)

async def my_pre_hook(ctx: PreHookContext) -> HookResult:
    """Example pre-hook that checks a condition."""
    # Access hook context
    print(f"Command: {ctx.command_name}")
    print(f"Category: {ctx.command_category}")
    print(f"Args: {ctx.args}")
    print(f"Player: {ctx.player_entity_id}")
    print(f"Metadata: {ctx.metadata}")

    # Check some condition
    if some_condition_fails:
        ctx.cancel("Sorry, you can't do that right now.")
        return HookResult.CANCEL

    return HookResult.CONTINUE

# Register the hook
registry = LayeredCommandRegistry()
registry.register_pre_hook(
    name="my_checker",
    handler=my_pre_hook,
    priority=HookPriority.NORMAL,
    commands=["attack", "cast"],  # Only these commands (None = all)
    categories=["combat"],         # Only this category (None = all)
)

Hook results: - HookResult.CONTINUE - Proceed to next hook and command - HookResult.CANCEL - Stop execution, send cancel message - HookResult.SKIP - Skip remaining pre-hooks, run command

Post-Command Hooks

Post-hooks run after the command handler completes (even if it raised an exception).

from maid_engine.commands import PostHookContext

async def my_post_hook(ctx: PostHookContext) -> None:
    """Example post-hook for logging."""
    print(f"Command: {ctx.command_name}")
    print(f"Success: {ctx.success}")
    print(f"Result: {ctx.result}")
    print(f"Exception: {ctx.exception}")
    print(f"Time: {ctx.execution_time_ms}ms")

# Register the hook
registry.register_post_hook(
    name="my_logger",
    handler=my_post_hook,
    priority=HookPriority.LAST,
)

Post-hooks cannot affect command results but can perform cleanup, logging, or event emission.

Built-in Hooks

MAID provides several built-in hooks:

Logging Hooks

from maid_engine.commands import (
    logging_pre_hook,
    logging_post_hook,
    HookPriority,
)

# Log all command attempts
registry.register_pre_hook(
    "command_logger_pre",
    logging_pre_hook,
    priority=HookPriority.FIRST,
)

# Log command results with timing
registry.register_post_hook(
    "command_logger_post",
    logging_post_hook,
    priority=HookPriority.LAST,
)

CooldownHook

Enforces command cooldowns based on metadata:

from maid_engine.commands import CooldownHook, HookPriority

cooldown_hook = CooldownHook()

registry.register_pre_hook(
    "cooldown",
    cooldown_hook,
    priority=HookPriority.HIGH,
)

# Commands define cooldowns in metadata
@command(
    name="teleport",
    metadata={"cooldown": 60},  # 60 second cooldown
)
async def cmd_teleport(ctx: CommandContext) -> bool:
    # ...
    pass

CooldownHook methods: - record_usage(player_id, command_name) - Record command usage - reset_cooldown(player_id, command_name) - Clear a cooldown - clear_player_cooldowns(player_id) - Clear all cooldowns for player

CombatStateHook

Blocks certain commands during combat:

from maid_engine.commands import CombatStateHook

combat_hook = CombatStateHook(
    blocked_commands=["quit", "teleport", "logout"],
    blocked_categories=["crafting", "social"],
)

registry.register_pre_hook(
    "combat_block",
    combat_hook,
    priority=HookPriority.HIGH,
)

MetricsHook

Collects command execution statistics:

from maid_engine.commands import MetricsHook

metrics_hook = MetricsHook()

registry.register_post_hook(
    "metrics",
    metrics_hook,
    priority=HookPriority.LAST,
)

# Later, get statistics
stats = metrics_hook.get_stats("attack")
if stats:
    print(f"Executions: {stats.execution_count}")
    print(f"Success rate: {stats.success_count / stats.execution_count}")
    print(f"Avg time: {stats.avg_time_ms}ms")
    print(f"Min/Max: {stats.min_time_ms}ms / {stats.max_time_ms}ms")

# Get all stats
all_stats = metrics_hook.get_all_stats()

Custom Hooks

Create custom hooks as sync or async functions:

from maid_engine.commands import (
    PreHookContext,
    PostHookContext,
    HookResult,
    HookPriority,
)

# Sync pre-hook
def sync_pre_hook(ctx: PreHookContext) -> HookResult:
    if ctx.player_entity_id is None:
        ctx.cancel("You must be logged in.")
        return HookResult.CANCEL
    return HookResult.CONTINUE

# Async pre-hook
async def async_pre_hook(ctx: PreHookContext) -> HookResult:
    # Can await async operations
    is_banned = await check_ban_status(ctx.player_entity_id)
    if is_banned:
        ctx.cancel("You are banned from using commands.")
        return HookResult.CANCEL
    return HookResult.CONTINUE

# Sync post-hook
def sync_post_hook(ctx: PostHookContext) -> None:
    log_to_database(ctx.command_name, ctx.success)

# Async post-hook
async def async_post_hook(ctx: PostHookContext) -> None:
    await broadcast_event(ctx.command_name, ctx.player_entity_id)

# Register all
registry.register_pre_hook("sync_check", sync_pre_hook, priority=HookPriority.FIRST)
registry.register_pre_hook("async_check", async_pre_hook, priority=HookPriority.HIGH)
registry.register_post_hook("sync_log", sync_post_hook, priority=HookPriority.NORMAL)
registry.register_post_hook("async_broadcast", async_post_hook, priority=HookPriority.LOW)

Hook priority levels (lower values run first):

Priority Value Use Case
FIRST 0 Rate limiting, authentication
HIGH 25 Permission checks, cooldowns
NORMAL 50 Default priority
LOW 75 Less critical checks
LAST 100 Logging, metrics

Lock System

Locks provide permission-based access control using expression syntax. They allow complex permission checks without writing custom code.

Lock Expressions

Lock expressions use a simple language with functions, boolean operators, and parentheses.

@command(
    name="admin_command",
    locks="perm(admin)",  # Simple permission check
)

@command(
    name="builder_tool",
    locks="perm(builder) OR perm(admin)",  # Either permission
)

@command(
    name="guild_ability",
    locks="level(10) AND in_guild(warriors)",  # Both conditions
)

@command(
    name="safe_zone_only",
    locks="NOT in_combat() AND in_room(safe_zone)",  # Negation
)

@command(
    name="owner_or_admin",
    locks="perm(admin) OR (owns() AND level(20))",  # Nested conditions
)

Expression grammar:

expr     := or_expr
or_expr  := and_expr (OR and_expr)*
and_expr := not_expr (AND not_expr)*
not_expr := NOT not_expr | primary
primary  := function | '(' expr ')' | TRUE | FALSE
function := name '(' args? ')'

Boolean operators: - AND (or &&) - Both conditions must be true - OR (or ||) - Either condition must be true - NOT (or !) - Negates the condition

Operators support short-circuit evaluation for efficiency.

Built-in Lock Functions

Function Description Usage
all() Always true all()
none() Always false none()
perm(level) Check access level perm(admin), perm(builder)
owns() Player owns target owns()
holds(item?) Player holds item holds(), holds(sword)
in_room(tag) Room has tag in_room(safe_zone)
has_flag(flag) Player has tag has_flag(vip)
in_guild(name?) In guild in_guild(), in_guild(warriors)
guild_rank(rank) Check guild rank guild_rank(officer)
level(n) Character level >= n level(10)
in_combat() Player in combat in_combat()
has_item(keyword) Has item in inventory has_item(key)
has_skill(name, level?) Has skill at level has_skill(lockpicking), has_skill(lockpicking, 5)

Permission levels for perm(): - player (0) - Default access - builder (50) - Building tools - admin (100) - Administrative commands - superuser (999) - Full access

Custom Lock Functions

Register custom lock functions for game-specific checks:

from maid_engine.commands import LockContext, LayeredCommandRegistry

def lock_has_gold(ctx: LockContext, args: list[str]) -> bool:
    """Check if player has enough gold. Usage: has_gold(100)"""
    if not args:
        return False

    try:
        required = int(args[0])
    except ValueError:
        return False

    player = ctx.get_player()
    if not player:
        return False

    # Check gold component
    for component in player.components:
        if hasattr(component, "gold"):
            return component.gold >= required

    return False

def lock_is_day(ctx: LockContext, args: list[str]) -> bool:
    """Check if it's daytime in-game. Usage: is_day()"""
    if not ctx.world:
        return False

    time_system = ctx.world.get_system("TimeSystem")
    if time_system:
        return time_system.is_day()
    return True  # Default to day if no time system

def lock_faction_rep(ctx: LockContext, args: list[str]) -> bool:
    """Check faction reputation. Usage: faction_rep(guild_name, 50)"""
    if len(args) < 2:
        return False

    faction_name = args[0]
    try:
        required_rep = int(args[1])
    except ValueError:
        return False

    player = ctx.get_player()
    if not player:
        return False

    # Check reputation component
    for component in player.components:
        if hasattr(component, "reputations"):
            rep = component.reputations.get(faction_name, 0)
            return rep >= required_rep

    return False

# Register custom functions
registry = LayeredCommandRegistry()
registry.register_lock_function("has_gold", lock_has_gold)
registry.register_lock_function("is_day", lock_is_day)
registry.register_lock_function("faction_rep", lock_faction_rep)

# Now use in commands
@command(
    name="buy_mount",
    locks="has_gold(500) AND level(10)",
)
async def cmd_buy_mount(ctx: CommandContext) -> bool:
    # ...
    pass

@command(
    name="night_hunt",
    locks="NOT is_day() AND has_skill(stealth)",
)
async def cmd_night_hunt(ctx: CommandContext) -> bool:
    # ...
    pass

@command(
    name="guild_summon",
    locks="faction_rep(mages_guild, 100) OR perm(admin)",
)
async def cmd_guild_summon(ctx: CommandContext) -> bool:
    # ...
    pass

LockContext provides: - player_entity_id - UUID of the player - world - Reference to the game world - target - Optional target entity being accessed - command_name - Name of the command being checked - session - Player session object - get_player() - Helper to get player entity


Migration Guide

Migrating from Simple Handlers

Before (manual parsing):

async def cmd_attack(ctx: CommandContext) -> bool:
    if not ctx.args:
        await ctx.session.send("Attack what?")
        return False

    target_name = ctx.args[0]
    count = 1
    if len(ctx.args) > 1:
        try:
            count = int(ctx.args[1])
        except ValueError:
            await ctx.session.send("Invalid count.")
            return False

    # Find target manually...
    target = find_entity(ctx.world, target_name)
    if not target:
        await ctx.session.send("Not found.")
        return False

    # Attack logic...
    return True

After (declarative parsing):

@command(name="attack", category="combat")
@arguments(
    ArgumentSpec("target", ArgumentType.ENTITY),
    ArgumentSpec("count", ArgumentType.INTEGER, required=False, default=1),
)
async def cmd_attack(ctx: CommandContext, args: ParsedArguments) -> bool:
    target = args["target"]  # Already resolved
    count = args["count"]    # Already validated

    if not target.found:
        await ctx.session.send("Not found.")
        return False

    # Attack logic...
    return True

Migrating Permission Checks

Before (manual checks):

async def cmd_admin_action(ctx: CommandContext) -> bool:
    player = ctx.world.get_entity(ctx.player_id)
    access = player.get("AccessComponent")
    if not access or access.level < 100:
        await ctx.session.send("Permission denied.")
        return False

    # Admin action...
    return True

After (lock expression):

@command(
    name="admin_action",
    locks="perm(admin)",
)
async def cmd_admin_action(ctx: CommandContext) -> bool:
    # Permission already checked
    # Admin action...
    return True

Adding Hooks to Existing Commands

If you have existing logging or rate-limiting code scattered across commands, centralize it with hooks:

Before (duplicated in each command):

async def cmd_teleport(ctx: CommandContext) -> bool:
    # Check cooldown
    last_use = get_cooldown(ctx.player_id, "teleport")
    if time.time() - last_use < 60:
        await ctx.session.send("On cooldown.")
        return False

    # Log command
    logger.info(f"Player {ctx.player_id} used teleport")

    # Actual logic
    # ...

    # Record cooldown
    set_cooldown(ctx.player_id, "teleport", time.time())
    return True

After (centralized hooks):

# Setup once
cooldown = CooldownHook()
registry.register_pre_hook("cooldown", cooldown, priority=HookPriority.HIGH)
registry.register_pre_hook("logger", logging_pre_hook, priority=HookPriority.FIRST)

# In a post-hook
async def record_cooldown(ctx: PostHookContext) -> None:
    if ctx.success and ctx.metadata.get("cooldown"):
        cooldown.record_usage(ctx.player_entity_id, ctx.command_name)

registry.register_post_hook("cooldown_record", record_cooldown)

# Command is now simple
@command(
    name="teleport",
    metadata={"cooldown": 60},
)
async def cmd_teleport(ctx: CommandContext) -> bool:
    # Just the actual logic
    # ...
    return True

See Also