write_query
Execute DDL/DML or other StarRocks commands without a ResultSet, enabling direct SQL operations on the StarRocks database through the MCP server interface.
Instructions
Execute a DDL/DML or other StarRocks command that do not have a ResultSet
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | SQL to execute |
Implementation Reference
- The complete implementation of the 'write_query' tool handler, including decorator (registration and schema), function definition, execution logic using db_client.execute for non-SELECT queries, logging, and returning ToolResult with results.
@mcp.tool(description="Execute a DDL/DML or other StarRocks command that do not have a ResultSet" + description_suffix) def write_query(query: Annotated[str, Field(description="SQL to execute")], db: Annotated[str|None, Field(description="database")] = None) -> ToolResult: logger.info(f"Executing write query: {query[:100]}{'...' if len(query) > 100 else ''}") result = db_client.execute(query, db=db) if not result.success: logger.error(f"Write query failed: {result.error_message}") elif result.rows_affected is not None and result.rows_affected >= 0: logger.info(f"Write query executed successfully, {result.rows_affected} rows affected in {result.execution_time:.2f}s") else: logger.info(f"Write query executed successfully in {result.execution_time:.2f}s") return ToolResult(content=[TextContent(type='text', text=result.to_string(limit=2000))], structured_content=result.to_dict())