StarRocks MCP Server

Instructions

Implementation Reference

• src/mcp_server_starrocks/server.py:324-404 (handler)

  Main handler function for the 'query_and_plotly_chart' tool. Executes an SQL query to obtain data as a Pandas DataFrame, validates and evaluates a Plotly Express expression to generate a chart, and returns the chart as a base64-encoded image (jpeg/png) or JSON for UI rendering, along with query results.
  @mcp.tool(description="using sql `query` to extract data from database, then using python `plotly_expr` to generate a chart for UI to display" + description_suffix) def query_and_plotly_chart( query: Annotated[str, Field(description="SQL query to execute")], plotly_expr: Annotated[ str, Field(description="a one function call expression, with 2 vars binded: `px` as `import plotly.express as px`, and `df` as dataframe generated by query `plotly_expr` example: `px.scatter(df, x=\"sepal_width\", y=\"sepal_length\", color=\"species\", marginal_y=\"violin\", marginal_x=\"box\", trendline=\"ols\", template=\"simple_white\")`")], format: Annotated[str, Field(description="chart output format, json|png|jpeg")] = "jpeg", db: Annotated[str|None, Field(description="database")] = None ) -> ToolResult: """ Executes an SQL query, creates a Pandas DataFrame, generates a Plotly chart using the provided expression, encodes the chart as a base64 PNG image, and returns it along with optional text. Args: query: The SQL query string to execute. plotly_expr: A Python string expression using 'px' (plotly.express) and 'df' (the DataFrame from the query) to generate a figure. Example: "px.scatter(df, x='col1', y='col2')" format: chat output format, json|png|jpeg, default is jpeg db: Optional database name to execute the query in. Returns: A list containing types.TextContent and types.ImageContent, or just types.TextContent in case of an error or no data. """ try: logger.info(f'query_and_plotly_chart query:{one_line_summary(query)}, plotly:{one_line_summary(plotly_expr)} format:{format}, db:{db}') result = db_client.execute(query, db=db, return_format="pandas") errmsg = None if not result.success: errmsg = result.error_message elif result.pandas is None: errmsg = 'Query did not return data suitable for plotting.' else: df = result.pandas if df.empty: errmsg = 'Query returned no data to plot.' if errmsg: logger.warning(f"Query or data issue: {errmsg}") return ToolResult( content=[TextContent(type='text', text=f'Error: {errmsg}')], structured_content={'success': False, 'error_message': errmsg}, ) # Validate and evaluate the plotly expression using px and df local_vars = {'df': df} validate_plotly_expr(plotly_expr) fig : plotly.graph_objs.Figure = eval(plotly_expr, {"px": px}, local_vars) if format == 'json': # return json representation of the figure for front-end rendering plot_json = json.loads(fig.to_json()) structured_content = result.to_dict() structured_content['data'] = plot_json['data'] structured_content['layout'] = plot_json['layout'] summary = result.to_string() return ToolResult( content=[ TextContent(type='text', text=f'{summary}\nChart Generated for UI rendering'), ], structured_content=structured_content, ) else: if not hasattr(fig, 'to_image'): raise ToolError(f"The evaluated expression did not return a Plotly figure object. Result type: {type(fig)}") if format == 'jpg': format = 'jpeg' img_bytes = fig.to_image(format=format, width=960, height=720) structured_content = result.to_dict() structured_content['img_bytes_base64'] = base64.b64encode(img_bytes) return ToolResult( content=[ TextContent(type='text', text=f'dataframe data:\n{df}\nChart generated but for UI only'), Image(data=img_bytes, format="jpeg").to_image_content() ], structured_content=structured_content ) except Exception as err: return ToolResult( content=[TextContent(type='text', text=f'Error: {err}')], structured_content={'success': False, 'error_message': str(err)}, )
• src/mcp_server_starrocks/server.py:326-331 (schema)

  Pydantic input schema defined via Annotated[ ] and Field( ) for the tool parameters: query (str), plotly_expr (str), format (str default 'jpeg'), db (str|None).
  query: Annotated[str, Field(description="SQL query to execute")], plotly_expr: Annotated[ str, Field(description="a one function call expression, with 2 vars binded: `px` as `import plotly.express as px`, and `df` as dataframe generated by query `plotly_expr` example: `px.scatter(df, x=\"sepal_width\", y=\"sepal_length\", color=\"species\", marginal_y=\"violin\", marginal_x=\"box\", trendline=\"ols\", template=\"simple_white\")`")], format: Annotated[str, Field(description="chart output format, json|png|jpeg")] = "jpeg", db: Annotated[str|None, Field(description="database")] = None ) -> ToolResult:
• src/mcp_server_starrocks/server.py:324-324 (registration)

  Registration of the tool using FastMCP @mcp.tool decorator with description.
  @mcp.tool(description="using sql `query` to extract data from database, then using python `plotly_expr` to generate a chart for UI to display" + description_suffix)
• src/mcp_server_starrocks/server.py:256-315 (helper)

  Helper function to securely validate the plotly_expr input, ensuring it's a safe single px.* call without nested calls or malicious code using ast parsing.
  def validate_plotly_expr(expr: str): """ Validates a string to ensure it represents a single call to a method of the 'px' object, without containing other statements or imports, and ensures its arguments do not contain nested function calls. Args: expr: The string expression to validate. Raises: ValueError: If the expression does not meet the security criteria. SyntaxError: If the expression is not valid Python syntax. """ # 1. Check for valid Python syntax try: tree = ast.parse(expr) except SyntaxError as e: raise SyntaxError(f"Invalid Python syntax in expression: {e}") from e # 2. Check that the tree contains exactly one top-level node (statement/expression) if len(tree.body) != 1: raise ValueError("Expression must be a single statement or expression.") node = tree.body[0] # 3. Check that the single node is an expression if not isinstance(node, ast.Expr): raise ValueError( "Expression must be a single expression, not a statement (like assignment, function definition, import, etc.).") # 4. Get the actual value of the expression and check it's a function call expr_value = node.value if not isinstance(expr_value, ast.Call): raise ValueError("Expression must be a function call.") # 5. Check that the function being called is an attribute lookup (like px.scatter) if not isinstance(expr_value.func, ast.Attribute): raise ValueError("Function call must be on an object attribute (e.g., px.scatter).") # 6. Check that the attribute is being accessed on a simple variable name if not isinstance(expr_value.func.value, ast.Name): raise ValueError("Function call must be on a simple variable name (e.g., px.scatter, not obj.px.scatter).") # 7. Check that the simple variable name is 'px' if expr_value.func.value.id != 'px': raise ValueError("Function call must be on the 'px' object.") # Check positional arguments for i, arg_node in enumerate(expr_value.args): for sub_node in ast.walk(arg_node): if isinstance(sub_node, ast.Call): raise ValueError(f"Positional argument at index {i} contains a disallowed nested function call.") # Check keyword arguments for kw in expr_value.keywords: for sub_node in ast.walk(kw.value): if isinstance(sub_node, ast.Call): keyword_name = kw.arg if kw.arg else '<unknown>' raise ValueError(f"Keyword argument '{keyword_name}' contains a disallowed nested function call.")
• src/mcp_server_starrocks/server.py:316-322 (helper)

  Helper utility to create a one-line truncated summary of text, used for logging the query and plotly_expr.
  def one_line_summary(text: str, limit:int=100) -> str: """Generate a one-line summary of the given text, truncated to the specified limit.""" single_line = ' '.join(text.split()) if len(single_line) > limit: return single_line[:limit-3] + '...' return single_line