execute_sql

Instructions

Implementation Reference

• src/mysql_mcp_server/server.py:109-149 (handler)

  The @app.call_tool() handler function that implements the logic for the 'execute_sql' tool, including SQL execution, result handling, and error management.

  @app.call_tool()
  async def call_tool(name: str, arguments: dict) -> list[TextContent]:
      """Execute SQL commands."""
      config = get_db_config()
      logger.info(f"Calling tool: {name} with arguments: {arguments}")
      
      if name != "execute_sql":
          raise ValueError(f"Unknown tool: {name}")
      
      query = arguments.get("query")
      if not query:
          raise ValueError("Query is required")
      
      try:
          with connect(**config) as conn:
              with conn.cursor() as cursor:
                  cursor.execute(query)
                  
                  # Special handling for SHOW TABLES
                  if query.strip().upper().startswith("SHOW TABLES"):
                      tables = cursor.fetchall()
                      result = ["Tables_in_" + config["database"]]  # Header
                      result.extend([table[0] for table in tables])
                      return [TextContent(type="text", text="\n".join(result))]
                  
                  # Handle all other queries that return result sets (SELECT, SHOW, DESCRIBE etc.)
                  elif cursor.description:
                      columns = [desc[0] for desc in cursor.description]
                      rows = cursor.fetchall()
                      result = [",".join(map(str, row)) for row in rows]
                      return [TextContent(type="text", text="\n".join([",".join(columns)] + result))]
                  
                  # Non-SELECT queries
                  else:
                      conn.commit()
                      return [TextContent(type="text", text=f"Query executed successfully. Rows affected: {cursor.rowcount}")]
                  
      except Error as e:
          logger.error(f"Error executing SQL '{query}': {e}")
          return [TextContent(type="text", text=f"Error executing query: {str(e)}")]

• src/mysql_mcp_server/server.py:96-105 (schema)

  The input schema definition for the 'execute_sql' tool, specifying the required 'query' parameter.

  inputSchema={
      "type": "object",
      "properties": {
          "query": {
              "type": "string",
              "description": "The SQL query to execute"
          }
      },
      "required": ["query"]
  }

• src/mysql_mcp_server/server.py:92-107 (registration)

  The registration of the 'execute_sql' tool in the list_tools() function, including name, description, and schema.

  return [
      Tool(
          name="execute_sql",
          description="Execute an SQL query on the MySQL server",
          inputSchema={
              "type": "object",
              "properties": {
                  "query": {
                      "type": "string",
                      "description": "The SQL query to execute"
                  }
              },
              "required": ["query"]
          }
      )
  ]