"""
MCP tools for querying AWS Managed Prometheus.
These tools wrap the AMP client to provide a user-friendly interface for PromQL queries.
"""
from prometheus_mcp.amp_client import AMPClient
from prometheus_mcp.promql.models import (
LabelList,
LabelValues,
MetricList,
QueryResult,
RangeQueryResult,
)
# Global client instance (initialized by server)
_client: AMPClient | None = None
def get_client() -> AMPClient:
"""Get the initialized AMP client."""
if _client is None:
raise RuntimeError("AMP client not initialized. Call init_client() first.")
return _client
def init_client(
workspace_id: str | None = None,
region: str | None = None,
) -> AMPClient:
"""
Initialize the global AMP client.
Args:
workspace_id: AMP workspace ID
region: AWS region
Returns:
The initialized AMPClient instance
"""
global _client
_client = AMPClient(workspace_id=workspace_id, region=region)
return _client
async def query_instant(
query: str,
time: str | None = None,
) -> QueryResult:
"""
Execute an instant PromQL query against AWS Managed Prometheus.
This tool evaluates a PromQL expression at a single point in time and returns
the current value of matching time series. Use this for getting current metric
values or point-in-time snapshots.
Args:
query: PromQL query expression (e.g., 'up', 'rate(http_requests_total[5m])')
time: Optional evaluation timestamp in RFC3339 format (e.g., '2024-01-15T10:30:00Z')
or Unix timestamp. Defaults to current server time.
Returns:
QueryResult containing the status and matching time series with their current values.
Examples:
- Get all up metrics: query_instant("up")
- Calculate request rate: query_instant("rate(http_requests_total[5m])")
- Filter by label: query_instant('http_requests_total{method="GET"}')
- At specific time: query_instant("up", time="2024-01-15T10:00:00Z")
"""
client = get_client()
response = await client.query(promql=query, time=time)
return QueryResult.from_prometheus_response(response)
async def query_range(
query: str,
start: str,
end: str,
step: str = "1m",
) -> RangeQueryResult:
"""
Execute a range PromQL query to get time series data over a time period.
This tool evaluates a PromQL expression over a range of time and returns
data points at regular intervals. Use this for historical data analysis,
graphing, and trend analysis.
Args:
query: PromQL query expression (e.g., 'up', 'rate(http_requests_total[5m])')
start: Start timestamp in RFC3339 format (e.g., '2024-01-15T00:00:00Z')
or Unix timestamp
end: End timestamp in RFC3339 format (e.g., '2024-01-15T12:00:00Z')
or Unix timestamp
step: Query resolution step width (e.g., '15s', '1m', '5m', '1h').
Defaults to '1m'. Smaller steps = more data points but slower queries.
Returns:
RangeQueryResult containing time series with values at each step interval.
Examples:
- Get CPU usage for last hour:
query_range("rate(node_cpu_seconds_total[5m])",
start="2024-01-15T09:00:00Z",
end="2024-01-15T10:00:00Z",
step="1m")
- Get request rate over a day:
query_range("rate(http_requests_total[5m])",
start="2024-01-14T00:00:00Z",
end="2024-01-15T00:00:00Z",
step="5m")
"""
client = get_client()
response = await client.query_range(promql=query, start=start, end=end, step=step)
return RangeQueryResult.from_prometheus_response(response)
async def list_labels(match: list[str] | None = None) -> LabelList:
"""
Get all label names from AWS Managed Prometheus.
This tool returns all unique label names present in the stored time series.
Use this for discovering what labels are available for filtering queries.
Args:
match: Optional list of series selectors to filter which series to consider.
For example: ['up', 'http_requests_total'] will only return labels
that exist on those metrics.
Returns:
LabelList containing all unique label names.
Examples:
- Get all labels: list_labels()
- Get labels for specific metrics: list_labels(match=["up", "http_requests_total"])
- Get labels matching a pattern: list_labels(match=['{job="prometheus"}'])
"""
client = get_client()
response = await client.labels(match=match)
return LabelList.from_prometheus_response(response)
async def get_label_values(
label_name: str,
match: list[str] | None = None,
) -> LabelValues:
"""
Get all values for a specific label from AWS Managed Prometheus.
This tool returns all unique values that exist for a given label name.
Use this to discover possible filter values for a specific label.
Args:
label_name: The label name to get values for (e.g., 'job', 'instance', 'namespace')
match: Optional list of series selectors to filter which series to consider.
For example: ['up'] will only return values from the 'up' metric.
Returns:
LabelValues containing all unique values for the specified label.
Examples:
- Get all job names: get_label_values("job")
- Get instances for a specific job: get_label_values("instance", match=['{job="prometheus"}'])
- Get namespaces: get_label_values("namespace")
"""
client = get_client()
response = await client.label_values(label_name=label_name, match=match)
return LabelValues.from_prometheus_response(response, label_name=label_name)
async def list_metrics(with_metadata: bool = False) -> MetricList:
"""
Get all metric names from AWS Managed Prometheus.
This tool returns all unique metric names available in the Prometheus workspace.
Optionally includes metadata like metric type, help text, and unit.
Args:
with_metadata: If True, fetches full metadata (type, help, unit) for each metric.
This is slower but provides more information. Defaults to False.
Returns:
MetricList containing all available metrics with optional metadata.
Examples:
- Get all metric names: list_metrics()
- Get metrics with full metadata: list_metrics(with_metadata=True)
"""
client = get_client()
if with_metadata:
response = await client.metadata()
return MetricList.from_metadata_response(response)
else:
response = await client.label_values("__name__")
return MetricList.from_label_values_response(response)
