Coverage for mcp/tools/analytics.py: 96%
44 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-15 15:07 +0000
1"""Analytics environment MCP tools (read-only)."""
3import asyncio
5import cli_runner
6from audit import audit_logged
7from server import mcp
10@mcp.tool(tags={"safe", "analytics"})
11@audit_logged
12async def analytics_doctor() -> str:
13 """`gco analytics doctor` — run analytics environment health checks."""
14 return await asyncio.to_thread(cli_runner._run_cli, "analytics", "doctor")
17@mcp.tool(tags={"safe", "analytics"})
18@audit_logged
19async def analytics_login_url(username: str) -> str:
20 """`gco analytics login-url` — get a SageMaker Studio login URL for a user.
22 Args:
23 username: Cognito username.
24 """
25 return await asyncio.to_thread(cli_runner._run_cli, "analytics", "login-url", username)
28@mcp.tool(tags={"safe", "analytics"})
29@audit_logged
30async def analytics_users_list() -> str:
31 """`gco analytics users list` — list Cognito users in the analytics user pool."""
32 return await asyncio.to_thread(cli_runner._run_cli, "analytics", "users", "list")
35# =============================================================================
36# Mutating tools (low-risk)
37# =============================================================================
40@mcp.tool(tags={"low-risk", "analytics"})
41@audit_logged
42async def enable_analytics() -> str:
43 """`gco stacks analytics enable` — flip the analytics environment on in cdk.json.
45 Note: this only edits the cdk.json toggle. The change does not take effect
46 until ``gco stacks deploy-all`` runs to provision the analytics stack.
47 """
48 return await asyncio.to_thread(cli_runner._run_cli, "stacks", "analytics", "enable", "-y")
51@mcp.tool(tags={"low-risk", "analytics"})
52@audit_logged
53async def disable_analytics() -> str:
54 """`gco stacks analytics disable` — flip the analytics environment off in cdk.json.
56 Note: this only edits the cdk.json toggle. The change does not take effect
57 until ``gco stacks deploy-all`` runs to tear down the analytics stack.
58 """
59 return await asyncio.to_thread(cli_runner._run_cli, "stacks", "analytics", "disable", "-y")
62@mcp.tool(tags={"low-risk", "analytics"})
63@audit_logged
64async def analytics_user_add(username: str, email: str) -> str:
65 """`gco analytics users add` — create a Cognito user in the analytics pool.
67 Args:
68 username: Cognito username for the new user.
69 email: Email address for the new user.
70 """
71 return await asyncio.to_thread(
72 cli_runner._run_cli, "analytics", "users", "add", username, "--email", email
73 )
76# =============================================================================
77# Destructive tools — gated by GCO_ENABLE_DESTRUCTIVE_OPERATIONS
78# =============================================================================
81import contextlib # noqa: E402
83from feature_flags import FLAG_DESTRUCTIVE_OPERATIONS, is_enabled # noqa: E402
86async def _ctx_warning(message: str) -> None:
87 """Emit ``ctx.warning(...)`` from inside a tool body, no-op when no Context."""
88 try:
89 from fastmcp.server.dependencies import get_context
91 ctx = get_context()
92 except Exception:
93 return
94 with contextlib.suppress(Exception):
95 await ctx.warning(message)
98if is_enabled(FLAG_DESTRUCTIVE_OPERATIONS):
100 @mcp.tool(tags={"destructive", "analytics"})
101 @audit_logged
102 async def analytics_user_remove(username: str) -> str:
103 """[gated by GCO_ENABLE_DESTRUCTIVE_OPERATIONS] destructive.
105 `gco analytics users remove` — delete a Cognito user from the
106 analytics user pool. Cannot be undone — the user record and any
107 Studio profile artefacts owned by them are permanently removed.
109 Args:
110 username: Cognito username to remove.
111 """
112 await _ctx_warning(f"Removing analytics user {username!r} — this cannot be undone.")
113 return await asyncio.to_thread(
114 cli_runner._run_cli,
115 "analytics",
116 "users",
117 "remove",
118 "--username",
119 username,
120 "--yes",
121 )