Coverage for mcp/tools/webhooks.py: 97%

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 name: str, 

48 url: str, 

49 events: list[str], 

50 region: str | None = None, 

51 secret_name: str | None = None, 

52) -> str: 

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

54 

55 Args: 

56 name: Webhook name. 

57 url: Destination URL for webhook deliveries. 

58 events: Event names to subscribe to (one ``--event`` flag per entry). 

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

60 secret_name: Optional Secrets Manager secret name for HMAC signing. 

61 """ 

62 args = ["webhooks", "create", name, "--url", url] 

63 for event in events: 

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

65 if region: 

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

67 if secret_name: 

68 args += ["--secret-name", secret_name] 

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

70 

71 

72# ============================================================================= 

73# Destructive tools — gated by GCO_ENABLE_DESTRUCTIVE_OPERATIONS 

74# ============================================================================= 

75 

76 

77import contextlib # noqa: E402 

78 

79from feature_flags import FLAG_DESTRUCTIVE_OPERATIONS, is_enabled # noqa: E402 

80 

81 

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

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

84 try: 

85 from fastmcp.server.dependencies import get_context 

86 

87 ctx = get_context() 

88 except Exception: 

89 return 

90 with contextlib.suppress(Exception): 

91 await ctx.warning(message) 

92 

93 

94if is_enabled(FLAG_DESTRUCTIVE_OPERATIONS): 

95 

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

97 @audit_logged 

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

99 """[gated by GCO_ENABLE_DESTRUCTIVE_OPERATIONS] destructive. 

100 

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

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

103 

104 Args: 

105 name: Webhook identifier. 

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

107 """ 

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

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

110 if region: 

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

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