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
- Basic Arguments
- Pattern-Based Arguments
- Argument Types
- Entity Resolution
- Hook System
- Pre-Command Hooks
- Post-Command Hooks
- Built-in Hooks
- Custom Hooks
- Lock System
- Lock Expressions
- Built-in Lock Functions
- Custom Lock Functions
- Migration Guide
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