fixSql
Read-only
Analyze and fix SQL errors by providing the problematic statement, error message, and database ID to resolve database query issues in Alibaba Cloud DMS.
Instructions
Analyze and fix the SQL error based on the provided SQL statement, error message, and database ID.If you don't know the databaseId, first use getDatabase or searchDatabase to retrieve it. (1) If you have the exact host, port, and database name, use getDatabase. (2) If you only know the database name, use searchDatabase. (3) If you don't know any information, ask the user to provide the necessary details. Note: searchDatabase may return multiple databases. In this case, let the user choose which one to use.
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| database_id | Yes | DMS databaseId | |
| question | No | Natural language question | |
| sql | Yes | The SQL that caused an error | |
| error | Yes | SQL error message | |
| model | No | Optional: if a specific model is desired, it can be specified here |
Implementation Reference
- Primary handler function implementing the core logic of the 'fixSql' tool by calling the Alibaba Cloud DMS API client.fix_sql_by_meta_agent with the provided parameters.
async def fix_sql_statement( database_id: str = Field(description="DMS databaseId"), question: Optional[str] = Field(default=None, description="Natural language question"), sql: str = Field(description="The SQL that caused an error"), error: str = Field(description="SQL error message"), model: Optional[str] = Field(default=None, description="Optional: if a specific model is desired, it can be specified here") ) -> Dict[str, Any]: client = create_client() req = dms_enterprise_20181101_models.FixSqlByMetaAgentRequest(db_id=database_id, query=question, sql=sql, error=error) # if mcp.state.real_login_uid: # req.real_login_user_uid = mcp.state.real_login_uid if model: req.model = model try: resp = client.fix_sql_by_meta_agent(req) if not resp or not resp.body: return None data = resp.body.to_map() return data except Exception as e: logger.error(f"Error in fix_sql_statement: {e}") raise - src/alibabacloud_dms_mcp_server/server.py:660-672 (registration)Registration of the 'fixSql' tool in configured mode (when default_database_id is set), using a wrapper function that calls the primary handler.
@self.mcp.tool(name="fixSql", description="Analyze and fix the SQL error based on the provided SQL statement and error message.", annotations={"title": "SQL修复", "readOnlyHint": True, "destructiveHint": False}) async def fix_sql_configured( question: Optional[str] = Field(default=None, description="Natural language question"), sql: str = Field(description="The SQL that caused an error"), error: str = Field(description="SQL error message"), model: Optional[str] = Field(default=None, description="Optional: if a specific model is desired, it can be specified here") ) -> Dict[str, Any]: result_obj = await fix_sql_statement(database_id=self.default_database_id, question=question, sql=sql, error=error, model=model) return result_obj - src/alibabacloud_dms_mcp_server/server.py:750-753 (registration)Direct registration of the primary 'fix_sql_statement' handler as the 'fixSql' tool in full toolset mode (when no default database is configured).
self.mcp.tool(name="fixSql", description=f"Analyze and fix the SQL error based on the provided " f"SQL statement, error message, and database ID." f"{DATABASE_ID_DESCRIPTION}", annotations={"title": "SQL修复", "readOnlyHint": True})(fix_sql_statement)