Skip multipart parts with empty Content-Type instead of raising

gql/transport/aiohttp.py

@@ -542,6 +542,9 @@ async def _parse_multipart_part(
         """
         # Verify the part has the correct content type
         content_type = part.headers.get(aiohttp.hdrs.CONTENT_TYPE, "")
+        if not content_type:
+            log.debug("Skipping part with no content-type (likely heartbeat)")
+            return None
         if not content_type.startswith("application/json"):
             raise TransportProtocolError(
                 f"Unexpected part content-type: {content_type}. "

tests/test_aiohttp_multipart.py

@@ -480,6 +480,38 @@ async def test_aiohttp_multipart_wrong_part_content_type(multipart_server):
         assert "text/html" in str(exc_info.value)
 
 
+@pytest.mark.asyncio
+async def test_aiohttp_multipart_empty_content_type_skipped(multipart_server):
+    """Test that parts with empty/missing content-type are skipped as heartbeats."""
+    from gql.transport.aiohttp import AIOHTTPTransport
+
+    book1_payload = json.dumps({"payload": {"data": {"book": book1}}})
+
+    parts = [
+        ("--graphql\r\n" "\r\n" "\r\n"),
+        (
+            "--graphql\r\n"
+            "Content-Type: application/json\r\n"
+            "\r\n"
+            f"{book1_payload}\r\n"
+        ),
+        "--graphql--\r\n",
+    ]
+
+    server = await multipart_server(parts)
+    url = server.make_url("/")
+    transport = AIOHTTPTransport(url=url)
+
+    async with Client(transport=transport) as session:
+        query = gql(subscription_str)
+        results = []
+        async for result in session.subscribe(query):
+            results.append(result)
+
+        assert len(results) == 1
+        assert results[0]["book"]["title"] == "Book 1"
+
+
 @pytest.mark.asyncio
 async def test_aiohttp_multipart_response_headers(multipart_server):
     """Test that response headers are captured in the transport."""