diff --git a/AUTHORS.rst b/AUTHORS.rst
index 5ba382f..d4cdc95 100644
--- a/AUTHORS.rst
+++ b/AUTHORS.rst
@@ -1,6 +1,7 @@
 The following individuals have contributed code to agate-sql:
 
 * `Christopher Groskopf <https://github.com/onyxfish>`_
+* `AC <https://github.com/ac1982>`_
 * `Adrian Klaver <https://github.com/aklaver>`_
 * `James McKinney <https://github.com/jpmckinney>`_
 * `Chris Keller <https://github.com/chrislkeller>`_
diff --git a/CHANGELOG.rst b/CHANGELOG.rst
index fe9fd3a..ab05d94 100644
--- a/CHANGELOG.rst
+++ b/CHANGELOG.rst
@@ -1,3 +1,8 @@
+Unreleased
+----------
+
+* Fix MySQL ``REPLACE`` prefixes to generate ``REPLACE INTO`` statements.
+
 0.7.3 - December 15, 2025
 -------------------------
 
diff --git a/agatesql/table.py b/agatesql/table.py
index 86daa6f..890b2e2 100644
--- a/agatesql/table.py
+++ b/agatesql/table.py
@@ -13,8 +13,9 @@
 from sqlalchemy.dialects.oracle import INTERVAL as ORACLE_INTERVAL
 from sqlalchemy.dialects.postgresql import INTERVAL as POSTGRES_INTERVAL
 from sqlalchemy.engine import Connection
+from sqlalchemy.ext.compiler import compiles
 from sqlalchemy.schema import CreateTable
-from sqlalchemy.sql import select
+from sqlalchemy.sql import Insert, select
 from sqlalchemy.types import BOOLEAN, DATE, DATETIME, DECIMAL, FLOAT, TEXT, TIMESTAMP, VARCHAR, Interval
 
 SQL_TYPE_MAP = {
@@ -45,6 +46,18 @@
 }
 
 
+class MySQLReplace(Insert):
+    """A MySQL replacement using SQLAlchemy's parameter and column handling."""
+
+    inherit_cache = True
+
+
+@compiles(MySQLReplace, 'mysql')
+def compile_mysql_replace(statement, compiler, **kwargs):
+    """Replace only the leading INSERT keyword, preserving bound parameters."""
+    return compiler.visit_insert(statement, **kwargs).replace('INSERT ', 'REPLACE ', 1)
+
+
 def get_engine_and_connection(connection_or_string=None):
     """
     Gets a connection to a specific SQL alchemy backend. If an existing
@@ -260,7 +273,8 @@ def to_sql(self, connection_or_string, table_name, overwrite=False,
     :param insert:
         Insert table data.
     :param prefixes:
-        Add prefixes to the insert query.
+        Add prefixes to the insert query. On MySQL, ``REPLACE`` uses
+        ``REPLACE INTO`` instead of ``INSERT INTO``.
     :param db_schema:
         Create table in the specified database schema.
     :param constraints:
@@ -289,6 +303,10 @@ def to_sql(self, connection_or_string, table_name, overwrite=False,
 
     if insert:
         insert = sql_table.insert()
+        prefixes = list(prefixes)
+        if dialect == 'mysql' and any(isinstance(prefix, str) and prefix.upper() == 'REPLACE' for prefix in prefixes):
+            insert = MySQLReplace(sql_table)
+            prefixes = [prefix for prefix in prefixes if not (isinstance(prefix, str) and prefix.upper() == 'REPLACE')]
         for prefix in prefixes:
             insert = insert.prefix_with(prefix)
         if chunk_size is None:
diff --git a/tests/test_agatesql.py b/tests/test_agatesql.py
index b6b0e03..aeede05 100644
--- a/tests/test_agatesql.py
+++ b/tests/test_agatesql.py
@@ -1,8 +1,11 @@
 from decimal import Decimal
 from textwrap import dedent
+from unittest.mock import Mock
 
 import agate
 from sqlalchemy import create_engine
+from sqlalchemy.dialects import mysql
+from sqlalchemy.engine import Connection
 from sqlalchemy.exc import IntegrityError
 
 import agatesql
@@ -102,6 +105,36 @@ def test_prefixes(self):
         self.assertEqual(len(table.rows), len(self.table.rows) - 1)
         self.assertSequenceEqual(table.rows[1], self.table.rows[2])
 
+    def test_mysql_replace(self):
+        for chunk_size in (None, 1):
+            for prefixes in (['REPLACE'], ['replace'], ['LOW_PRIORITY', 'REPLACE']):
+                with self.subTest(chunk_size=chunk_size, prefixes=prefixes):
+                    original_prefixes = prefixes[:]
+                    connection = Mock(spec=Connection)
+                    connection.engine = Mock(dialect=mysql.dialect())
+                    self.table.to_sql(connection, 'insert table', create=False,
+                                      prefixes=prefixes, chunk_size=chunk_size)
+                    expected = 'REPLACE LOW_PRIORITY INTO' if 'LOW_PRIORITY' in prefixes else 'REPLACE INTO'
+                    calls = connection.execute.call_args_list
+                    self.assertEqual(len(calls), 1 if chunk_size is None else len(self.rows))
+                    for call in calls:
+                        statement, parameters = call.args
+                        sql = str(statement.compile(dialect=connection.engine.dialect))
+                        self.assertTrue(sql.startswith(expected + ' `insert table`'), sql)
+                        self.assertNotIn('INSERT REPLACE', sql)
+                        self.assertTrue(parameters)
+                    self.assertEqual(prefixes, original_prefixes)
+
+    def test_mysql_insert_prefixes(self):
+        for prefixes, expected in (([], 'INSERT INTO'), (['IGNORE'], 'INSERT IGNORE INTO')):
+            with self.subTest(prefixes=prefixes):
+                connection = Mock(spec=Connection)
+                connection.engine = Mock(dialect=mysql.dialect())
+                self.table.to_sql(connection, 'test', create=False, prefixes=prefixes)
+                statement = connection.execute.call_args.args[0]
+                sql = str(statement.compile(dialect=connection.engine.dialect))
+                self.assertTrue(sql.startswith(expected + ' test'), sql)
+
     def test_to_sql_create_statement(self):
         statement = self.table.to_sql_create_statement('test_table')