Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions src/macos_mcp/__main__.py
Original file line number Diff line number Diff line change
Expand Up @@ -734,6 +734,9 @@ def _handle_sigint(signum, frame):
show_banner=False,
middleware=_http_middleware(auth_key=auth_key, ip_allowlist=parsed_allowlist, oauth_validator=oauth_validator),
uvicorn_config=uvicorn_config or None,
# No tools need per-client HTTP session state; avoid retaining
# abandoned SDK transports when clients reconnect without DELETE.
stateless_http=transport == "streamable-http",
)
case _:
raise ValueError(f"Invalid transport: {transport}")
Expand Down
45 changes: 45 additions & 0 deletions tests/test_server_transport.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,45 @@
from click.testing import CliRunner

import macos_mcp.__main__ as server


def _run_serve(mocker, tmp_path, transport: str):
"""Invoke the serve command with FastMCP execution mocked."""
config = tmp_path / "macos-mcp.toml"
config.write_text("")
mocker.patch.object(server, "validate_permissions")
run = mocker.patch.object(server.mcp, "run")

result = CliRunner().invoke(
server.main,
[
"serve",
"--transport",
transport,
"--host",
"127.0.0.1",
"--port",
"18199",
"--config",
str(config),
],
)

assert result.exit_code == 0, result.output
return run.call_args.kwargs


def test_streamable_http_runs_stateless(mocker, tmp_path):
"""Streamable HTTP must use stateless sessions to bound retained resources."""
kwargs = _run_serve(mocker, tmp_path, "streamable-http")

assert kwargs["transport"] == "streamable-http"
assert kwargs["stateless_http"] is True


def test_sse_keeps_stateful_transport_default(mocker, tmp_path):
"""SSE must retain its existing stateful transport behavior."""
kwargs = _run_serve(mocker, tmp_path, "sse")

assert kwargs["transport"] == "sse"
assert kwargs["stateless_http"] is False