diff --git a/datasette/utils/__init__.py b/datasette/utils/__init__.py
index 9d1894592f..039533eb63 100644
--- a/datasette/utils/__init__.py
+++ b/datasette/utils/__init__.py
@@ -1038,6 +1038,11 @@ def getlist(self, name):
         """Return full list"""
         return self._data.get(name) or []
 
+    def items(self):
+        """Yield (key, first_value) pairs, matching ``__getitem__`` semantics."""
+        for key, values in self._data.items():
+            yield key, values[0]
+
 
 class ConnectionProblem(Exception):
     pass
diff --git a/datasette/utils/asgi.py b/datasette/utils/asgi.py
index 35f243b6a3..b531cf5537 100644
--- a/datasette/utils/asgi.py
+++ b/datasette/utils/asgi.py
@@ -17,7 +17,7 @@
     DEFAULT_MIN_FREE_DISK_BYTES,
 )
 from mimetypes import guess_type
-from urllib.parse import parse_qs, urlunparse, parse_qsl
+from urllib.parse import parse_qs, urlunparse
 from pathlib import Path
 from http.cookies import SimpleCookie, Morsel
 import aiofiles
@@ -153,7 +153,7 @@ async def post_body(self):
 
     async def post_vars(self):
         body = await self.post_body()
-        return dict(parse_qsl(body.decode("utf-8"), keep_blank_values=True))
+        return MultiParams(parse_qs(qs=body.decode("utf-8"), keep_blank_values=True))
 
     async def form(
         self,
diff --git a/datasette/views/query_helpers.py b/datasette/views/query_helpers.py
index f30a30bc1a..f114ba6626 100644
--- a/datasette/views/query_helpers.py
+++ b/datasette/views/query_helpers.py
@@ -279,9 +279,9 @@ def _execute_write_disabled_reason(sql, analysis_error, analysis_rows):
 
 
 def _coerce_execute_write_payload(data, is_json):
-    if not isinstance(data, dict):
-        raise QueryValidationError("JSON must be a dictionary")
     if is_json:
+        if not isinstance(data, dict):
+            raise QueryValidationError("JSON must be a dictionary")
         invalid_keys = set(data) - {"sql", "params"}
         if invalid_keys:
             raise QueryValidationError(
diff --git a/datasette/views/stored_queries.py b/datasette/views/stored_queries.py
index 8c4e849ebb..73133627f9 100644
--- a/datasette/views/stored_queries.py
+++ b/datasette/views/stored_queries.py
@@ -357,14 +357,17 @@ async def post(self, request):
         query_data = {}
         try:
             data, is_json = await _json_or_form_payload(request)
-            if not isinstance(data, dict):
-                raise QueryValidationError("JSON must be a dictionary")
-            query_data = data.get("query") if is_json else data
-            if not isinstance(query_data, dict):
-                raise QueryValidationError("JSON must contain a query dictionary")
+            if is_json:
+                if not isinstance(data, dict):
+                    raise QueryValidationError("JSON must be a dictionary")
+                query_data = data.get("query")
+                if not isinstance(query_data, dict):
+                    raise QueryValidationError("JSON must contain a query dictionary")
+            else:
+                query_data = data
             prepared = await _prepare_query_create(self.ds, request, db, query_data)
         except QueryValidationError as ex:
-            if not is_json and isinstance(query_data, dict):
+            if not is_json:
                 return await self._error_response(
                     request, db, query_data, ex.message, ex.status
                 )
@@ -375,7 +378,7 @@ async def post(self, request):
         try:
             await self.ds.add_query(db.name, name, replace=False, **prepared)
         except sqlite3.IntegrityError as ex:
-            if not is_json and isinstance(query_data, dict):
+            if not is_json:
                 return await self._error_response(request, db, query_data, str(ex), 400)
             return _error([str(ex)], 400)
 
@@ -426,8 +429,8 @@ async def post(self, request):
             return _error(["Trusted queries cannot be updated using the API"], 403)
 
         try:
-            data, _ = await _json_or_form_payload(request)
-            if not isinstance(data, dict):
+            data, is_json = await _json_or_form_payload(request)
+            if is_json and not isinstance(data, dict):
                 raise QueryValidationError("JSON must be a dictionary")
             invalid_keys = set(data) - {"update", "return"}
             if invalid_keys:
diff --git a/docs/changelog.rst b/docs/changelog.rst
index 4f9ffdbbcd..d27439b027 100644
--- a/docs/changelog.rst
+++ b/docs/changelog.rst
@@ -43,6 +43,7 @@ Bug fixes
 
 - Fixed a bug where visiting ``/<database>/-/query`` without a ``?sql=`` parameter returned a 500 error. (:issue:`2743`)
 - The ``datasette inspect`` command now correctly records row counts for tables with more than 10,000 rows. (:issue:`2712`)
+- ``await request.post_vars()`` now returns a :ref:`MultiParams <internals_multiparams>` object instead of a ``dict``, so multiple values for the same form field are preserved. Use ``.getlist(key)`` to retrieve every value. Existing ``post_vars[key]`` access continues to work, but now returns the *first* submitted value rather than the *last* (matching ``request.args`` semantics). (:issue:`2425`)
 
 .. _v1_0_a30:
 
diff --git a/docs/internals.rst b/docs/internals.rst
index 4980ee8bb5..036dbf8405 100644
--- a/docs/internals.rst
+++ b/docs/internals.rst
@@ -103,8 +103,8 @@ The object also has the following awaitable methods:
 
     Don't forget to read about :ref:`internals_csrf`!
 
-``await request.post_vars()`` - dictionary
-    Returns a dictionary of form variables that were submitted in the request body via ``POST`` using ``application/x-www-form-urlencoded`` encoding. For multipart forms or file uploads, use ``request.form()`` instead.
+``await request.post_vars()`` - MultiParams
+    Returns a :ref:`MultiParams <internals_multiparams>` object of form variables that were submitted in the request body via ``POST`` using ``application/x-www-form-urlencoded`` encoding. This has the same shape as ``request.args``, so use ``.getlist(key)`` to retrieve every value submitted for keys with multiple values (such as ``<select multiple>`` fields). For multipart forms or file uploads, use ``request.form()`` instead.
 
 ``await request.post_body()`` - bytes
     Returns the un-parsed body of a request submitted by ``POST`` - useful for things like incoming JSON data.
diff --git a/tests/plugins/my_plugin.py b/tests/plugins/my_plugin.py
index 1d2378b4ed..8eacd6b1c4 100644
--- a/tests/plugins/my_plugin.py
+++ b/tests/plugins/my_plugin.py
@@ -237,7 +237,8 @@ async def post(request):
         if request.method == "GET":
             return Response.html(request.scope["csrftoken"]())
         else:
-            return Response.json(await request.post_vars())
+            # post_vars() returns a MultiParams; convert to a plain dict for JSON
+            return Response.json(dict(await request.post_vars()))
 
     async def csrftoken_form(request, datasette):
         return Response.html(
diff --git a/tests/test_internals_request.py b/tests/test_internals_request.py
index d1ca1f46be..b5814cb5cd 100644
--- a/tests/test_internals_request.py
+++ b/tests/test_internals_request.py
@@ -3,8 +3,7 @@
 import pytest
 
 
-@pytest.mark.asyncio
-async def test_request_post_vars():
+def _post_request(body: bytes) -> Request:
     scope = {
         "http_version": "1.1",
         "method": "POST",
@@ -19,12 +18,44 @@ async def test_request_post_vars():
     async def receive():
         return {
             "type": "http.request",
-            "body": b"foo=bar&baz=1&empty=",
+            "body": body,
             "more_body": False,
         }
 
-    request = Request(scope, receive)
-    assert {"foo": "bar", "baz": "1", "empty": ""} == await request.post_vars()
+    return Request(scope, receive)
+
+
+@pytest.mark.asyncio
+async def test_request_post_vars():
+    request = _post_request(b"foo=bar&baz=1&empty=")
+    post_vars = await request.post_vars()
+    assert post_vars["foo"] == "bar"
+    assert post_vars["baz"] == "1"
+    assert post_vars["empty"] == ""
+    assert post_vars.get("missing") is None
+    assert set(post_vars.keys()) == {"foo", "baz", "empty"}
+    assert dict(post_vars.items()) == {"foo": "bar", "baz": "1", "empty": ""}
+
+
+@pytest.mark.asyncio
+async def test_request_post_vars_multi():
+    # post_vars() returns a MultiParams so multiple values for the same key are
+    # preserved, matching the behaviour of request.args. See issue #2425.
+    request = _post_request(b"multi=1&multi=2&single=3")
+    post_vars = await request.post_vars()
+    assert post_vars.get("multi") == "1"
+    assert post_vars.get("single") == "3"
+    assert post_vars["multi"] == "1"
+    assert post_vars["single"] == "3"
+    assert post_vars.getlist("multi") == ["1", "2"]
+    assert post_vars.getlist("single") == ["3"]
+    assert post_vars.getlist("missing") == []
+    assert "multi" in post_vars
+    assert "missing" not in post_vars
+    assert list(post_vars.keys()) == ["multi", "single"]
+    assert len(post_vars) == 2
+    with pytest.raises(KeyError):
+        post_vars["missing"]
 
 
 @pytest.mark.asyncio