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
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | SQL to execute |
Input Schema (JSON Schema)
{
"properties": {
"query": {
"description": "SQL to execute",
"title": "Query",
"type": "string"
}
},
"required": [
"query"
],
"type": "object"
}
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())