test-connection
Verify Apache Pinot connection status and retrieve diagnostics to ensure proper integration and functionality with the StarTree MCP Server.
Instructions
Test Pinot connection and return diagnostics
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- mcp_pinot/server.py:48-55 (handler)MCP tool handler for 'test-connection'. Decorated with @mcp.tool, calls pinot_client.test_connection() and serializes result to JSON.
@mcp.tool def test_connection() -> str: """Test Pinot connection and return diagnostics""" try: results = pinot_client.test_connection() return json.dumps(results, indent=2) except Exception as e: return f"Error: {str(e)}" - mcp_pinot/pinot_client.py:162-208 (helper)Core implementation of connection testing in PinotClient class. Performs connection validation, sample query execution, table listing, and returns comprehensive diagnostics.
def test_connection(self) -> dict[str, Any]: """Test the connection and return diagnostic information""" result = { "connection_test": False, "query_test": False, "tables_test": False, "error": None, "config": { "broker_host": self.config.broker_host, "broker_port": self.config.broker_port, "broker_scheme": self.config.broker_scheme, "controller_url": self.config.controller_url, "database": self.config.database, "use_msqe": self.config.use_msqe, "has_token": bool(self.config.token), "has_username": bool(self.config.username), "timeout_config": { "connection": self.config.connection_timeout, "request": self.config.request_timeout, "query": self.config.query_timeout, }, }, } try: # Test basic connection conn = self.get_connection() result["connection_test"] = True # Test simple query curs = conn.cursor() curs.execute("SELECT 1 as test_column") test_result = curs.fetchall() result["query_test"] = True result["query_result"] = test_result # Test tables listing tables = self.get_tables() result["tables_test"] = True result["tables_count"] = len(tables) result["sample_tables"] = tables[:5] if tables else [] except Exception as e: result["error"] = str(e) logger.error(f"Connection test failed: {e}") return result - mcp_pinot/pinot_client.py:25-30 (helper)Utility function to test database connection by executing a simple SELECT 1 query. Used internally by PinotClient.
def test_connection_query(connection) -> None: """Test connection with a simple query""" test_cursor = connection.cursor() test_cursor.execute("SELECT 1") test_result = test_cursor.fetchall() logger.debug(f"Connection test successful: {test_result}")