"""Helpers for MCP tool responses: text content and error formatting.""" import json from typing import Any from src.utils.errors import ( AuthorizationError, AuthServiceError, DeltaError, NetworkError, PrometheusAPIError, PrometheusTimeoutError, ValidationError, ) def tool_text_response(text: str) -> list[dict[str, Any]]: """Return the standard MCP tool response list with a single text content.""" return [{"type": "text", "text": text}] def tool_json_response( data: Any, *, indent: int = 2, ensure_ascii: bool = False ) -> list[dict[str, Any]]: """Return the standard MCP tool response list with JSON-serialized text content.""" text = json.dumps(data, indent=indent, ensure_ascii=ensure_ascii) return [{"type": "text", "text": text}] def format_tool_error(exc: Exception) -> str: """ Format an exception as Markdown for tool output. Handles known error types (AuthServiceError, ValidationError, etc.) with consistent prefixes and optional details (e.g. HTTP status, Prometheus detail). """ if isinstance(exc, AuthServiceError): suffix = f" (HTTP {exc.status_code})" if getattr(exc, "status_code", None) else "" return f"**Auth Error**: {exc.message}{suffix}" if isinstance(exc, AuthorizationError): return f"**Error**: {exc.message}\n\n401 Unauthorized" if isinstance(exc, ValidationError): return f"**Validation Error**: {exc.message}" if isinstance(exc, (NetworkError, PrometheusTimeoutError)): return f"**Error**: {exc.message}" if isinstance(exc, PrometheusAPIError): detail = getattr(exc, "prometheus_error", None) if detail: return f"**Error**: Invalid PromQL syntax. Error details: {detail}" return f"**Error**: {exc.message}" if isinstance(exc, DeltaError): return f"**Delta Error**: {exc.message}" return f"**Unexpected Error**: {str(exc)}"