Coverage for gco_mcp/tools/analytics.py: 96.36%
51 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-30 21:22 +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_status() -> str:
20 """`gco analytics status` — show the analytics environment configuration."""
21 return await asyncio.to_thread(cli_runner._run_cli, "analytics", "status")
24@mcp.tool(tags={"safe", "analytics"})
25@audit_logged
26async def analytics_login_url(username: str, password: str | None = None) -> str:
27 """`gco analytics studio login` — get a SageMaker Studio presigned login URL.
29 SRP-authenticates the user against Cognito and prints the presigned Studio
30 URL. The password is read from ``$GCO_STUDIO_PASSWORD`` when not passed here.
32 Args:
33 username: Cognito username to sign in with.
34 password: Cognito password. Omit to use ``$GCO_STUDIO_PASSWORD``.
35 """
36 args = ["analytics", "studio", "login", "--username", username]
37 if password:
38 args += ["--password", password]
39 return await asyncio.to_thread(cli_runner._run_cli, *args)
42@mcp.tool(tags={"safe", "analytics"})
43@audit_logged
44async def analytics_users_list() -> str:
45 """`gco analytics users list` — list Cognito users in the analytics user pool."""
46 return await asyncio.to_thread(cli_runner._run_cli, "analytics", "users", "list")
49# =============================================================================
50# Mutating tools (low-risk)
51# =============================================================================
54@mcp.tool(tags={"low-risk", "analytics"})
55@audit_logged
56async def enable_analytics() -> str:
57 """`gco analytics enable` — flip the analytics environment on in cdk.json.
59 Note: this only edits the cdk.json toggle. The change does not take effect
60 until ``gco stacks deploy-all`` runs to provision the analytics stack.
61 """
62 return await asyncio.to_thread(cli_runner._run_cli, "analytics", "enable", "-y")
65@mcp.tool(tags={"low-risk", "analytics"})
66@audit_logged
67async def disable_analytics() -> str:
68 """`gco analytics disable` — flip the analytics environment off in cdk.json.
70 Note: this only edits the cdk.json toggle. The change does not take effect
71 until ``gco stacks deploy-all`` runs to tear down the analytics stack.
72 """
73 return await asyncio.to_thread(cli_runner._run_cli, "analytics", "disable", "-y")
76@mcp.tool(tags={"low-risk", "analytics"})
77@audit_logged
78async def analytics_user_add(username: str, email: str) -> str:
79 """`gco analytics users add` — create a Cognito user in the analytics pool.
81 Args:
82 username: Cognito username for the new user.
83 email: Email address for the new user.
84 """
85 return await asyncio.to_thread(
86 cli_runner._run_cli, "analytics", "users", "add", username, "--email", email
87 )
90# =============================================================================
91# Destructive tools — gated by GCO_ENABLE_DESTRUCTIVE_OPERATIONS
92# =============================================================================
95import contextlib # noqa: E402
97from feature_flags import FLAG_DESTRUCTIVE_OPERATIONS, is_enabled # noqa: E402
100async def _ctx_warning(message: str) -> None:
101 """Emit ``ctx.warning(...)`` from inside a tool body, no-op when no Context."""
102 try:
103 from fastmcp.server.dependencies import get_context
105 ctx = get_context()
106 except Exception:
107 return
108 with contextlib.suppress(Exception):
109 await ctx.warning(message)
112if is_enabled(FLAG_DESTRUCTIVE_OPERATIONS):
114 @mcp.tool(tags={"destructive", "analytics"})
115 @audit_logged
116 async def analytics_user_remove(username: str) -> str:
117 """[gated by GCO_ENABLE_DESTRUCTIVE_OPERATIONS] destructive.
119 `gco analytics users remove` — delete a Cognito user from the
120 analytics user pool. Cannot be undone — the user record and any
121 Studio profile artefacts owned by them are permanently removed.
123 Args:
124 username: Cognito username to remove.
125 """
126 await _ctx_warning(f"Removing analytics user {username!r} — this cannot be undone.")
127 return await asyncio.to_thread(
128 cli_runner._run_cli,
129 "analytics",
130 "users",
131 "remove",
132 "--username",
133 username,
134 "--yes",
135 )