"""
Prometheus metrics for monitoring.
This module defines all metrics collected by the application.
Metrics are exposed on a separate port for Prometheus to scrape.
Key Metrics:
- Request counts by tool and status
- Request latency histograms
- Active connections
- LLM API calls and token usage
- Error rates
"""
from __future__ import annotations
from prometheus_client import Counter, Histogram, Gauge, Info
from functools import wraps
import time
from typing import Callable, Any
from src.config import settings
class Metrics:
"""
Centralized metrics collector.
All metrics are defined here for easy discovery and management.
"""
def __init__(self):
"""Initialize all metrics."""
# MCP Server Metrics
self.mcp_requests_total = Counter(
"mcp_requests_total",
"Total MCP tool requests",
["tool_name", "status"] # Labels for grouping
)
self.mcp_request_duration = Histogram(
"mcp_request_duration_seconds",
"MCP tool request duration",
["tool_name"],
buckets=[0.01, 0.05, 0.1, 0.5, 1.0, 2.0, 5.0, 10.0]
)
self.mcp_active_connections = Gauge(
"mcp_active_connections",
"Number of active MCP connections"
)
# LLM Provider Metrics
self.llm_api_calls_total = Counter(
"llm_api_calls_total",
"Total LLM API calls",
["provider", "model", "status"]
)
self.llm_api_duration = Histogram(
"llm_api_duration_seconds",
"LLM API call duration",
["provider", "model"],
buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0]
)
self.llm_tokens_used = Counter(
"llm_tokens_used_total",
"Total tokens used",
["provider", "model", "token_type"] # token_type: prompt, completion
)
# Web API Metrics
self.http_requests_total = Counter(
"http_requests_total",
"Total HTTP requests",
["method", "endpoint", "status_code"]
)
self.http_request_duration = Histogram(
"http_request_duration_seconds",
"HTTP request duration",
["method", "endpoint"],
buckets=[0.01, 0.05, 0.1, 0.5, 1.0, 2.0, 5.0]
)
# Error Metrics
self.errors_total = Counter(
"errors_total",
"Total errors",
["error_type", "component"]
)
# System Info
self.system_info = Info(
"updation_mcp_info",
"System information"
)
self.system_info.info({
"version": "1.0.0",
"environment": settings.environment,
"llm_provider": settings.llm_provider,
})
def track_mcp_request(self, tool_name: str) -> Callable:
"""
Decorator to track MCP tool execution.
Automatically records:
- Request count
- Duration
- Success/failure status
Example:
```python
@metrics.track_mcp_request("get_contracts")
async def get_contracts(...):
...
```
"""
def decorator(func: Callable) -> Callable:
@wraps(func)
async def wrapper(*args, **kwargs) -> Any:
start_time = time.time()
status = "success"
try:
result = await func(*args, **kwargs)
return result
except Exception as e:
status = "error"
self.errors_total.labels(
error_type=type(e).__name__,
component="mcp_tool"
).inc()
raise
finally:
duration = time.time() - start_time
self.mcp_requests_total.labels(
tool_name=tool_name,
status=status
).inc()
self.mcp_request_duration.labels(
tool_name=tool_name
).observe(duration)
return wrapper
return decorator
def track_llm_call(
self,
provider: str,
model: str,
prompt_tokens: int,
completion_tokens: int,
duration: float,
success: bool = True
) -> None:
"""
Record LLM API call metrics.
Args:
provider: LLM provider name (openai, anthropic, google)
model: Model name
prompt_tokens: Input tokens used
completion_tokens: Output tokens generated
duration: API call duration in seconds
success: Whether call succeeded
"""
status = "success" if success else "error"
self.llm_api_calls_total.labels(
provider=provider,
model=model,
status=status
).inc()
self.llm_api_duration.labels(
provider=provider,
model=model
).observe(duration)
if success:
self.llm_tokens_used.labels(
provider=provider,
model=model,
token_type="prompt"
).inc(prompt_tokens)
self.llm_tokens_used.labels(
provider=provider,
model=model,
token_type="completion"
).inc(completion_tokens)
# Global metrics instance
metrics = Metrics()
