Skip to content

Commit 38a5480

Browse files
SergeyMenshykhCopilot
SergeyMenshykh
and
Copilot
authored
Python: Default MCP SSE server samples to loopback with host validation (#14127)
## Motivation and Context The Python MCP server demos under `python/samples/demos/mcp_server/` can optionally run over the SSE transport (`--transport sse`). This updates that sample wiring to follow the Model Context Protocol guidance for local development servers, so developers who use these demos as a starting point inherit sensible defaults. ## Description - **Loopback by default**: the SSE samples now bind to `127.0.0.1` instead of `0.0.0.0`. A new `--host` argument makes binding to other interfaces an explicit opt-in that logs a warning. - **Host/Origin validation**: added Starlette `TrustedHostMiddleware` plus a small Origin allowlist middleware so the local listener only serves loopback callers (requests without an `Origin` header are still allowed, for non-browser MCP clients). - **Sample hygiene**: switched `Starlette(debug=True)` to `debug=False` so the demos don't ship verbose debug output. - **Docs**: the README now describes the loopback-by-default behavior, the `--host` opt-in, and points to the existing `mcp_with_oauth` sample for authenticated, network-reachable deployments. Applies to `sk_mcp_server.py` and `agent_as_server.py`. The stdio transport (the default) and the other stdio-only samples are unchanged. ## Contribution Checklist - [x] The code builds clean without any errors or warnings - [x] I didn't break anyone 😄 --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
1 parent efa3268 commit 38a5480

3 files changed

Lines changed: 189 additions & 10 deletions

File tree

‎python/samples/demos/mcp_server/README.md‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,20 @@ uv --directory=<path to sk project>/semantic-kernel/python/samples/demos/mcp_ser
6161

6262
This will start a server that listens for incoming requests on port `8000`.
6363

64+
> [!NOTE]
65+
> By default the SSE server binds to `127.0.0.1` (loopback) and only accepts requests
66+
> with a loopback `Host` header and, when present, a loopback `Origin` header. A local
67+
> MCP server exposes tools, plugins and model providers backed by your own credentials,
68+
> so it is good practice to keep it reachable only from your own machine. The
69+
> [MCP specification](https://modelcontextprotocol.io/) recommends validating `Origin`
70+
> and binding to loopback, in part to guard against [DNS rebinding](https://en.wikipedia.org/wiki/DNS_rebinding).
71+
>
72+
> You can override the bind address with `--host`, e.g. `--host 0.0.0.0` to expose the
73+
> server on the network. Do this only on a trusted network. The bundled Host/Origin
74+
> checks only allow loopback callers, so a non-loopback deployment needs proper
75+
> authentication - see the [`mcp_with_oauth`](../mcp_with_oauth/) sample for the
76+
> authenticated, Streamable-HTTP pattern recommended for production.
77+
6478
---
6579

6680
In both cases, `uv` will ensure that `semantic-kernel` is installed with the `mcp` extra in a temporary virtual environment.

‎python/samples/demos/mcp_server/agent_as_server.py‎

Lines changed: 88 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
# ///
66
# Copyright (c) Microsoft. All rights reserved.
77
import argparse
8+
import ipaddress
89
import logging
910
from typing import Annotated, Any, Literal
1011

@@ -51,6 +52,16 @@
5152
"""
5253

5354

55+
def is_loopback_host(host: str) -> bool:
56+
"""Return True if the host refers to a loopback interface (incl. IPv6 ::1)."""
57+
if host == "localhost":
58+
return True
59+
try:
60+
return ipaddress.ip_address(host).is_loopback
61+
except ValueError:
62+
return False
63+
64+
5465
def parse_arguments():
5566
parser = argparse.ArgumentParser(description="Run the Semantic Kernel MCP server.")
5667
parser.add_argument(
@@ -66,7 +77,20 @@ def parse_arguments():
6677
default=None,
6778
help="Port to use for SSE transport (required if transport is 'sse').",
6879
)
69-
return parser.parse_args()
80+
parser.add_argument(
81+
"--host",
82+
type=str,
83+
default="127.0.0.1",
84+
help=(
85+
"Host/interface to bind the SSE server to (default: 127.0.0.1). "
86+
"Binding to anything other than loopback (e.g. 0.0.0.0) exposes the server "
87+
"to the network and should only be done on a trusted network with authentication added."
88+
),
89+
)
90+
args = parser.parse_args()
91+
if args.transport == "sse" and args.port is None:
92+
parser.error("--port is required when --transport is 'sse'.")
93+
return args
7094

7195

7296
# Define a simple plugin for the sample
@@ -88,7 +112,7 @@ def get_item_price(
88112
return "$9.99"
89113

90114

91-
async def run(transport: Literal["sse", "stdio"] = "stdio", port: int | None = None) -> None:
115+
async def run(transport: Literal["sse", "stdio"] = "stdio", port: int | None = None, host: str = "127.0.0.1") -> None:
92116
async with (
93117
# 1. Login to Azure and create a Azure AI Project Client
94118
AzureCliCredential() as creds,
@@ -110,7 +134,53 @@ async def run(transport: Literal["sse", "stdio"] = "stdio", port: int | None = N
110134
import uvicorn
111135
from mcp.server.sse import SseServerTransport
112136
from starlette.applications import Starlette
137+
from starlette.middleware import Middleware
138+
from starlette.middleware.trustedhost import TrustedHostMiddleware
139+
from starlette.responses import PlainTextResponse
113140
from starlette.routing import Mount, Route
141+
from starlette.types import ASGIApp, Receive, Scope, Send
142+
143+
# A local MCP server is a security boundary, not a generic web server: it exposes
144+
# tools, plugins and model providers backed by the developer's credentials. Without
145+
# Host/Origin validation a malicious web page could use DNS rebinding to reach this
146+
# loopback listener from the victim's browser and invoke the exposed MCP tools.
147+
# The MCP spec therefore requires servers to validate Origin and bind to loopback.
148+
allowed_hosts = [
149+
"localhost",
150+
"127.0.0.1",
151+
"[::1]",
152+
f"localhost:{port}",
153+
f"127.0.0.1:{port}",
154+
f"[::1]:{port}",
155+
]
156+
allowed_origins = {
157+
"http://localhost",
158+
"http://127.0.0.1",
159+
"http://[::1]",
160+
f"http://localhost:{port}",
161+
f"http://127.0.0.1:{port}",
162+
f"http://[::1]:{port}",
163+
}
164+
165+
class OriginValidationMiddleware:
166+
"""Reject requests with an untrusted Origin header (DNS-rebinding defense)."""
167+
168+
def __init__(self, app: ASGIApp) -> None:
169+
self.app = app
170+
171+
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
172+
if scope["type"] == "http":
173+
origin = dict(scope["headers"]).get(b"origin")
174+
if origin is not None:
175+
try:
176+
origin_value = origin.decode("ascii")
177+
except UnicodeDecodeError:
178+
origin_value = None
179+
if origin_value not in allowed_origins:
180+
response = PlainTextResponse("Forbidden: invalid Origin header", status_code=403)
181+
await response(scope, receive, send)
182+
return
183+
await self.app(scope, receive, send)
114184

115185
sse = SseServerTransport("/messages/")
116186

@@ -122,14 +192,27 @@ async def handle_sse(request):
122192
await server.run(read_stream, write_stream, server.create_initialization_options())
123193

124194
starlette_app = Starlette(
125-
debug=True,
195+
debug=False,
126196
routes=[
127197
Route("/sse", endpoint=handle_sse),
128198
Mount("/messages/", app=sse.handle_post_message),
129199
],
200+
middleware=[
201+
Middleware(TrustedHostMiddleware, allowed_hosts=allowed_hosts),
202+
Middleware(OriginValidationMiddleware),
203+
],
130204
)
205+
206+
if not is_loopback_host(host):
207+
logger.warning(
208+
"Binding the MCP SSE server to %s exposes it beyond loopback. The bundled Host/Origin "
209+
"checks only allow loopback callers; for a network-reachable or credentialed deployment "
210+
"add proper authentication (see the mcp_with_oauth sample) before doing this.",
211+
host,
212+
)
213+
131214
nest_asyncio.apply()
132-
uvicorn.run(starlette_app, host="0.0.0.0", port=port) # nosec
215+
uvicorn.run(starlette_app, host=host, port=port) # nosec
133216
elif transport == "stdio":
134217
from mcp.server.stdio import stdio_server
135218

@@ -142,4 +225,4 @@ async def handle_stdin(stdin: Any | None = None, stdout: Any | None = None) -> N
142225

143226
if __name__ == "__main__":
144227
args = parse_arguments()
145-
anyio.run(run, args.transport, args.port)
228+
anyio.run(run, args.transport, args.port, args.host)

‎python/samples/demos/mcp_server/sk_mcp_server.py‎

Lines changed: 87 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@
55
# ///
66
# Copyright (c) Microsoft. All rights reserved.
77
import argparse
8+
import ipaddress
89
import logging
910
from typing import Any, Literal
1011

@@ -54,6 +55,16 @@
5455
"""
5556

5657

58+
def is_loopback_host(host: str) -> bool:
59+
"""Return True if the host refers to a loopback interface (incl. IPv6 ::1)."""
60+
if host == "localhost":
61+
return True
62+
try:
63+
return ipaddress.ip_address(host).is_loopback
64+
except ValueError:
65+
return False
66+
67+
5768
def parse_arguments():
5869
parser = argparse.ArgumentParser(description="Run the Semantic Kernel MCP server.")
5970
parser.add_argument(
@@ -69,10 +80,23 @@ def parse_arguments():
6980
default=None,
7081
help="Port to use for SSE transport (required if transport is 'sse').",
7182
)
72-
return parser.parse_args()
83+
parser.add_argument(
84+
"--host",
85+
type=str,
86+
default="127.0.0.1",
87+
help=(
88+
"Host/interface to bind the SSE server to (default: 127.0.0.1). "
89+
"Binding to anything other than loopback (e.g. 0.0.0.0) exposes the server "
90+
"to the network and should only be done on a trusted network with authentication added."
91+
),
92+
)
93+
args = parser.parse_args()
94+
if args.transport == "sse" and args.port is None:
95+
parser.error("--port is required when --transport is 'sse'.")
96+
return args
7397

7498

75-
def run(transport: Literal["sse", "stdio"] = "stdio", port: int | None = None) -> None:
99+
def run(transport: Literal["sse", "stdio"] = "stdio", port: int | None = None, host: str = "127.0.0.1") -> None:
76100
kernel = Kernel()
77101

78102
@kernel_function()
@@ -112,7 +136,53 @@ def echo_function(message: str, extra: str = "") -> str:
112136
import uvicorn
113137
from mcp.server.sse import SseServerTransport
114138
from starlette.applications import Starlette
139+
from starlette.middleware import Middleware
140+
from starlette.middleware.trustedhost import TrustedHostMiddleware
141+
from starlette.responses import PlainTextResponse
115142
from starlette.routing import Mount, Route
143+
from starlette.types import ASGIApp, Receive, Scope, Send
144+
145+
# A local MCP server is a security boundary, not a generic web server: it exposes
146+
# tools, plugins and model providers backed by the developer's credentials. Without
147+
# Host/Origin validation a malicious web page could use DNS rebinding to reach this
148+
# loopback listener from the victim's browser and invoke the exposed MCP tools.
149+
# The MCP spec therefore requires servers to validate Origin and bind to loopback.
150+
allowed_hosts = [
151+
"localhost",
152+
"127.0.0.1",
153+
"[::1]",
154+
f"localhost:{port}",
155+
f"127.0.0.1:{port}",
156+
f"[::1]:{port}",
157+
]
158+
allowed_origins = {
159+
"http://localhost",
160+
"http://127.0.0.1",
161+
"http://[::1]",
162+
f"http://localhost:{port}",
163+
f"http://127.0.0.1:{port}",
164+
f"http://[::1]:{port}",
165+
}
166+
167+
class OriginValidationMiddleware:
168+
"""Reject requests with an untrusted Origin header (DNS-rebinding defense)."""
169+
170+
def __init__(self, app: ASGIApp) -> None:
171+
self.app = app
172+
173+
async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
174+
if scope["type"] == "http":
175+
origin = dict(scope["headers"]).get(b"origin")
176+
if origin is not None:
177+
try:
178+
origin_value = origin.decode("ascii")
179+
except UnicodeDecodeError:
180+
origin_value = None
181+
if origin_value not in allowed_origins:
182+
response = PlainTextResponse("Forbidden: invalid Origin header", status_code=403)
183+
await response(scope, receive, send)
184+
return
185+
await self.app(scope, receive, send)
116186

117187
sse = SseServerTransport("/messages/")
118188

@@ -121,14 +191,26 @@ async def handle_sse(request):
121191
await server.run(read_stream, write_stream, server.create_initialization_options())
122192

123193
starlette_app = Starlette(
124-
debug=True,
194+
debug=False,
125195
routes=[
126196
Route("/sse", endpoint=handle_sse),
127197
Mount("/messages/", app=sse.handle_post_message),
128198
],
199+
middleware=[
200+
Middleware(TrustedHostMiddleware, allowed_hosts=allowed_hosts),
201+
Middleware(OriginValidationMiddleware),
202+
],
129203
)
130204

131-
uvicorn.run(starlette_app, host="0.0.0.0", port=port) # nosec
205+
if not is_loopback_host(host):
206+
logger.warning(
207+
"Binding the MCP SSE server to %s exposes it beyond loopback. The bundled Host/Origin "
208+
"checks only allow loopback callers; for a network-reachable or credentialed deployment "
209+
"add proper authentication (see the mcp_with_oauth sample) before doing this.",
210+
host,
211+
)
212+
213+
uvicorn.run(starlette_app, host=host, port=port) # nosec
132214
elif transport == "stdio":
133215
import anyio
134216
from mcp.server.stdio import stdio_server
@@ -142,4 +224,4 @@ async def handle_stdin(stdin: Any | None = None, stdout: Any | None = None) -> N
142224

143225
if __name__ == "__main__":
144226
args = parse_arguments()
145-
run(transport=args.transport, port=args.port)
227+
run(transport=args.transport, port=args.port, host=args.host)

0 commit comments

Comments
 (0)