execute_query
Run SQL queries on StarRocks databases to retrieve data and analyze information through Claude AI assistants.
Instructions
Execute a SQL query against StarRocks and return results.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| sql | Yes | SQL query to execute |
Implementation Reference
- src/starrocks_mcp/client.py:61-82 (handler)The actual implementation of execute_query which interacts with the StarRocks database using pymysql.
def execute_query(self, query: str) -> List[Dict[str, Any]]: """ Execute SQL query and return results. Args: query: SQL query string Returns: List of dictionaries representing rows """ if not self._connection: self.connect() try: with self._connection.cursor(pymysql.cursors.DictCursor) as cursor: cursor.execute(query) results = cursor.fetchall() logger.info(f"Query executed successfully, returned {len(results)} rows") return results except Exception as e: logger.error(f"Query execution failed: {e}") raise - The MCP tool handler logic that calls StarRocksClient.execute_query.
if name == "execute_query": sql = arguments.get("sql") if not sql: raise ValueError("SQL query is required") validate_sql(sql) results = client.execute_query(sql) truncated_results, was_truncated = truncate_results(results) result_text = f"Query executed successfully. Returned {len(results)} rows" if was_truncated: result_text += f" (showing first {len(truncated_results)} rows)" result_text += f":\n\n{format_json(truncated_results)}" return [TextContent(type="text", text=result_text)] - The tool registration and schema definition for execute_query.
Tool( name="execute_query", description="Execute a SQL query against StarRocks and return results.", inputSchema={ "type": "object", "properties": { "sql": { "type": "string", "description": "SQL query to execute", }, }, "required": ["sql"], }, ),