InfluxDB MCP Server

from influxdb import InfluxDBClient from .config import ( INFLUXDB_HOST, INFLUXDB_PORT, INFLUXDB_USERNAME, INFLUXDB_PASSWORD, INFLUXDB_SSL, INFLUXDB_VERIFY_SSL, INFLUXDB_TIMEOUT ) class InfluxClient: """Client for interacting with InfluxDB.""" def __init__(self): """Initialize the InfluxDB client with configuration from environment.""" self.client = InfluxDBClient( host=INFLUXDB_HOST, port=INFLUXDB_PORT, username=INFLUXDB_USERNAME, password=INFLUXDB_PASSWORD, ssl=INFLUXDB_SSL, verify_ssl=INFLUXDB_VERIFY_SSL, timeout=INFLUXDB_TIMEOUT ) def list_databases(self): """List all available databases.""" return self.client.get_list_database() def list_measurements(self, database): """List all measurements in a database.""" self.client.switch_database(database) return self.client.get_list_measurements() def execute_query(self, database, query): """ Execute a read-only query against InfluxDB. Args: database (str): The database to query query (str): The InfluxQL query to execute Returns: list: The query results Raises: ValueError: If the query is not read-only """ # Ensure query is read-only (starts with SELECT) query = query.strip() if not query.upper().startswith("SELECT"): raise ValueError("Only SELECT queries are allowed for read-only access") self.client.switch_database(database) return self.client.query(query).raw