MCPcopy Create free account
hub / github.com/agentforce314/clawcodex / CommandEngine

Class CommandEngine

src/command_system/engine.py:95–287  ·  view source on GitHub ↗

Command execution engine.

Source from the content-addressed store, hash-verified

93
94@dataclass
95class CommandEngine:
96 """Command execution engine."""
97
98 registry: CommandRegistry
99 workspace_root: Path
100 context: CommandContext
101 _command_hooks: list[Callable[[str, CommandResult], None]] = field(
102 default_factory=list
103 )
104
105 async def execute(
106 self,
107 command_input: str,
108 ) -> CommandResult:
109 """
110 Execute a command.
111
112 Args:
113 command_input: Command input (e.g., "/help args")
114
115 Returns:
116 CommandResult with the execution result
117 """
118 # Parse command and args
119 if not command_input.startswith("/"):
120 return CommandResult.error(
121 "",
122 "Commands must start with '/'",
123 )
124
125 parts = command_input[1:].split(maxsplit=1)
126 command_name = parts[0].strip()
127 args = parts[1].strip() if len(parts) > 1 else ""
128
129 # Get command
130 command = self.registry.get(command_name)
131 if command is None:
132 return CommandResult.error(
133 command_name,
134 f"Unknown command: {command_name}",
135 )
136
137 # Check if command is enabled
138 if not command.is_enabled():
139 return CommandResult.error(
140 command_name,
141 f"Command {command_name} is disabled",
142 )
143
144 # Execute based on type
145 result: CommandResult
146 if command.command_type == CommandType.LOCAL:
147 result = await self._execute_local(command, args)
148 elif command.command_type == CommandType.PROMPT:
149 result = await self._execute_prompt(command, args)
150 elif command.command_type == CommandType.INTERACTIVE:
151 result = await self._execute_interactive(command, args)
152 else:

Calls

no outgoing calls