"""
Command executor for MCP-Ables tools
"""
import asyncio
import shlex
from typing import Any, Dict
from jinja2 import Template, TemplateError
from .yaml_parser import ToolSpec, ArgumentSpec
class ExecutionError(Exception):
"""Raised when command execution fails"""
pass
class CommandExecutor:
"""Executes shell commands with templated arguments"""
def __init__(self, tool_spec: ToolSpec):
"""Initialize executor with a tool specification"""
self.tool_spec = tool_spec
self._validate_template()
def _validate_template(self):
"""Validate that the command template is valid Jinja2"""
try:
Template(self.tool_spec.run_cmd)
except TemplateError as e:
raise ValueError(f"Invalid command template: {e}")
async def execute(self, inputs: Dict[str, Any], timeout: int = 30) -> Dict[str, Any]:
"""
Execute the command with provided inputs
Args:
inputs: Dictionary of argument names to values
timeout: Command timeout in seconds (default: 30)
Returns:
Dictionary with 'stdout', 'stderr', 'exit_code'
Raises:
ExecutionError: If execution fails
"""
# Validate and prepare arguments
validated_args = self._prepare_arguments(inputs)
# Render command template
try:
template = Template(self.tool_spec.run_cmd)
rendered_cmd = template.render(**validated_args)
except TemplateError as e:
raise ExecutionError(f"Failed to render command template: {e}")
# Parse command safely
try:
cmd_parts = shlex.split(rendered_cmd)
except ValueError as e:
raise ExecutionError(f"Failed to parse command: {e}")
if not cmd_parts:
raise ExecutionError("Rendered command is empty")
# Execute command
try:
proc = await asyncio.create_subprocess_exec(
*cmd_parts,
stdout=asyncio.subprocess.PIPE,
stderr=asyncio.subprocess.PIPE
)
try:
stdout, stderr = await asyncio.wait_for(
proc.communicate(),
timeout=timeout
)
except asyncio.TimeoutError:
proc.kill()
await proc.wait()
raise ExecutionError(f"Command timed out after {timeout} seconds")
return {
'stdout': stdout.decode('utf-8', errors='replace'),
'stderr': stderr.decode('utf-8', errors='replace'),
'exit_code': proc.returncode,
'command': rendered_cmd
}
except FileNotFoundError:
raise ExecutionError(f"Command not found: {cmd_parts[0]}")
except Exception as e:
raise ExecutionError(f"Execution failed: {e}")
def _prepare_arguments(self, inputs: Dict[str, Any]) -> Dict[str, Any]:
"""
Validate inputs and apply defaults
Args:
inputs: Raw input values from user
Returns:
Validated arguments with defaults applied
Raises:
ExecutionError: If validation fails
"""
validated = {}
# Check for unknown arguments
unknown_args = set(inputs.keys()) - set(self.tool_spec.args.keys())
if unknown_args:
raise ExecutionError(f"Unknown arguments: {', '.join(unknown_args)}")
# Process each argument in the spec
for arg_name, arg_spec in self.tool_spec.args.items():
value = inputs.get(arg_name)
# Handle missing required arguments
if value is None:
if arg_spec.required:
raise ExecutionError(f"Missing required argument: {arg_name}")
elif arg_spec.default is not None:
value = arg_spec.default
else:
# Optional with no default - skip
continue
# Validate and convert type
try:
validated[arg_name] = arg_spec.validate_type(value)
except ValueError as e:
raise ExecutionError(str(e))
return validated
