diff --git a/airflow/api_connexion/endpoints/task_instance_endpoint.py b/airflow/api_connexion/endpoints/task_instance_endpoint.py
index c4f464a754940..60f847d8979fa 100644
--- a/airflow/api_connexion/endpoints/task_instance_endpoint.py
+++ b/airflow/api_connexion/endpoints/task_instance_endpoint.py
@@ -20,6 +20,7 @@
 from marshmallow import ValidationError
 from sqlalchemy import and_, func
 from sqlalchemy.orm.exc import NoResultFound
+from sqlalchemy.sql.expression import or_
 
 from airflow.api_connexion import security
 from airflow.api_connexion.exceptions import BadRequest, NotFound
@@ -73,9 +74,16 @@ def get_task_instance(dag_id: str, dag_run_id: str, task_id: str, session=None):
     return task_instance_schema.dump(task_instance)
 
 
+def _convert_state(states):
+    if not states:
+        return None
+    return [State.NONE if s == "none" else s for s in states]
+
+
 def _apply_array_filter(query, key, values):
     if values is not None:
-        query = query.filter(key.in_(values))
+        cond = ((key == v) for v in values)
+        query = query.filter(or_(*cond))
     return query
 
 
@@ -118,13 +126,16 @@ def get_task_instances(
     end_date_lte: Optional[str] = None,
     duration_gte: Optional[float] = None,
     duration_lte: Optional[float] = None,
-    state: Optional[str] = None,
+    state: Optional[List[str]] = None,
     pool: Optional[List[str]] = None,
     queue: Optional[List[str]] = None,
     offset: Optional[int] = None,
     session=None,
 ):
     """Get list of task instances."""
+    # Because state can be 'none'
+    states = _convert_state(state)
+
     base_query = session.query(TI).join(TI.dag_run)
 
     if dag_id != "~":
@@ -141,7 +152,7 @@ def get_task_instances(
     )
     base_query = _apply_range_filter(base_query, key=TI.end_date, value_range=(end_date_gte, end_date_lte))
     base_query = _apply_range_filter(base_query, key=TI.duration, value_range=(duration_gte, duration_lte))
-    base_query = _apply_array_filter(base_query, key=TI.state, values=state)
+    base_query = _apply_array_filter(base_query, key=TI.state, values=states)
     base_query = _apply_array_filter(base_query, key=TI.pool, values=pool)
     base_query = _apply_array_filter(base_query, key=TI.queue, values=queue)
 
@@ -180,6 +191,7 @@ def get_task_instances_batch(session=None):
         data = task_instance_batch_form.load(body)
     except ValidationError as err:
         raise BadRequest(detail=str(err.messages))
+    states = _convert_state(data['state'])
     base_query = session.query(TI).join(TI.dag_run)
 
     base_query = _apply_array_filter(base_query, key=TI.dag_id, values=data["dag_ids"])
@@ -199,7 +211,7 @@ def get_task_instances_batch(session=None):
     base_query = _apply_range_filter(
         base_query, key=TI.duration, value_range=(data["duration_gte"], data["duration_lte"])
     )
-    base_query = _apply_array_filter(base_query, key=TI.state, values=data["state"])
+    base_query = _apply_array_filter(base_query, key=TI.state, values=states)
     base_query = _apply_array_filter(base_query, key=TI.pool, values=data["pool"])
     base_query = _apply_array_filter(base_query, key=TI.queue, values=data["queue"])
 
diff --git a/airflow/api_connexion/openapi/v1.yaml b/airflow/api_connexion/openapi/v1.yaml
index a4d75c5357001..5ff393714cce2 100644
--- a/airflow/api_connexion/openapi/v1.yaml
+++ b/airflow/api_connexion/openapi/v1.yaml
@@ -3237,7 +3237,7 @@ components:
         state:
           type: array
           items:
-            type: string
+            $ref: '#/components/schemas/TaskState'
           description:
             The value can be repeated to retrieve multiple matching values (OR condition).
         pool:
diff --git a/airflow/api_connexion/schemas/task_instance_schema.py b/airflow/api_connexion/schemas/task_instance_schema.py
index 956246b9f6d1d..76d284dd20d6f 100644
--- a/airflow/api_connexion/schemas/task_instance_schema.py
+++ b/airflow/api_connexion/schemas/task_instance_schema.py
@@ -43,7 +43,7 @@ class Meta:
     start_date = auto_field()
     end_date = auto_field()
     duration = auto_field()
-    state = auto_field()
+    state = TaskInstanceStateField()
     _try_number = auto_field(data_key="try_number")
     max_tries = auto_field()
     hostname = auto_field()
diff --git a/tests/api_connexion/endpoints/test_task_instance_endpoint.py b/tests/api_connexion/endpoints/test_task_instance_endpoint.py
index 2a2e0f10dbb27..4e887c50a4e1b 100644
--- a/tests/api_connexion/endpoints/test_task_instance_endpoint.py
+++ b/tests/api_connexion/endpoints/test_task_instance_endpoint.py
@@ -370,13 +370,26 @@ class TestGetTaskInstances(TestTaskInstanceEndpoint):
                     {"state": State.RUNNING},
                     {"state": State.QUEUED},
                     {"state": State.SUCCESS},
+                    {"state": State.NONE},
                 ],
                 False,
                 (
                     "/api/v1/dags/example_python_operator/dagRuns/"
-                    "TEST_DAG_RUN_ID/taskInstances?state=running,queued"
+                    "TEST_DAG_RUN_ID/taskInstances?state=running,queued,none"
                 ),
-                2,
+                3,
+            ),
+            (
+                "test null states with no filter",
+                [
+                    {"state": State.NONE},
+                    {"state": State.NONE},
+                    {"state": State.NONE},
+                    {"state": State.NONE},
+                ],
+                False,
+                ("/api/v1/dags/example_python_operator/dagRuns/" "TEST_DAG_RUN_ID/taskInstances"),
+                4,
             ),
             (
                 "test pool filter",
@@ -503,10 +516,24 @@ class TestGetTaskInstancesBatch(TestTaskInstanceEndpoint):
                     {"state": State.RUNNING},
                     {"state": State.QUEUED},
                     {"state": State.SUCCESS},
+                    {"state": State.NONE},
                 ],
                 False,
-                {"state": ["running", "queued"]},
-                2,
+                {"state": ["running", "queued", "none"]},
+                3,
+                "test_task_read_only",
+            ),
+            (
+                "test dag with null states",
+                [
+                    {"state": State.NONE},
+                    {"state": State.NONE},
+                    {"state": State.NONE},
+                    {"state": State.NONE},
+                ],
+                False,
+                {},
+                4,
                 "test_task_read_only",
             ),
             (
