write_query
Execute SQL commands on StarRocks for data definition or modification, including DDL and DML statements that do not return a result set.
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 | |
| db | No | database |
Implementation Reference
- src/mcp_server_starrocks/server.py:205-207 (registration)The write_query tool is registered via the @mcp.tool decorator on line 205, making it an MCP tool named 'write_query'. The registration is implicit through FastMCP's decorator pattern.
@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: - The handler function for the 'write_query' tool. It executes a DDL/DML/command SQL via db_client.execute(), logs the result (success/error, rows affected, execution time), and returns a ToolResult with text content (limited to 2000 chars) and structured data.
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()) - Input schema for write_query: 'query' is a required string param (the SQL to execute), and 'db' is an optional string param (database name, defaults to None). Returns a ToolResult.
def write_query(query: Annotated[str, Field(description="SQL to execute")], db: Annotated[str|None, Field(description="database")] = None) -> ToolResult: - The DBClient.execute() method is the underlying database execution helper called by the write_query handler. It handles connection management, database switching, and returns a ResultSet with success status, affected rows, and timing info.
def execute( self, statement: str, db: Optional[str] = None, return_format: Literal["raw", "pandas"] = "raw" ) -> ResultSet: """