execute_command
Enables execution of terminal commands via the MCP Terminal server, supporting real-time output retrieval and customizable timeout settings for streamlined task automation.
Instructions
Executes a terminal command
Input Schema
| Name | Required | Description | Default |
|---|---|---|---|
| command | Yes | ||
| timeout | No | ||
| wait_for_output | No |
Input Schema (JSON Schema)
{
"properties": {
"command": {
"title": "Command",
"type": "string"
},
"timeout": {
"default": 10,
"title": "Timeout",
"type": "integer"
},
"wait_for_output": {
"default": true,
"title": "Wait For Output",
"type": "boolean"
}
},
"required": [
"command"
],
"title": "execute_commandArguments",
"type": "object"
}
Implementation Reference
- The primary handler function for the 'execute_command' MCP tool. It performs command filtering for security, initializes the controller if needed, executes the command via the controller, and formats the response using the ExecuteCommandResponse model.@mcp.tool(name="execute_command", description="Executes a terminal command") async def execute_command( command: str, wait_for_output: bool = True, timeout: int = 10, ) -> ExecuteCommandResponse: try: # Check if command is allowed is_allowed, reason = self.command_filter.is_command_allowed(command) if not is_allowed: logger.warning( f"Command execution denied: {command}. Reason: {reason}" ) return ExecuteCommandResponse( success=False, error=f"Command not allowed: {reason}", ) # Ensure we have a controller if not self.controller: self._init_controller() # Execute the command result = await self.controller.execute_command( command, wait_for_output, timeout ) # Convert to response model return ExecuteCommandResponse( success=result.get("success", False), output=result.get("output"), error=result.get("error"), return_code=result.get("return_code"), warning=result.get("warning"), ) except Exception as e: logger.error(f"Error executing command: {e}") return ExecuteCommandResponse( success=False, error=f"Error executing command: {str(e)}" )
- Pydantic BaseModel classes defining the input schema (ExecuteCommandRequest) and output schema (ExecuteCommandResponse) for the 'execute_command' tool.class ExecuteCommandRequest(BaseModel): """Request model for executing a terminal command.""" command: str = Field(..., description="The command to execute in the terminal") wait_for_output: bool = Field( True, description="Whether to wait for the command output" ) timeout: int = Field( 10, description="Timeout in seconds for waiting for the command output" ) class ExecuteCommandResponse(BaseModel): """Response model for the executed terminal command.""" success: bool = Field( ..., description="Whether the command execution was successful" ) output: Optional[str] = Field(None, description="The command output if available") error: Optional[str] = Field( None, description="Error message if the command failed" ) return_code: Optional[int] = Field( None, description="The command return code if available" ) warning: Optional[str] = Field(None, description="Warning message if any")
- src/mcp_terminal/server.py:94-104 (registration)Instantiation of TerminalTool and call to register_mcp which registers the 'execute_command' tool (among others) with the FastMCP server instance.terminal_tool = TerminalTool( self.controller_type, whitelist_file=self.whitelist_file, blacklist_file=self.blacklist_file, whitelist_mode=self.whitelist_mode, ) file_tool = FileTool() terminal_tool.register_mcp(self.mcp) file_tool.register_mcp(self.mcp) self.tools["terminal"] = terminal_tool self.tools["file"] = file_tool
- The SubprocessTerminalController's execute_command method, which implements the actual subprocess execution logic called by the tool handler. This is the concrete implementation for cross-platform command execution.async def execute_command( self, command: str, wait_for_output: bool = True, timeout: int = 10 ) -> Dict[str, Any]: """ Execute a command using subprocess. Args: command: The command to execute wait_for_output: Whether to wait for output timeout: Timeout in seconds Returns: A dictionary with the result of the command execution """ try: # Create subprocess process = await asyncio.create_subprocess_shell( command, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, ) if wait_for_output: try: # Wait for the process to complete with timeout stdout, stderr = await asyncio.wait_for( process.communicate(), timeout=timeout ) return { "success": process.returncode == 0, "output": stdout.decode("utf-8", errors="replace"), "error": stderr.decode("utf-8", errors="replace"), "return_code": process.returncode, } except asyncio.TimeoutError: # Kill the process if it times out try: process.kill() except ProcessLookupError: pass return { "success": False, "error": f"Command timed out after {timeout} seconds", } else: # Don't wait for output return { "success": True, "output": "Command sent (output not captured)", } except Exception as e: return { "success": False, "error": f"Error executing command: {str(e)}", }