Coverage for gco_mcp/tools/templates.py: 97.18%

1"""Job template management MCP tools (read-only).""" 

2 

3import asyncio 

4 

5import cli_runner 

6from audit import audit_logged 

7from server import mcp 

8 

9 

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

11@audit_logged 

12async def templates_list(region: str | None = None) -> str: 

13 """`gco templates list` — list job templates. 

14 

15 Args: 

16 region: Region to query (any region works). 

17 """ 

18 args = ["templates", "list"] 

19 if region: 

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

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

22 

23 

24@mcp.tool(tags={"safe", "templates"}) 

25@audit_logged 

26async def templates_get(name: str, region: str | None = None) -> str: 

27 """`gco templates get` — fetch a single job template by name. 

28 

29 Args: 

30 name: Template name. 

31 region: Region to query (any region works). 

32 """ 

33 args = ["templates", "get", name] 

34 if region: 

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

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

37 

38 

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

40# Mutating tools (low-risk) 

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

42 

43 

44@mcp.tool(tags={"low-risk", "templates"}) 

45@audit_logged 

46async def templates_create( 

47 name: str, 

48 manifest_path: str, 

49 region: str | None = None, 

50 description: str | None = None, 

51) -> str: 

52 """`gco templates create` — register a new job template from a manifest. 

53 

54 Args: 

55 name: Template name. 

56 manifest_path: Path to the source manifest YAML. 

57 region: Region to use (any region works). 

58 description: Optional human-readable description. 

59 """ 

60 args = ["templates", "create", name, manifest_path] 

61 if region: 

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

63 if description: 

64 args += ["-d", description] 

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

66 

67 

68@mcp.tool(tags={"low-risk", "templates"}) 

69@audit_logged 

70async def templates_run( 

71 template_name: str, 

72 job_name: str, 

73 region: str, 

74 namespace: str | None = None, 

75 params: list[str] | None = None, 

76) -> str: 

77 """`gco templates run` — instantiate and run a job from a stored template. 

78 

79 Args: 

80 template_name: Name of the stored template to instantiate. 

81 job_name: Name for the resulting job. 

82 region: Region in which to run the job. 

83 namespace: Override the Kubernetes namespace (default: gco-jobs). 

84 params: Parameter overrides as ``key=value`` strings (one ``-p`` each). 

85 """ 

86 args = ["templates", "run", template_name, "--name", job_name, "-r", region] 

87 if namespace: 

88 args += ["--namespace", namespace] 

89 for param in params or []: 

90 args += ["-p", param] 

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

92 

93 

94# ============================================================================= 

95# Destructive tools — gated by GCO_ENABLE_DESTRUCTIVE_OPERATIONS 

96# ============================================================================= 

97 

98 

99import contextlib # noqa: E402 

100 

101from feature_flags import FLAG_DESTRUCTIVE_OPERATIONS, is_enabled # noqa: E402 

102 

103 

104async def _ctx_warning(message: str) -> None: 

105 """Emit ``ctx.warning(...)`` from inside a tool body, no-op when no Context.""" 

106 try: 

107 from fastmcp.server.dependencies import get_context 

108 

109 ctx = get_context() 

110 except Exception: 

111 return 

112 with contextlib.suppress(Exception): 

113 await ctx.warning(message) 

114 

115 

116if is_enabled(FLAG_DESTRUCTIVE_OPERATIONS): 

117 

118 @mcp.tool(tags={"destructive", "templates"}) 

119 @audit_logged 

120 async def delete_template(name: str, region: str | None = None) -> str: 

121 """[gated by GCO_ENABLE_DESTRUCTIVE_OPERATIONS] destructive. 

122 

123 `gco templates delete` — delete a job template. 

124 Cannot be undone — the template definition is permanently removed. 

125 

126 Args: 

127 name: Template name. 

128 region: Region to use (any region works). 

129 """ 

130 await _ctx_warning(f"Deleting template {name!r} — this cannot be undone.") 

131 args = ["templates", "delete", name, "-y"] 

132 if region: 

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

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