feat: emit DeprecationWarning from all SSE transport entry points

Add deprecation warnings to SseServerTransport, sse_client,
MCPServer.sse_app, MCPServer.run_sse_async, and MCPServer.run(transport="sse")
directing users to migrate to Streamable HTTP transport.

Each warning follows the same pattern:
  "<name> is deprecated. Use <replacement> instead.
   SSE transport will be removed in a future major release."

Tests verify the warnings are emitted from SseServerTransport,
sse_client, and MCPServer.sse_app. A pyproject.toml filterwarnings
entry prevents existing SSE tests from failing due to the new warnings.

Github-Issue: #2278

Author: omar-y-abdi

pyproject.toml

@@ -184,6 +184,8 @@ filterwarnings = [
     "ignore:Returning str or bytes.*:DeprecationWarning:mcp.server.lowlevel",
     # pywin32 internal deprecation warning
     "ignore:getargs.*The 'u' format is deprecated:DeprecationWarning",
+    # SSE transport deprecation (callers should migrate to Streamable HTTP)
+    "ignore:.*is deprecated.*Use.*instead.*SSE transport:DeprecationWarning",
 ]
 
 [tool.markdown.lint]

src/mcp/client/sse.py

@@ -1,4 +1,5 @@
 import logging
+import warnings
 from collections.abc import Callable
 from contextlib import asynccontextmanager
 from typing import Any
@@ -39,6 +40,10 @@ async def sse_client(
 ):
     """Client transport for SSE.
 
+    .. deprecated::
+        The SSE transport is deprecated. Use :func:`~mcp.client.streamable_http.streamable_http_client`
+        (Streamable HTTP) instead. SSE will be removed in a future major release.
+
     `sse_read_timeout` determines how long (in seconds) the client will wait for a new
     event before disconnecting. All other HTTP operations are controlled by `timeout`.
 
@@ -51,6 +56,12 @@ async def sse_client(
         auth: Optional HTTPX authentication handler.
         on_session_created: Optional callback invoked with the session ID when received.
     """
+    warnings.warn(
+        "sse_client is deprecated. Use streamable_http_client instead. "
+        "SSE transport will be removed in a future major release.",
+        DeprecationWarning,
+        stacklevel=2,
+    )
     read_stream: MemoryObjectReceiveStream[SessionMessage | Exception]
     read_stream_writer: MemoryObjectSendStream[SessionMessage | Exception]
 

src/mcp/server/mcpserver/server.py

@@ -6,6 +6,7 @@
 import inspect
 import json
 import re
+import warnings
 from collections.abc import AsyncIterator, Awaitable, Callable, Iterable, Sequence
 from contextlib import AbstractAsyncContextManager, asynccontextmanager
 from typing import Any, Generic, Literal, TypeVar, overload
@@ -286,6 +287,12 @@ def run(
             case "stdio":
                 anyio.run(self.run_stdio_async)
             case "sse":  # pragma: no cover
+                warnings.warn(
+                    'run(transport="sse") is deprecated. Use run(transport="streamable-http") instead. '
+                    "SSE transport will be removed in a future major release.",
+                    DeprecationWarning,
+                    stacklevel=2,
+                )
                 anyio.run(lambda: self.run_sse_async(**kwargs))
             case "streamable-http":  # pragma: no cover
                 anyio.run(lambda: self.run_streamable_http_async(**kwargs))
@@ -854,7 +861,18 @@ async def run_sse_async(  # pragma: no cover
         message_path: str = "/messages/",
         transport_security: TransportSecuritySettings | None = None,
     ) -> None:
-        """Run the server using SSE transport."""
+        """Run the server using SSE transport.
+
+        .. deprecated::
+            Use :meth:`run_streamable_http_async` instead. SSE transport will be
+            removed in a future major release.
+        """
+        warnings.warn(
+            "run_sse_async is deprecated. Use run_streamable_http_async instead. "
+            "SSE transport will be removed in a future major release.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
         import uvicorn
 
         starlette_app = self.sse_app(
@@ -915,7 +933,18 @@ def sse_app(
         transport_security: TransportSecuritySettings | None = None,
         host: str = "127.0.0.1",
     ) -> Starlette:
-        """Return an instance of the SSE server app."""
+        """Return an instance of the SSE server app.
+
+        .. deprecated::
+            Use :meth:`streamable_http_app` instead. SSE transport will be
+            removed in a future major release.
+        """
+        warnings.warn(
+            "sse_app is deprecated. Use streamable_http_app instead. "
+            "SSE transport will be removed in a future major release.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
         # Auto-enable DNS rebinding protection for localhost (IPv4 and IPv6)
         if transport_security is None and host in ("127.0.0.1", "localhost", "::1"):
             transport_security = TransportSecuritySettings(

src/mcp/server/sse.py

@@ -37,6 +37,7 @@ async def handle_sse(request):
 """
 
 import logging
+import warnings
 from contextlib import asynccontextmanager
 from typing import Any
 from urllib.parse import quote
@@ -61,8 +62,15 @@ async def handle_sse(request):
 
 
 class SseServerTransport:
-    """SSE server transport for MCP. This class provides two ASGI applications,
-    suitable for use with a framework like Starlette and a server like Hypercorn:
+    """SSE server transport for MCP.
+
+    .. deprecated::
+        The SSE transport is deprecated. Use
+        :class:`~mcp.server.streamable_http.StreamableHTTPServerTransport` instead.
+        SSE will be removed in a future major release.
+
+    This class provides two ASGI applications, suitable for use with a framework
+    like Starlette and a server like Hypercorn:
 
         1. connect_sse() is an ASGI application which receives incoming GET requests,
            and sets up a new SSE stream to send server messages to the client.
@@ -97,6 +105,13 @@ def __init__(self, endpoint: str, security_settings: TransportSecuritySettings |
             ValueError: If the endpoint is a full URL instead of a relative path
         """
 
+        warnings.warn(
+            "SseServerTransport is deprecated. Use StreamableHTTPServerTransport instead. "
+            "SSE transport will be removed in a future major release.",
+            DeprecationWarning,
+            stacklevel=2,
+        )
+
         super().__init__()
 
         # Validate that endpoint is a relative path and not a full URL

tests/server/mcpserver/test_server.py

@@ -1,4 +1,5 @@
 import base64
+import warnings
 from pathlib import Path
 from typing import Any
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -82,6 +83,16 @@ async def test_sse_app_returns_starlette_app(self):
         assert sse_routes[0].path == "/sse"
         assert mount_routes[0].path == "/messages"
 
+    async def test_sse_app_emits_deprecation_warning(self):
+        """Test that sse_app emits a DeprecationWarning."""
+        mcp = MCPServer("test")
+        with warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter("always")
+            mcp.sse_app(host="0.0.0.0")
+
+        deprecations = [w for w in caught if issubclass(w.category, DeprecationWarning)]
+        assert any("sse_app is deprecated" in str(w.message) for w in deprecations)
+
     async def test_non_ascii_description(self):
         """Test that MCPServer handles non-ASCII characters in descriptions correctly"""
         mcp = MCPServer()

tests/shared/test_sse.py

@@ -1,6 +1,7 @@
 import json
 import multiprocessing
 import socket
+import warnings
 from collections.abc import AsyncGenerator, Generator
 from typing import Any
 from unittest.mock import AsyncMock, MagicMock, Mock, patch
@@ -497,6 +498,34 @@ async def test_request_context_isolation(context_server: None, server_url: str)
         assert ctx["headers"].get("x-custom-value") == f"value-{i}"
 
 
+# --- Deprecation warning tests ---
+
+
+def test_sse_server_transport_emits_deprecation_warning():
+    """SseServerTransport.__init__ should emit a DeprecationWarning."""
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
+        SseServerTransport("/messages/")
+
+    deprecations = [w for w in caught if issubclass(w.category, DeprecationWarning)]
+    assert len(deprecations) == 1
+    assert "SseServerTransport is deprecated" in str(deprecations[0].message)
+    assert "StreamableHTTPServerTransport" in str(deprecations[0].message)
+
+
+@pytest.mark.anyio
+async def test_sse_client_emits_deprecation_warning(server: None, server_url: str):
+    """sse_client should emit a DeprecationWarning."""
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter("always")
+        async with sse_client(server_url + "/sse") as streams:
+            async with ClientSession(*streams) as session:  # pragma: no branch
+                await session.initialize()
+
+    deprecations = [w for w in caught if issubclass(w.category, DeprecationWarning)]
+    assert any("sse_client is deprecated" in str(w.message) for w in deprecations)
+
+
 def test_sse_message_id_coercion():
     """Previously, the `RequestId` would coerce a string that looked like an integer into an integer.