fix(streamable_http): allow Content-Type header override

Allow StreamableHTTPTransport and streamable_http_client to accept a
custom content_type parameter instead of hardcoding 'application/json'.
This enables clients to override the Content-Type header to include
custom charsets (e.g. 'application/json; charset=utf-8') or other
attributes required by specific server implementations.

Fixes: #2375

Author: Alberto Farah

src/mcp/client/streamable_http.py

@@ -72,13 +72,17 @@ class RequestContext:
 class StreamableHTTPTransport:
     """StreamableHTTP client transport implementation."""
 
-    def __init__(self, url: str) -> None:
+    def __init__(self, url: str, content_type: str = "application/json") -> None:
         """Initialize the StreamableHTTP transport.
 
         Args:
             url: The endpoint URL.
+            content_type: The Content-Type header value for POST requests.
+                Defaults to "application/json". Can be overridden to include
+                custom charsets (e.g. "application/json; charset=utf-8").
         """
         self.url = url
+        self.content_type = content_type
         self.session_id: str | None = None
         self.protocol_version: str | None = None
 
@@ -90,7 +94,7 @@ def _prepare_headers(self) -> dict[str, str]:
         """
         headers: dict[str, str] = {
             "accept": "application/json, text/event-stream",
-            "content-type": "application/json",
+            "content-type": self.content_type,
         }
         # Add session headers if available
         if self.session_id:
@@ -515,6 +519,7 @@ async def streamable_http_client(
     *,
     http_client: httpx.AsyncClient | None = None,
     terminate_on_close: bool = True,
+    content_type: str = "application/json",
 ) -> AsyncGenerator[TransportStreams, None]:
     """Client transport for StreamableHTTP.
 
@@ -524,6 +529,9 @@ async def streamable_http_client(
             client with recommended MCP timeouts will be created. To configure headers,
             authentication, or other HTTP settings, create an httpx.AsyncClient and pass it here.
         terminate_on_close: If True, send a DELETE request to terminate the session when the context exits.
+        content_type: The Content-Type header value for POST requests.
+            Defaults to "application/json". Can be overridden to include custom charsets
+            (e.g. "application/json; charset=utf-8").
 
     Yields:
         Tuple containing:
@@ -544,7 +552,7 @@ async def streamable_http_client(
         # Create default client with recommended MCP timeouts
         client = create_mcp_http_client()
 
-    transport = StreamableHTTPTransport(url)
+    transport = StreamableHTTPTransport(url, content_type=content_type)
 
     async with anyio.create_task_group() as tg:
         try:

tests/client/test_http_content_type.py

@@ -0,0 +1,35 @@
+"""Tests for Content-Type override in streamable HTTP transport.
+
+Verifies that the content_type parameter allows overriding the Content-Type header
+to include custom charsets or other attributes.
+"""
+
+import pytest
+from mcp.client.streamable_http import StreamableHTTPTransport
+
+
+def test_streamable_http_transport_default_content_type() -> None:
+    """Test that the default Content-Type is 'application/json'."""
+    transport = StreamableHTTPTransport("http://example.com/mcp")
+    headers = transport._prepare_headers()
+    assert headers["content-type"] == "application/json"
+
+
+def test_streamable_http_transport_custom_content_type() -> None:
+    """Test that a custom Content-Type with charset can be specified."""
+    transport = StreamableHTTPTransport(
+        "http://example.com/mcp",
+        content_type="application/json; charset=utf-8",
+    )
+    headers = transport._prepare_headers()
+    assert headers["content-type"] == "application/json; charset=utf-8"
+
+
+def test_streamable_http_transport_content_type_preserved_in_headers() -> None:
+    """Test that Content-Type is correctly placed in prepared headers."""
+    custom_type = "application/json; charset=utf-8; boundary=npm"
+    transport = StreamableHTTPTransport("http://example.com/mcp", content_type=custom_type)
+    headers = transport._prepare_headers()
+    # The accept header should also be present
+    assert "accept" in headers
+    assert headers["content-type"] == custom_type