Coverage for gco_mcp/tools/webhooks.py: 97.14%

1"""Webhook 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", "webhooks"}) 

11@audit_logged 

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

13 """`gco webhooks list` — list configured webhooks. 

14 

15 Args: 

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

17 """ 

18 args = ["webhooks", "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", "webhooks"}) 

25@audit_logged 

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

27 """`gco webhooks get` — fetch a single webhook by name. 

28 

29 Args: 

30 name: Webhook name. 

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

32 """ 

33 args = ["webhooks", "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", "webhooks"}) 

45@audit_logged 

46async def webhooks_create( 

47 url: str, 

48 events: list[str], 

49 region: str | None = None, 

50 namespace: str | None = None, 

51 secret: str | None = None, 

52) -> str: 

53 """`gco webhooks create` — register a new webhook subscription. 

54 

55 Args: 

56 url: Destination URL for webhook deliveries. 

57 events: Event names to subscribe to — one ``--event`` per entry 

58 (job.started / job.completed / job.failed). 

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

60 namespace: Only deliver events for jobs in this namespace. 

61 secret: Optional secret for HMAC signature verification. 

62 """ 

63 if not events: 

64 raise ValueError("events must contain at least one webhook event") 

65 

66 args = ["webhooks", "create", "--url", url] 

67 for event in events: 

68 args += ["--event", event] 

69 if namespace: 

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

71 if region: 

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

73 if secret: 

74 args += ["--secret", secret] 

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

76 

77 

78# ============================================================================= 

79# Destructive tools — gated by GCO_ENABLE_DESTRUCTIVE_OPERATIONS 

80# ============================================================================= 

81 

82 

83import contextlib # noqa: E402 

84 

85from feature_flags import FLAG_DESTRUCTIVE_OPERATIONS, is_enabled # noqa: E402 

86 

87 

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

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

90 try: 

91 from fastmcp.server.dependencies import get_context 

92 

93 ctx = get_context() 

94 except Exception: 

95 return 

96 with contextlib.suppress(Exception): 

97 await ctx.warning(message) 

98 

99 

100if is_enabled(FLAG_DESTRUCTIVE_OPERATIONS): 

101 

102 @mcp.tool(tags={"destructive", "webhooks"}) 

103 @audit_logged 

104 async def delete_webhook(name: str, region: str | None = None) -> str: 

105 """[gated by GCO_ENABLE_DESTRUCTIVE_OPERATIONS] destructive. 

106 

107 `gco webhooks delete` — delete a webhook subscription. 

108 Cannot be undone — the webhook record is permanently removed. 

109 

110 Args: 

111 name: Webhook identifier. 

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

113 """ 

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

115 args = ["webhooks", "delete", name, "-y"] 

116 if region: 

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

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