#!/usr/bin/env python3 """ Automated Grafana Cloud Setup & Team Seat Invitation Utility. Connects to Grafana (Cloud or self-hosted) via Grafana REST API & MCP, configures Prometheus data source, imports production dashboards, and invites team members. """ import os import sys import json import urllib.request import urllib.error from typing import Dict, Any, Optional CLOUDFRONT_METRICS_URL = "https://d1ca9upoitnq9b.cloudfront.net/metrics" DASHBOARD_FILE = os.path.join( os.path.dirname(__file__), "..", "grafana", "dashboards", "lifecycle-tracker-overview.json" ) class GrafanaClient: def __init__(self, base_url: str, token: str): self.base_url = base_url.rstrip("/") self.token = token.strip() self.headers = { "Authorization": f"Bearer {self.token}", "Content-Type": "application/json", "Accept": "application/json", } def _request(self, method: str, path: str, payload: Optional[Dict[str, Any]] = None) -> Dict[str, Any]: url = f"{self.base_url}{path}" data = json.dumps(payload).encode("utf-8") if payload is not None else None req = urllib.request.Request(url, data=data, headers=self.headers, method=method) try: with urllib.request.urlopen(req) as resp: status = resp.status body = resp.read().decode("utf-8") return json.loads(body) if body else {"status": status} except urllib.error.HTTPError as err: err_body = err.read().decode("utf-8") try: err_json = json.loads(err_body) msg = err_json.get("message", err_body) except Exception: msg = err_body raise RuntimeError(f"Grafana API error ({err.code} {err.reason}) on {method} {path}: {msg}") except urllib.error.URLError as err: raise RuntimeError(f"Connection failed to {url}: {err.reason}") def check_health(self) -> Dict[str, Any]: return self._request("GET", "/api/health") def get_org(self) -> Dict[str, Any]: return self._request("GET", "/api/org") def ensure_datasource(self, name: str = "AWS-CloudFront-Prometheus", url: str = CLOUDFRONT_METRICS_URL) -> Dict[str, Any]: """Ensure Prometheus datasource pointing to CloudFront /metrics exists.""" datasources = self._request("GET", "/api/datasources") for ds in datasources: if ds.get("name") == name: print(f"[*] Datasource '{name}' already exists (id={ds.get('id')}). Updating...") update_payload = { "name": name, "type": "prometheus", "access": "proxy", "url": url, "isDefault": True, "jsonData": { "httpMethod": "GET" } } return self._request("PUT", f"/api/datasources/{ds['id']}", update_payload) print(f"[*] Creating new Prometheus datasource '{name}' pointing to {url}...") create_payload = { "name": name, "type": "prometheus", "access": "proxy", "url": url, "isDefault": True, "jsonData": { "httpMethod": "GET" } } return self._request("POST", "/api/datasources", create_payload) def import_dashboard(self, dashboard_path: str = DASHBOARD_FILE, ds_uid: Optional[str] = None) -> Dict[str, Any]: """Import the production Lifecycle Tracker Overview dashboard.""" if not os.path.exists(dashboard_path): raise FileNotFoundError(f"Dashboard file not found: {dashboard_path}") with open(dashboard_path, "r", encoding="utf-8") as f: dashboard_json = json.load(f) # Remove id for new import dashboard_json["id"] = None dashboard_json["uid"] = "lifecycle-tracker-overview" if ds_uid: dashboard_str = json.dumps(dashboard_json) dashboard_str = dashboard_str.replace('"uid": "prometheus"', f'"uid": "{ds_uid}"') dashboard_json = json.loads(dashboard_str) payload = { "dashboard": dashboard_json, "overwrite": True, "message": "Automated import via Coding Agent Grafana Setup" } print(f"[*] Importing dashboard '{dashboard_json.get('title')}' into Grafana...") result = self._request("POST", "/api/dashboards/db", payload) return result def invite_team_member(self, email: str, role: str = "Viewer", name: str = "") -> Dict[str, Any]: """Invite a team member to the Grafana organization.""" payload = { "name": name or email.split("@")[0], "email": email, "role": role, # Viewer, Editor, Admin "sendEmail": True } print(f"[*] Inviting team member {email} with role '{role}'...") return self._request("POST", "/api/org/invites", payload) def list_invites(self) -> Dict[str, Any]: return self._request("GET", "/api/org/invites") def main(): grafana_url = os.getenv("GRAFANA_URL", "").strip() grafana_token = os.getenv("GRAFANA_SERVICE_ACCOUNT_TOKEN", "").strip() if len(sys.argv) >= 3: grafana_url = sys.argv[1].strip() grafana_token = sys.argv[2].strip() if not grafana_url or not grafana_token: print("Usage:") print(" python3 grafana_cloud_setup.py [INVITE_EMAIL_1] [INVITE_EMAIL_2] ...") print("Or set environment variables:") print(" export GRAFANA_URL=\"https://.grafana.net\"") print(" export GRAFANA_SERVICE_ACCOUNT_TOKEN=\"glsa_...\"") sys.exit(1) print(f"[*] Connecting to Grafana at: {grafana_url}") client = GrafanaClient(grafana_url, grafana_token) health = client.check_health() print(f"[+] Grafana connection healthy: version={health.get('version', 'unknown')}, commit={health.get('commit', 'unknown')}") try: org = client.get_org() print(f"[+] Connected to Organization: {org.get('name')} (id={org.get('id')})") except Exception as e: print(f"[!] Note: Org fetch: {e}") # 1. Identify Prometheus Datasource (e.g. grafanacloud-prom) datasources = client._request("GET", "/api/datasources") prom_ds = None for ds in datasources: if ds.get("uid") == "grafanacloud-prom": prom_ds = ds break elif ds.get("type") == "prometheus" and not prom_ds: prom_ds = ds ds_uid = prom_ds.get("uid") if prom_ds else "grafanacloud-prom" ds_name = prom_ds.get("name") if prom_ds else "grafanacloud-prom" print(f"[+] Binding Dashboard to hosted Prometheus Datasource: {ds_name} (UID={ds_uid})") # 2. Import Dashboard dash_res = client.import_dashboard(ds_uid=ds_uid) dash_url = dash_res.get("url", "/d/lifecycle-tracker-overview") full_dash_url = f"{client.base_url}{dash_url}" print(f"[+] Dashboard imported successfully!") print(f" 👉 Live Dashboard URL: {full_dash_url}") # 3. Handle any invited emails passed as CLI args invite_emails = sys.argv[3:] for email in invite_emails: try: client.invite_team_member(email, role="Viewer") print(f"[+] Successfully invited {email}!") except Exception as e: print(f"[-] Failed to invite {email}: {e}") print("\n=================================================================") print(" Grafana Cloud Setup Complete!") print("=================================================================") print(f" • Grafana Dashboard: {full_dash_url}") print(f" • Loki Logs Explorer: {client.base_url}/explore?left=%7B\"datasource\":\"grafanacloud-logs\"%7D") print(f" • Tempo Traces Explorer: {client.base_url}/explore?left=%7B\"datasource\":\"grafanacloud-traces\"%7D") print(f" • Team Management: {client.base_url}/admin/users") print("=================================================================") if __name__ == "__main__": main()