docs(examples): add DNS rebinding protection to low-level transport examples

The low-level example servers use SseServerTransport and
StreamableHTTPSessionManager directly, bypassing the auto-enable logic
in sse_app()/streamable_http_app() that was added in v1.23.0
(GHSA-9h52-p55h-vw2f).

Per the advisory guidance, low-level transport users should explicitly
configure TransportSecuritySettings. These examples now demonstrate the
correct pattern — the allowlist matches what the high-level API
auto-configures for localhost binds.

Github-Issue: #2269

Author: maxisbey

docs/experimental/tasks-server.md

@@ -418,6 +418,7 @@ from starlette.routing import Mount
 from mcp.server import Server
 from mcp.server.experimental.task_context import ServerTaskContext
 from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
+from mcp.server.transport_security import TransportSecuritySettings
 from mcp.types import (
     CallToolResult, CreateTaskResult, TextContent, Tool, ToolExecution, TASK_REQUIRED,
 )
@@ -463,7 +464,13 @@ async def handle_tool(name: str, arguments: dict) -> CallToolResult | CreateTask
 
 
 def create_app():
-    session_manager = StreamableHTTPSessionManager(app=server)
+    session_manager = StreamableHTTPSessionManager(
+        app=server,
+        security_settings=TransportSecuritySettings(
+            allowed_hosts=["127.0.0.1:*", "localhost:*", "[::1]:*"],
+            allowed_origins=["http://127.0.0.1:*", "http://localhost:*", "http://[::1]:*"],
+        ),
+    )
 
     @asynccontextmanager
     async def lifespan(app: Starlette) -> AsyncIterator[None]:

examples/servers/simple-pagination/mcp_simple_pagination/server.py

@@ -163,11 +163,18 @@ def main(port: int, transport: str) -> int:
 
     if transport == "sse":
         from mcp.server.sse import SseServerTransport
+        from mcp.server.transport_security import TransportSecuritySettings
         from starlette.applications import Starlette
         from starlette.responses import Response
         from starlette.routing import Mount, Route
 
-        sse = SseServerTransport("/messages/")
+        sse = SseServerTransport(
+            "/messages/",
+            security_settings=TransportSecuritySettings(
+                allowed_hosts=["127.0.0.1:*", "localhost:*", "[::1]:*"],
+                allowed_origins=["http://127.0.0.1:*", "http://localhost:*", "http://[::1]:*"],
+            ),
+        )
 
         async def handle_sse(request: Request):
             async with sse.connect_sse(request.scope, request.receive, request._send) as streams:  # type: ignore[reportPrivateUsage]

examples/servers/simple-prompt/mcp_simple_prompt/server.py

@@ -85,11 +85,18 @@ def main(port: int, transport: str) -> int:
 
     if transport == "sse":
         from mcp.server.sse import SseServerTransport
+        from mcp.server.transport_security import TransportSecuritySettings
         from starlette.applications import Starlette
         from starlette.responses import Response
         from starlette.routing import Mount, Route
 
-        sse = SseServerTransport("/messages/")
+        sse = SseServerTransport(
+            "/messages/",
+            security_settings=TransportSecuritySettings(
+                allowed_hosts=["127.0.0.1:*", "localhost:*", "[::1]:*"],
+                allowed_origins=["http://127.0.0.1:*", "http://localhost:*", "http://[::1]:*"],
+            ),
+        )
 
         async def handle_sse(request: Request):
             async with sse.connect_sse(request.scope, request.receive, request._send) as streams:  # type: ignore[reportPrivateUsage]

examples/servers/simple-resource/mcp_simple_resource/server.py

@@ -78,11 +78,18 @@ def main(port: int, transport: str) -> int:
 
     if transport == "sse":
         from mcp.server.sse import SseServerTransport
+        from mcp.server.transport_security import TransportSecuritySettings
         from starlette.applications import Starlette
         from starlette.responses import Response
         from starlette.routing import Mount, Route
 
-        sse = SseServerTransport("/messages/")
+        sse = SseServerTransport(
+            "/messages/",
+            security_settings=TransportSecuritySettings(
+                allowed_hosts=["127.0.0.1:*", "localhost:*", "[::1]:*"],
+                allowed_origins=["http://127.0.0.1:*", "http://localhost:*", "http://[::1]:*"],
+            ),
+        )
 
         async def handle_sse(request: Request):
             async with sse.connect_sse(request.scope, request.receive, request._send) as streams:  # type: ignore[reportPrivateUsage]

examples/servers/simple-streamablehttp-stateless/mcp_simple_streamablehttp_stateless/server.py

@@ -8,6 +8,7 @@
 from mcp import types
 from mcp.server import Server, ServerRequestContext
 from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
+from mcp.server.transport_security import TransportSecuritySettings
 from starlette.applications import Starlette
 from starlette.middleware.cors import CORSMiddleware
 from starlette.routing import Mount
@@ -110,6 +111,10 @@ def main(
         event_store=None,
         json_response=json_response,
         stateless=True,
+        security_settings=TransportSecuritySettings(
+            allowed_hosts=["127.0.0.1:*", "localhost:*", "[::1]:*"],
+            allowed_origins=["http://127.0.0.1:*", "http://localhost:*", "http://[::1]:*"],
+        ),
     )
 
     async def handle_streamable_http(scope: Scope, receive: Receive, send: Send) -> None:

examples/servers/simple-streamablehttp/mcp_simple_streamablehttp/server.py

@@ -7,6 +7,7 @@
 from mcp import types
 from mcp.server import Server, ServerRequestContext
 from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
+from mcp.server.transport_security import TransportSecuritySettings
 from starlette.applications import Starlette
 from starlette.middleware.cors import CORSMiddleware
 from starlette.routing import Mount
@@ -132,6 +133,10 @@ def main(
         app=app,
         event_store=event_store,  # Enable resumability
         json_response=json_response,
+        security_settings=TransportSecuritySettings(
+            allowed_hosts=["127.0.0.1:*", "localhost:*", "[::1]:*"],
+            allowed_origins=["http://127.0.0.1:*", "http://localhost:*", "http://[::1]:*"],
+        ),
     )
 
     # ASGI handler for streamable HTTP connections

examples/servers/simple-task-interactive/mcp_simple_task_interactive/server.py

@@ -16,6 +16,7 @@
 from mcp.server import Server, ServerRequestContext
 from mcp.server.experimental.task_context import ServerTaskContext
 from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
+from mcp.server.transport_security import TransportSecuritySettings
 from starlette.applications import Starlette
 from starlette.routing import Mount
 
@@ -149,7 +150,13 @@ async def app_lifespan(app: Starlette) -> AsyncIterator[None]:
 @click.command()
 @click.option("--port", default=8000, help="Port to listen on")
 def main(port: int) -> int:
-    session_manager = StreamableHTTPSessionManager(app=server)
+    session_manager = StreamableHTTPSessionManager(
+        app=server,
+        security_settings=TransportSecuritySettings(
+            allowed_hosts=["127.0.0.1:*", "localhost:*", "[::1]:*"],
+            allowed_origins=["http://127.0.0.1:*", "http://localhost:*", "http://[::1]:*"],
+        ),
+    )
     starlette_app = create_app(session_manager)
     print(f"Starting server on http://localhost:{port}/mcp")
     uvicorn.run(starlette_app, host="127.0.0.1", port=port)

examples/servers/simple-task/mcp_simple_task/server.py

@@ -10,6 +10,7 @@
 from mcp.server import Server, ServerRequestContext
 from mcp.server.experimental.task_context import ServerTaskContext
 from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
+from mcp.server.transport_security import TransportSecuritySettings
 from starlette.applications import Starlette
 from starlette.routing import Mount
 
@@ -69,7 +70,13 @@ async def work(task: ServerTaskContext) -> types.CallToolResult:
 @click.command()
 @click.option("--port", default=8000, help="Port to listen on")
 def main(port: int) -> int:
-    session_manager = StreamableHTTPSessionManager(app=server)
+    session_manager = StreamableHTTPSessionManager(
+        app=server,
+        security_settings=TransportSecuritySettings(
+            allowed_hosts=["127.0.0.1:*", "localhost:*", "[::1]:*"],
+            allowed_origins=["http://127.0.0.1:*", "http://localhost:*", "http://[::1]:*"],
+        ),
+    )
 
     @asynccontextmanager
     async def app_lifespan(app: Starlette) -> AsyncIterator[None]:

examples/servers/simple-tool/mcp_simple_tool/server.py

@@ -67,11 +67,18 @@ def main(port: int, transport: str) -> int:
 
     if transport == "sse":
         from mcp.server.sse import SseServerTransport
+        from mcp.server.transport_security import TransportSecuritySettings
         from starlette.applications import Starlette
         from starlette.responses import Response
         from starlette.routing import Mount, Route
 
-        sse = SseServerTransport("/messages/")
+        sse = SseServerTransport(
+            "/messages/",
+            security_settings=TransportSecuritySettings(
+                allowed_hosts=["127.0.0.1:*", "localhost:*", "[::1]:*"],
+                allowed_origins=["http://127.0.0.1:*", "http://localhost:*", "http://[::1]:*"],
+            ),
+        )
 
         async def handle_sse(request: Request):
             async with sse.connect_sse(request.scope, request.receive, request._send) as streams:  # type: ignore[reportPrivateUsage]

examples/servers/sse-polling-demo/mcp_sse_polling_demo/server.py

@@ -21,6 +21,7 @@
 from mcp import types
 from mcp.server import Server, ServerRequestContext
 from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
+from mcp.server.transport_security import TransportSecuritySettings
 from starlette.applications import Starlette
 from starlette.routing import Mount
 from starlette.types import Receive, Scope, Send
@@ -157,6 +158,10 @@ def main(port: int, log_level: str, retry_interval: int) -> int:
         app=app,
         event_store=event_store,
         retry_interval=retry_interval,
+        security_settings=TransportSecuritySettings(
+            allowed_hosts=["127.0.0.1:*", "localhost:*", "[::1]:*"],
+            allowed_origins=["http://127.0.0.1:*", "http://localhost:*", "http://[::1]:*"],
+        ),
     )
 
     async def handle_streamable_http(scope: Scope, receive: Receive, send: Send) -> None: