MCP Server

import time import logging from prometheus_client import Counter, Histogram, Gauge, generate_latest, CONTENT_TYPE_LATEST from flask import request, g, Response logger = logging.getLogger(__name__) # Prometheus metrics REQUEST_COUNT = Counter( 'mcp_requests_total', 'Total number of requests', ['method', 'endpoint', 'status'] ) REQUEST_DURATION = Histogram( 'mcp_request_duration_seconds', 'Request duration in seconds', ['method', 'endpoint'] ) ACTIVE_CONNECTIONS = Gauge( 'mcp_active_connections', 'Number of active connections' ) TOOL_CALLS = Counter( 'mcp_tool_calls_total', 'Total number of tool calls', ['tool_name', 'status'] ) LLM_REQUESTS = Counter( 'mcp_llm_requests_total', 'Total number of LLM requests', ['model', 'stream'] ) RESOURCE_OPERATIONS = Counter( 'mcp_resource_operations_total', 'Total number of resource operations', ['operation', 'status'] ) def setup_metrics(app): """Setup Prometheus metrics for Flask app""" @app.before_request def before_request(): g.start_time = time.time() ACTIVE_CONNECTIONS.inc() @app.after_request def after_request(response): # Record request metrics duration = time.time() - g.start_time REQUEST_COUNT.labels( method=request.method, endpoint=request.endpoint or 'unknown', status=response.status_code ).inc() REQUEST_DURATION.labels( method=request.method, endpoint=request.endpoint or 'unknown' ).observe(duration) ACTIVE_CONNECTIONS.dec() return response @app.route('/metrics') def metrics(): """Prometheus metrics endpoint""" return Response(generate_latest(), mimetype=CONTENT_TYPE_LATEST) logger.info("Prometheus metrics initialized") def record_tool_call(tool_name: str, success: bool): """Record tool call metrics""" status = 'success' if success else 'error' TOOL_CALLS.labels(tool_name=tool_name, status=status).inc() def record_llm_request(model: str, stream: bool): """Record LLM request metrics""" LLM_REQUESTS.labels(model=model, stream=str(stream).lower()).inc() def record_resource_operation(operation: str, success: bool): """Record resource operation metrics""" status = 'success' if success else 'error' RESOURCE_OPERATIONS.labels(operation=operation, status=status).inc()