wooo/awoooi
1
1
Fork 0
Code Issues 2 Pull Requests 1 Actions 2 Packages Projects Releases Wiki Activity

feat(api): Phase 13.2 #83 Grafana MCP Tool

Browse Source 
New MCP provider for Grafana dashboard integration:
- list_dashboards: List available dashboards with filtering
- get_dashboard: Get dashboard details by UID
- get_panel_data: Query panel data via Grafana Query API
- generate_dashboard_url: Generate shareable dashboard URLs

Security:
- API key authentication (Bearer token)
- Dashboard UID validation (alphanumeric + dash/underscore)
- Read-only operations only
- 30s request timeout

Config:
- GRAFANA_URL (default: http://192.168.0.188:3000)
- GRAFANA_API_KEY

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
This commit is contained in:
OG T
2026-03-26 15:36:17 +08:00
parent 14c81f728f
commit e58da5c534
3 changed files with 730 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
apps/api/src/plugins/mcp/providers/__init__.py
View File

@@ -6,11 +6,19 @@ MCP Tool Providers - ADR-015 模組化架構
- K8sProvider: Kubernetes 操作 (kubectl)
- SignOzProvider: 監控指標查詢
- DatabaseProvider: 資料庫查詢 (Approval/Incident)
- FilesystemProvider: 安全受限的文件讀取 (#82)
- GrafanaProvider: Grafana Dashboard 查詢 (#83)
@see docs/adr/ADR-015-mcp-modular-architecture.md
變更紀錄:
- 2026-03-26 v1.2: 新增 GrafanaProvider (#83) - Claude Code
- 2026-03-26 v1.1: 新增 FilesystemProvider (#82) - Claude Code
"""
from src.plugins.mcp.providers.database_provider import DatabaseProvider
from src.plugins.mcp.providers.filesystem_provider import FilesystemProvider
from src.plugins.mcp.providers.grafana_provider import GrafanaProvider
from src.plugins.mcp.providers.k8s_provider import K8sProvider
from src.plugins.mcp.providers.signoz_provider import SignOzProvider
@@ -18,6 +26,8 @@ __all__ = [
"K8sProvider",
"SignOzProvider",
"DatabaseProvider",
"FilesystemProvider",
"GrafanaProvider",
]
@@ -32,3 +42,5 @@ def register_all_providers() -> None:
register_provider(K8sProvider())
register_provider(SignOzProvider())
register_provider(DatabaseProvider())
register_provider(FilesystemProvider())
register_provider(GrafanaProvider())

665
apps/api/src/plugins/mcp/providers/grafana_provider.py Normal file
View File

@@ -0,0 +1,665 @@
"""
Grafana MCP Tool Provider - Phase 13.2 #83
==========================================
提供 Grafana Dashboard 查詢工具：
- list_dashboards: 列出可用 Dashboard
- get_dashboard: 取得 Dashboard 詳細資訊
- get_panel_data: 取得 Panel 數據
- generate_dashboard_url: 產生 Dashboard URL
安全機制:
1. API Key 認證 (Bearer Token)
2. 唯讀操作 (Read-Only)
3. Dashboard UID 驗證
透過 DI 注入配置，不直接 import services。
@see docs/adr/ADR-015-mcp-modular-architecture.md
@author Claude Code
@version 1.0
@created 2026-03-26 (台北時區)
@issue #83
"""
import re
import uuid
from typing import Any
from urllib.parse import urljoin
import httpx
import structlog
from src.plugins.mcp.interfaces import MCPTool, MCPToolProvider, MCPToolResult
logger = structlog.get_logger(__name__)
# =============================================================================
# Configuration Constants
# =============================================================================
# Default Grafana URL (can be overridden via settings)
DEFAULT_GRAFANA_URL = "http://192.168.0.188:3000"
# Dashboard UID validation pattern (alphanumeric, dash, underscore)
DASHBOARD_UID_PATTERN = re.compile(r"^[a-zA-Z0-9_-]+$")
# Request timeout in seconds
REQUEST_TIMEOUT = 30.0
class GrafanaError(Exception):
"""Grafana API 錯誤"""
pass
class GrafanaProvider(MCPToolProvider):
"""
Grafana MCP Tool Provider
提供 Grafana Dashboard 查詢功能：
1. 列出所有 Dashboard
2. 取得 Dashboard 詳細資訊
3. 取得 Panel 數據
4. 產生 Dashboard URL
Security Notes:
- 所有操作都是唯讀
- Dashboard UID 會被驗證格式
- API Key 透過 Bearer Token 認證
"""
def __init__(
self,
grafana_url: str | None = None,
api_key: str | None = None,
) -> None:
"""
初始化 Provider
Args:
grafana_url: Grafana URL (default: from settings)
api_key: Grafana API Key (default: from settings)
"""
self._grafana_url = grafana_url
self._api_key = api_key
self._client: httpx.AsyncClient | None = None
logger.info(
"grafana_provider_initialized",
grafana_url=self._get_grafana_url(),
)
def _get_grafana_url(self) -> str:
"""取得 Grafana URL (lazy load from settings)"""
if self._grafana_url:
return self._grafana_url
try:
from src.core.config import get_settings
settings = get_settings()
return getattr(settings, "GRAFANA_URL", DEFAULT_GRAFANA_URL)
except Exception:
return DEFAULT_GRAFANA_URL
def _get_api_key(self) -> str | None:
"""取得 API Key (lazy load from settings)"""
if self._api_key:
return self._api_key
try:
from src.core.config import get_settings
settings = get_settings()
return getattr(settings, "GRAFANA_API_KEY", None)
except Exception:
return None
def _get_headers(self) -> dict[str, str]:
"""取得 HTTP Headers (含認證)"""
headers = {
"Content-Type": "application/json",
"Accept": "application/json",
}
api_key = self._get_api_key()
if api_key:
headers["Authorization"] = f"Bearer {api_key}"
return headers
async def _get_client(self) -> httpx.AsyncClient:
"""取得 HTTP Client (lazy initialization)"""
if self._client is None:
self._client = httpx.AsyncClient(
base_url=self._get_grafana_url(),
headers=self._get_headers(),
timeout=REQUEST_TIMEOUT,
)
return self._client
@property
def name(self) -> str:
return "grafana"
async def list_tools(self) -> list[MCPTool]:
return [
MCPTool(
name="list_dashboards",
description="List available Grafana dashboards. Returns dashboard titles, UIDs, and folder info.",
input_schema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "Search query to filter dashboards by title",
},
"folder_id": {
"type": "integer",
"description": "Filter by folder ID (0 = General folder)",
},
"tag": {
"type": "string",
"description": "Filter by tag (e.g., 'kubernetes', 'prometheus')",
},
"limit": {
"type": "integer",
"description": "Max results (default: 100)",
"default": 100,
},
},
},
server_name=self.name,
),
MCPTool(
name="get_dashboard",
description="Get dashboard details by UID. Returns panels, variables, and annotations.",
input_schema={
"type": "object",
"properties": {
"uid": {
"type": "string",
"description": "Dashboard UID (e.g., 'abc123')",
},
},
"required": ["uid"],
},
server_name=self.name,
),
MCPTool(
name="get_panel_data",
description="Get data from a specific panel in a dashboard. Useful for fetching metrics.",
input_schema={
"type": "object",
"properties": {
"dashboard_uid": {
"type": "string",
"description": "Dashboard UID",
},
"panel_id": {
"type": "integer",
"description": "Panel ID within the dashboard",
},
"from_time": {
"type": "string",
"description": "Start time (e.g., 'now-1h', '2026-03-26T00:00:00Z')",
"default": "now-1h",
},
"to_time": {
"type": "string",
"description": "End time (e.g., 'now', '2026-03-26T01:00:00Z')",
"default": "now",
},
},
"required": ["dashboard_uid", "panel_id"],
},
server_name=self.name,
),
MCPTool(
name="generate_dashboard_url",
description="Generate a URL to view a dashboard with optional time range and variables.",
input_schema={
"type": "object",
"properties": {
"uid": {
"type": "string",
"description": "Dashboard UID",
},
"from_time": {
"type": "string",
"description": "Start time (e.g., 'now-1h', 'now-24h')",
"default": "now-1h",
},
"to_time": {
"type": "string",
"description": "End time (e.g., 'now')",
"default": "now",
},
"variables": {
"type": "object",
"description": "Template variables as key-value pairs (e.g., {'namespace': 'awoooi-prod'})",
},
"panel_id": {
"type": "integer",
"description": "Focus on specific panel (opens panel view)",
},
},
"required": ["uid"],
},
server_name=self.name,
),
]
async def execute(
self,
tool_name: str,
parameters: dict[str, Any],
) -> MCPToolResult:
execution_id = str(uuid.uuid4())[:8]
try:
if tool_name == "list_dashboards":
output = await self._list_dashboards(parameters)
elif tool_name == "get_dashboard":
output = await self._get_dashboard(parameters)
elif tool_name == "get_panel_data":
output = await self._get_panel_data(parameters)
elif tool_name == "generate_dashboard_url":
output = await self._generate_dashboard_url(parameters)
else:
return MCPToolResult(
success=False,
execution_id=execution_id,
error=f"Unknown tool: {tool_name}",
)
logger.info(
"grafana_tool_executed",
tool=tool_name,
execution_id=execution_id,
)
return MCPToolResult(
success=True,
execution_id=execution_id,
output=output,
)
except GrafanaError as e:
logger.warning(
"grafana_api_error",
tool=tool_name,
execution_id=execution_id,
error=str(e),
)
return MCPToolResult(
success=False,
execution_id=execution_id,
error=f"Grafana API error: {e}",
)
except httpx.HTTPStatusError as e:
logger.warning(
"grafana_http_error",
tool=tool_name,
execution_id=execution_id,
status_code=e.response.status_code,
)
return MCPToolResult(
success=False,
execution_id=execution_id,
error=f"HTTP {e.response.status_code}: {e.response.text[:200]}",
)
except httpx.RequestError as e:
logger.warning(
"grafana_request_error",
tool=tool_name,
execution_id=execution_id,
error=str(e),
)
return MCPToolResult(
success=False,
execution_id=execution_id,
error=f"Request failed: {e}",
)
except Exception as e:
logger.exception(
"grafana_provider_error",
tool=tool_name,
execution_id=execution_id,
error=str(e),
)
return MCPToolResult(
success=False,
execution_id=execution_id,
error=str(e),
)
# =========================================================================
# Validation Methods
# =========================================================================
def _validate_dashboard_uid(self, uid: str) -> str:
"""
驗證 Dashboard UID 格式
Args:
uid: Dashboard UID
Returns:
str: 驗證後的 UID
Raises:
GrafanaError: UID 格式不正確
"""
if not uid:
raise GrafanaError("Dashboard UID is required")
uid = uid.strip()
if len(uid) > 40:
raise GrafanaError(f"Dashboard UID too long: {len(uid)} chars (max: 40)")
if not DASHBOARD_UID_PATTERN.match(uid):
raise GrafanaError(
f"Invalid dashboard UID format: {uid}. "
"Only alphanumeric, dash, and underscore allowed."
)
return uid
# =========================================================================
# Tool Implementations
# =========================================================================
async def _list_dashboards(self, parameters: dict) -> dict:
"""
列出 Dashboard
Grafana API: GET /api/search
"""
query = parameters.get("query")
folder_id = parameters.get("folder_id")
tag = parameters.get("tag")
limit = parameters.get("limit", 100)
# 構建查詢參數
params: dict[str, Any] = {
"type": "dash-db", # 只搜尋 dashboard
"limit": min(limit, 500), # 最多 500
}
if query:
params["query"] = query
if folder_id is not None:
params["folderIds"] = folder_id
if tag:
params["tag"] = tag
# 呼叫 Grafana API
client = await self._get_client()
response = await client.get("/api/search", params=params)
response.raise_for_status()
dashboards = response.json()
# 轉換回應格式
results = []
for db in dashboards:
results.append({
"uid": db.get("uid"),
"title": db.get("title"),
"folder_title": db.get("folderTitle", "General"),
"folder_id": db.get("folderId", 0),
"url": db.get("url"),
"tags": db.get("tags", []),
"type": db.get("type"),
})
return {
"dashboards": results,
"count": len(results),
"filters": {
"query": query,
"folder_id": folder_id,
"tag": tag,
},
}
async def _get_dashboard(self, parameters: dict) -> dict:
"""
取得 Dashboard 詳細資訊
Grafana API: GET /api/dashboards/uid/:uid
"""
uid = parameters.get("uid", "")
uid = self._validate_dashboard_uid(uid)
# 呼叫 Grafana API
client = await self._get_client()
response = await client.get(f"/api/dashboards/uid/{uid}")
response.raise_for_status()
data = response.json()
dashboard = data.get("dashboard", {})
meta = data.get("meta", {})
# 提取 Panel 資訊
panels = []
for panel in dashboard.get("panels", []):
panels.append({
"id": panel.get("id"),
"title": panel.get("title"),
"type": panel.get("type"),
"description": panel.get("description"),
"datasource": panel.get("datasource"),
})
# 提取變數資訊
variables = []
for var in dashboard.get("templating", {}).get("list", []):
variables.append({
"name": var.get("name"),
"label": var.get("label"),
"type": var.get("type"),
"current_value": var.get("current", {}).get("value"),
})
return {
"uid": uid,
"title": dashboard.get("title"),
"description": dashboard.get("description"),
"tags": dashboard.get("tags", []),
"folder": meta.get("folderTitle", "General"),
"created_by": meta.get("createdBy"),
"updated_by": meta.get("updatedBy"),
"version": dashboard.get("version"),
"panels": panels,
"panel_count": len(panels),
"variables": variables,
"url": meta.get("url"),
}
async def _get_panel_data(self, parameters: dict) -> dict:
"""
取得 Panel 數據
使用 Grafana Query API 取得 Panel 數據
Grafana API: POST /api/ds/query
"""
dashboard_uid = parameters.get("dashboard_uid", "")
dashboard_uid = self._validate_dashboard_uid(dashboard_uid)
panel_id = parameters.get("panel_id")
if panel_id is None:
raise GrafanaError("Panel ID is required")
from_time = parameters.get("from_time", "now-1h")
to_time = parameters.get("to_time", "now")
# 先取得 Dashboard 取得 Panel 定義
client = await self._get_client()
dash_response = await client.get(f"/api/dashboards/uid/{dashboard_uid}")
dash_response.raise_for_status()
dashboard = dash_response.json().get("dashboard", {})
panel = None
# 尋找 Panel
for p in dashboard.get("panels", []):
if p.get("id") == panel_id:
panel = p
break
if not panel:
raise GrafanaError(f"Panel {panel_id} not found in dashboard {dashboard_uid}")
# 取得 Panel 基本資訊
panel_info = {
"dashboard_uid": dashboard_uid,
"panel_id": panel_id,
"panel_title": panel.get("title"),
"panel_type": panel.get("type"),
"from_time": from_time,
"to_time": to_time,
}
# 取得 Panel 的查詢定義
targets = panel.get("targets", [])
if not targets:
return {
**panel_info,
"data": None,
"message": "Panel has no query targets",
}
# 使用 Grafana 的 Query API
# 注意: 這需要 datasource 支援且有權限
datasource = panel.get("datasource")
if not datasource:
return {
**panel_info,
"data": None,
"message": "Panel has no datasource configured",
"targets": [t.get("expr", t.get("query", "N/A")) for t in targets],
}
# 嘗試執行查詢
try:
query_payload = {
"queries": [
{
**target,
"datasource": datasource,
"refId": target.get("refId", "A"),
}
for target in targets
],
"from": from_time,
"to": to_time,
}
query_response = await client.post("/api/ds/query", json=query_payload)
query_response.raise_for_status()
query_data = query_response.json()
# 解析回應
results = query_data.get("results", {})
frames = []
for ref_id, result in results.items():
for frame in result.get("frames", []):
schema = frame.get("schema", {})
data = frame.get("data", {})
frames.append({
"ref_id": ref_id,
"name": schema.get("name"),
"fields": [
{
"name": f.get("name"),
"type": f.get("type"),
}
for f in schema.get("fields", [])
],
"row_count": len(data.get("values", [[]])[0]) if data.get("values") else 0,
})
return {
**panel_info,
"datasource": datasource,
"frames": frames,
"frame_count": len(frames),
}
except httpx.HTTPStatusError as e:
# 查詢失敗，回傳 Panel 資訊但無數據
return {
**panel_info,
"data": None,
"message": f"Query failed: HTTP {e.response.status_code}",
"targets": [t.get("expr", t.get("query", "N/A")) for t in targets],
}
async def _generate_dashboard_url(self, parameters: dict) -> dict:
"""
產生 Dashboard URL
支援時間範圍、變數、Panel 聚焦
"""
uid = parameters.get("uid", "")
uid = self._validate_dashboard_uid(uid)
from_time = parameters.get("from_time", "now-1h")
to_time = parameters.get("to_time", "now")
variables = parameters.get("variables", {})
panel_id = parameters.get("panel_id")
# 構建 URL
base_url = self._get_grafana_url()
dashboard_path = f"/d/{uid}"
# 構建查詢參數
query_params = [
f"from={from_time}",
f"to={to_time}",
]
# 添加變數
for key, value in variables.items():
# Grafana 變數格式: var-<name>=<value>
query_params.append(f"var-{key}={value}")
# 添加 Panel 聚焦
if panel_id is not None:
query_params.append(f"viewPanel={panel_id}")
# 組合 URL
query_string = "&".join(query_params)
full_url = urljoin(base_url, f"{dashboard_path}?{query_string}")
return {
"uid": uid,
"url": full_url,
"from_time": from_time,
"to_time": to_time,
"variables": variables,
"panel_id": panel_id,
}
async def health_check(self) -> bool:
"""
健康檢查
驗證 Grafana API 可連線
"""
try:
client = await self._get_client()
response = await client.get("/api/health")
return response.status_code == 200
except Exception as e:
logger.warning("grafana_health_check_failed", error=str(e))
return False
async def close(self) -> None:
"""關閉 HTTP Client"""
if self._client:
await self._client.aclose()
self._client = None