Coverage for mcp/tools/storage.py: 94%

1"""File storage MCP tools.""" 

2 

3import asyncio 

4 

5import cli_runner 

6from audit import audit_logged 

7from server import mcp 

8 

9 

10@mcp.tool(tags={"safe", "storage"}) 

11@audit_logged 

12def list_storage_contents(region: str, path: str = "/") -> str: 

13 """List contents of shared EFS storage. 

14 

15 Args: 

16 region: AWS region. 

17 path: Directory path to list (default: root). 

18 """ 

19 args = ["files", "ls", "-r", region] 

20 if path != "/": 

21 args.append(path) 

22 return cli_runner._run_cli(*args) 

23 

24 

25@mcp.tool(tags={"safe", "storage"}) 

26@audit_logged 

27def list_file_systems(region: str | None = None) -> str: 

28 """List EFS and FSx file systems. 

29 

30 Args: 

31 region: Specific region, or omit for all. 

32 """ 

33 args = ["files", "list"] 

34 if region: 34 ↛ 35line 34 didn't jump to line 35 because the condition on line 34 was never true

35 args += ["-r", region] 

36 return cli_runner._run_cli(*args) 

37 

38 

39# ============================================================================= 

40# Read-only inspection tools (async) 

41# ============================================================================= 

42 

43 

44@mcp.tool(tags={"safe", "files"}) 

45@audit_logged 

46async def files_get(path: str, region: str) -> str: 

47 """`gco files get` — fetch a single file from EFS. 

48 

49 Args: 

50 path: File path relative to the storage root. 

51 region: AWS region. 

52 """ 

53 return await asyncio.to_thread(cli_runner._run_cli, "files", "get", path, "-r", region) 

54 

55 

56@mcp.tool(tags={"safe", "files"}) 

57@audit_logged 

58async def files_access_points(region: str | None = None) -> str: 

59 """`gco files access-points` — list EFS access points. 

60 

61 Args: 

62 region: AWS region. 

63 """ 

64 args = ["files", "access-points"] 

65 if region: 

66 args += ["-r", region] 

67 return await asyncio.to_thread(cli_runner._run_cli, *args)