From 2a683aac2eab905dabaae9dbbe2f2269e3fa3415 Mon Sep 17 00:00:00 2001 From: Safi Date: Sun, 7 Jun 2026 00:43:40 +0100 Subject: [PATCH] feat(serve): add Streamable HTTP transport to MCP server (#1143) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit python -m graphify.serve graph.json --transport http --port 8080 serves the same MCP tools over the Streamable HTTP transport (spec 2025-03-26) so a single shared process can serve the graph for a whole team. - _build_server() refactors server registration into a shared factory (stdio behavior is byte-for-byte unchanged — all 52 existing tests pass) - _ApiKeyMiddleware: raw ASGI (not BaseHTTPMiddleware) preserves SSE streaming; constant-time compare; RFC-6750 case-insensitive Bearer; blank-key normalized to no-auth - DNS-rebinding protection via TransportSecuritySettings; wildcard binds disable it and print an exposure warning when no api-key is set - session_idle_timeout reaps idle stateful sessions (default 3600s) so a long-running shared server does not leak memory on client disconnect - Dockerfile + .dockerignore for containerized team deployment - 16 new tests via in-process ASGI test client (importorskip-guarded) - stdio remains the default; no change for existing setups Co-Authored-By: Claude Sonnet 4.6 --- .dockerignore | 25 ++++ Dockerfile | 25 ++++ graphify/serve.py | 282 ++++++++++++++++++++++++++++++++++++++- tests/test_serve_http.py | 232 ++++++++++++++++++++++++++++++++ 4 files changed, 559 insertions(+), 5 deletions(-) create mode 100644 .dockerignore create mode 100644 Dockerfile create mode 100644 tests/test_serve_http.py diff --git a/.dockerignore b/.dockerignore new file mode 100644 index 00000000..44aa4943 --- /dev/null +++ b/.dockerignore @@ -0,0 +1,25 @@ +.git +.github +.venv +venv +__pycache__ +*.pyc +.pytest_cache +.ruff_cache +.mypy_cache +dist +build +*.egg-info +graphify-out +graphify-benchmark +graphify_eval +graphify_test +worked +llm-stack-corpus +llm-stack-demo +product-site +ebook +tests +docs +*.md +!README.md diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 00000000..a313833c --- /dev/null +++ b/Dockerfile @@ -0,0 +1,25 @@ +# graphify MCP server as a shared HTTP service (issue #1143). +# +# Build: docker build -t graphify . +# Run: docker run -p 8080:8080 -v "$(pwd)/graphify-out:/data" graphify \ +# /data/graph.json --transport http --host 0.0.0.0 --api-key "$SECRET" +# +# Builds from source so the image includes the Streamable HTTP transport even +# before it lands on PyPI. The graph.json is mounted at runtime (-v), never +# baked into the image. +FROM python:3.12-slim + +WORKDIR /app +COPY . /app + +# The [mcp] extra pulls mcp + starlette + uvicorn, which the HTTP transport needs. +RUN pip install --no-cache-dir ".[mcp]" + +# Run as a non-root user — the server is network-exposed. +RUN useradd --create-home --uid 10001 graphify +USER graphify + +EXPOSE 8080 + +ENTRYPOINT ["python", "-m", "graphify.serve"] +CMD ["/data/graph.json", "--transport", "http", "--host", "0.0.0.0", "--port", "8080"] diff --git a/graphify/serve.py b/graphify/serve.py index 23148204..4c55c675 100644 --- a/graphify/serve.py +++ b/graphify/serve.py @@ -472,13 +472,18 @@ def _filter_blank_stdin() -> None: sys.stdin = open(0, "r", closefd=False) -def serve(graph_path: str = "graphify-out/graph.json") -> None: - """Start the MCP server. Requires pip install mcp.""" +def _build_server(graph_path: str): + """Build the configured low-level MCP Server (shared by every transport). + + All graph query tools and resources are registered here over a single + ``mcp.server.Server`` instance; the caller picks the transport (stdio or + Streamable HTTP) and runs it. Hot-reload of graph.json works the same way + regardless of transport, since reloads happen inside the tool handlers. + """ import threading try: from mcp.server import Server - from mcp.server.stdio import stdio_server from mcp import types from mcp.types import AnyUrl except ImportError as e: @@ -992,8 +997,19 @@ def serve(graph_path: str = "graphify-out/graph.json") -> None: except Exception as exc: return [types.TextContent(type="text", text=f"Error executing {name}: {exc}")] + return server + + +def serve(graph_path: str = "graphify-out/graph.json") -> None: + """Start the MCP server over stdio (the default, per-developer transport).""" + try: + from mcp.server.stdio import stdio_server + except ImportError as e: + raise ImportError('mcp not installed. Run: pip install "graphifyy[mcp]"') from e import asyncio + server = _build_server(graph_path) + async def main() -> None: async with stdio_server() as streams: await server.run(streams[0], streams[1], server.create_initialization_options()) @@ -1002,6 +1018,262 @@ def serve(graph_path: str = "graphify-out/graph.json") -> None: asyncio.run(main()) +class _MCPASGIApp: + """Raw-ASGI wrapper around the Streamable HTTP session manager. + + Passed to a Starlette ``Route`` as a class instance (not a function) so + Starlette treats it as an ASGI app: it serves the exact mount path for all + methods (GET/POST/DELETE) with no request/response wrapping and no + trailing-slash redirect — mirroring how FastMCP mounts the same manager. + """ + + def __init__(self, manager) -> None: + self._manager = manager + + async def __call__(self, scope, receive, send) -> None: + await self._manager.handle_request(scope, receive, send) + + +class _ApiKeyMiddleware: + """Pure-ASGI API-key gate for the HTTP transport. + + Implemented as raw ASGI (not Starlette's BaseHTTPMiddleware) on purpose: + BaseHTTPMiddleware buffers responses and breaks the Streamable HTTP SSE + stream. This short-circuits with 401 before the request ever reaches the + session manager, leaving the streaming path untouched for authorized calls. + """ + + def __init__(self, app, api_key: str) -> None: + self.app = app + self._expected = api_key.encode("utf-8") + + async def __call__(self, scope, receive, send) -> None: + if scope["type"] != "http": + await self.app(scope, receive, send) + return + import hmac + headers = dict(scope.get("headers") or []) + provided = headers.get(b"x-api-key") + if provided is None: + # RFC 6750: the auth scheme token is case-insensitive. + scheme, _, token = headers.get(b"authorization", b"").partition(b" ") + if scheme.lower() == b"bearer" and token: + provided = token.strip() + # Constant-time compare; reject when no key was supplied at all. + if provided is None or not hmac.compare_digest(provided, self._expected): + body = b'{"error": "unauthorized"}' + await send({ + "type": "http.response.start", + "status": 401, + "headers": [ + (b"content-type", b"application/json"), + (b"content-length", str(len(body)).encode("ascii")), + ], + }) + await send({"type": "http.response.body", "body": body}) + return + await self.app(scope, receive, send) + + +def _build_http_app( + graph_path: str, + *, + host: str = "127.0.0.1", + port: int = 8080, + api_key: str | None = None, + path: str = "/mcp", + json_response: bool = False, + stateless: bool = False, + session_timeout: float | None = 3600.0, +): + """Build the Starlette ASGI app for the Streamable HTTP transport. + + Split out from :func:`serve_http` (which blocks on uvicorn) so the wiring + can be exercised with an in-process ASGI test client. + + ``session_timeout`` reaps stateful sessions idle for that many seconds so a + long-running shared server does not leak memory when IDE clients disconnect + without sending a DELETE. ``None`` (or <= 0) disables reaping; it is forced + to ``None`` in stateless mode, which has no sessions to reap. + """ + try: + import contextlib + + from starlette.applications import Starlette + from starlette.middleware import Middleware + from starlette.routing import Route + + from mcp.server.streamable_http_manager import StreamableHTTPSessionManager + from mcp.server.transport_security import TransportSecuritySettings + except ImportError as e: + raise ImportError( + 'HTTP transport needs the mcp extra (mcp + starlette + uvicorn). ' + 'Run: pip install "graphifyy[mcp]"' + ) from e + + # A blank key (e.g. --api-key "" or an empty GRAPHIFY_API_KEY) must not be + # mistaken for "auth on" — normalize it to None so the gate is unambiguous. + api_key = (api_key or "").strip() or None + + server = _build_server(graph_path) + + # DNS-rebinding protection. When the operator binds a wildcard address they + # are intentionally exposing the server, so accept any Host header; for a + # loopback/specific bind, restrict Host to that address (with and without + # the port) plus the localhost aliases. + if host in ("0.0.0.0", "::", ""): + security = TransportSecuritySettings(enable_dns_rebinding_protection=False) + else: + allowed = {host, "localhost", "127.0.0.1"} + allowed |= {f"{h}:{port}" for h in list(allowed)} + security = TransportSecuritySettings(allowed_hosts=sorted(allowed)) + + # The SDK rejects a non-positive timeout and forbids one in stateless mode. + idle_timeout = None if (stateless or not session_timeout or session_timeout <= 0) else session_timeout + + manager = StreamableHTTPSessionManager( + app=server, + json_response=json_response, + stateless=stateless, + security_settings=security, + session_idle_timeout=idle_timeout, + ) + + @contextlib.asynccontextmanager + async def lifespan(_app): + # The session manager owns an anyio task group that must wrap the whole + # server lifetime, so enter it here rather than per-request. + async with manager.run(): + yield + + middleware = [] + if api_key: + middleware.append(Middleware(_ApiKeyMiddleware, api_key=api_key)) + + return Starlette( + routes=[Route(path, endpoint=_MCPASGIApp(manager))], + middleware=middleware, + lifespan=lifespan, + ) + + +def serve_http( + graph_path: str = "graphify-out/graph.json", + *, + host: str = "127.0.0.1", + port: int = 8080, + api_key: str | None = None, + path: str = "/mcp", + json_response: bool = False, + stateless: bool = False, + session_timeout: float | None = 3600.0, +) -> None: + """Start the MCP server over Streamable HTTP (MCP spec 2025-03-26). + + Serves the same tools/resources as the stdio transport, so a single shared + process can host the graph for a whole team. Clients point their IDE MCP + config at ``http://:`` (default ``/mcp``). + + ``api_key`` (or the ``GRAPHIFY_API_KEY`` env var) enables a simple header + check (``Authorization: Bearer `` or ``X-API-Key: ``). OAuth is a + deliberate follow-up. Binding ``0.0.0.0`` exposes the server beyond + localhost — set an api_key when you do. + """ + try: + import uvicorn + except ImportError as e: + raise ImportError( + 'HTTP transport needs the mcp extra (mcp + starlette + uvicorn). ' + 'Run: pip install "graphifyy[mcp]"' + ) from e + + api_key = (api_key or "").strip() or None + + app = _build_http_app( + graph_path, + host=host, + port=port, + api_key=api_key, + path=path, + json_response=json_response, + stateless=stateless, + session_timeout=session_timeout, + ) + + auth_note = "api-key required" if api_key else "no auth (set --api-key to require one)" + print( + f"graphify MCP server (streamable-http) on http://{host}:{port}{path} - {auth_note}", + file=sys.stderr, + ) + if host in ("0.0.0.0", "::", "") and not api_key: + print( + f"WARNING: binding {host or '0.0.0.0'} with no api-key exposes the graph " + "unauthenticated on the network. Set --api-key (or GRAPHIFY_API_KEY).", + file=sys.stderr, + ) + uvicorn.run(app, host=host, port=port) + + +def _main(argv: list[str] | None = None) -> None: + import argparse + import os + + parser = argparse.ArgumentParser( + prog="python -m graphify.serve", + description="Serve a graphify knowledge graph over MCP (stdio or Streamable HTTP).", + ) + parser.add_argument( + "graph_path", + nargs="?", + default="graphify-out/graph.json", + help="Path to graph.json (default: graphify-out/graph.json)", + ) + parser.add_argument( + "--transport", + choices=["stdio", "http"], + default="stdio", + help="Transport to serve on (default: stdio)", + ) + parser.add_argument("--host", default="127.0.0.1", help="HTTP bind host (default: 127.0.0.1)") + parser.add_argument("--port", type=int, default=8080, help="HTTP bind port (default: 8080)") + parser.add_argument( + "--api-key", + default=os.environ.get("GRAPHIFY_API_KEY"), + help="Require this key on the HTTP transport (env: GRAPHIFY_API_KEY)", + ) + parser.add_argument("--path", default="/mcp", help="HTTP mount path (default: /mcp)") + parser.add_argument( + "--json-response", + action="store_true", + help="Return plain JSON responses instead of SSE streams", + ) + parser.add_argument( + "--stateless", + action="store_true", + help="Run without per-session state (for load-balanced / CI deployments)", + ) + parser.add_argument( + "--session-timeout", + type=float, + default=3600.0, + help="Reap stateful sessions idle this many seconds (default: 3600; 0 disables)", + ) + args = parser.parse_args(argv) + + if args.transport == "http": + serve_http( + args.graph_path, + host=args.host, + port=args.port, + api_key=args.api_key, + path=args.path, + json_response=args.json_response, + stateless=args.stateless, + session_timeout=args.session_timeout, + ) + else: + serve(args.graph_path) + + if __name__ == "__main__": - graph_path = sys.argv[1] if len(sys.argv) > 1 else "graphify-out/graph.json" - serve(graph_path) + _main() diff --git a/tests/test_serve_http.py b/tests/test_serve_http.py new file mode 100644 index 00000000..e5820533 --- /dev/null +++ b/tests/test_serve_http.py @@ -0,0 +1,232 @@ +"""Tests for the Streamable HTTP transport on the MCP server (issue #1143). + +These exercise the ASGI wiring in-process (no uvicorn, no real socket) via +Starlette's TestClient, so they stay fast and offline. The stdio path is +unchanged and covered elsewhere. +""" +from __future__ import annotations + +import json +from pathlib import Path + +import pytest + +pytest.importorskip("mcp") +pytest.importorskip("starlette") + +from starlette.testclient import TestClient # noqa: E402 + +from graphify import serve as serve_mod # noqa: E402 + +SAMPLE_GRAPH = { + "directed": True, + "nodes": [ + {"id": "a", "label": "Alpha", "community": 0}, + {"id": "b", "label": "Beta", "community": 0}, + ], + "edges": [ + {"source": "a", "target": "b", "relation": "calls", "confidence": "EXTRACTED"}, + ], +} + +_INIT_BODY = { + "jsonrpc": "2.0", + "id": 1, + "method": "initialize", + "params": { + "protocolVersion": "2025-03-26", + "capabilities": {}, + "clientInfo": {"name": "test", "version": "0"}, + }, +} + +_MCP_HEADERS = { + "Content-Type": "application/json", + "Accept": "application/json, text/event-stream", +} + + +def _graph_file(tmp_path: Path) -> str: + p = tmp_path / "graph.json" + p.write_text(json.dumps(SAMPLE_GRAPH), encoding="utf-8") + return str(p) + + +def _client(app) -> TestClient: + # Default host is 127.0.0.1, so the DNS-rebinding guard only accepts that + # Host header (TestClient otherwise sends the disallowed "testserver"). + return TestClient(app, base_url="http://127.0.0.1") + + +def test_app_builds_and_initialize_succeeds(tmp_path): + app = serve_mod._build_http_app(_graph_file(tmp_path), json_response=True) + with _client(app) as client: + resp = client.post("/mcp", headers=_MCP_HEADERS, json=_INIT_BODY) + assert resp.status_code == 200 + # json_response=True returns a single JSON-RPC envelope. + payload = resp.json() + assert payload["jsonrpc"] == "2.0" + assert payload["result"]["serverInfo"]["name"] == "graphify" + + +def test_unknown_path_is_404(tmp_path): + app = serve_mod._build_http_app(_graph_file(tmp_path), json_response=True) + with _client(app) as client: + resp = client.post("/nope", headers=_MCP_HEADERS, json=_INIT_BODY) + assert resp.status_code == 404 + + +def test_api_key_missing_is_401(tmp_path): + app = serve_mod._build_http_app(_graph_file(tmp_path), api_key="s3cret", json_response=True) + with _client(app) as client: + resp = client.post("/mcp", headers=_MCP_HEADERS, json=_INIT_BODY) + assert resp.status_code == 401 + assert resp.json()["error"] == "unauthorized" + + +def test_api_key_wrong_is_401(tmp_path): + app = serve_mod._build_http_app(_graph_file(tmp_path), api_key="s3cret", json_response=True) + with _client(app) as client: + resp = client.post( + "/mcp", + headers={**_MCP_HEADERS, "Authorization": "Bearer nope"}, + json=_INIT_BODY, + ) + assert resp.status_code == 401 + + +def test_api_key_bearer_ok(tmp_path): + app = serve_mod._build_http_app(_graph_file(tmp_path), api_key="s3cret", json_response=True) + with _client(app) as client: + resp = client.post( + "/mcp", + headers={**_MCP_HEADERS, "Authorization": "Bearer s3cret"}, + json=_INIT_BODY, + ) + assert resp.status_code == 200 + assert resp.json()["result"]["serverInfo"]["name"] == "graphify" + + +def test_api_key_x_api_key_header_ok(tmp_path): + app = serve_mod._build_http_app(_graph_file(tmp_path), api_key="s3cret", json_response=True) + with _client(app) as client: + resp = client.post( + "/mcp", + headers={**_MCP_HEADERS, "X-API-Key": "s3cret"}, + json=_INIT_BODY, + ) + assert resp.status_code == 200 + + +def test_blank_api_key_means_no_auth(tmp_path): + # An empty/whitespace key must normalize to "no auth", not a key of "". + app = serve_mod._build_http_app(_graph_file(tmp_path), api_key=" ", json_response=True) + with _client(app) as client: + resp = client.post("/mcp", headers=_MCP_HEADERS, json=_INIT_BODY) + assert resp.status_code == 200 + + +def test_api_key_bearer_scheme_case_insensitive(tmp_path): + app = serve_mod._build_http_app(_graph_file(tmp_path), api_key="s3cret", json_response=True) + with _client(app) as client: + resp = client.post( + "/mcp", + headers={**_MCP_HEADERS, "Authorization": "bearer s3cret"}, + json=_INIT_BODY, + ) + assert resp.status_code == 200 + + +def test_custom_mount_path(tmp_path): + app = serve_mod._build_http_app(_graph_file(tmp_path), path="/graph", json_response=True) + with _client(app) as client: + ok = client.post("/graph", headers=_MCP_HEADERS, json=_INIT_BODY) + assert ok.status_code == 200 + missing = client.post("/mcp", headers=_MCP_HEADERS, json=_INIT_BODY) + assert missing.status_code == 404 + + +def test_tools_list_over_http(tmp_path): + """A full initialize -> tools/list round trip works over the HTTP transport.""" + app = serve_mod._build_http_app(_graph_file(tmp_path), json_response=True) + with _client(app) as client: + init = client.post("/mcp", headers=_MCP_HEADERS, json=_INIT_BODY) + assert init.status_code == 200 + session_id = init.headers.get("mcp-session-id") + assert session_id, "stateful transport should return a session id" + notify_headers = {**_MCP_HEADERS, "mcp-session-id": session_id} + client.post( + "/mcp", + headers=notify_headers, + json={"jsonrpc": "2.0", "method": "notifications/initialized"}, + ) + resp = client.post( + "/mcp", + headers=notify_headers, + json={"jsonrpc": "2.0", "id": 2, "method": "tools/list", "params": {}}, + ) + assert resp.status_code == 200 + names = {t["name"] for t in resp.json()["result"]["tools"]} + assert {"query_graph", "get_node", "graph_stats"} <= names + + +def test_stateless_mode_initialize(tmp_path): + app = serve_mod._build_http_app(_graph_file(tmp_path), stateless=True, json_response=True) + with _client(app) as client: + resp = client.post("/mcp", headers=_MCP_HEADERS, json=_INIT_BODY) + assert resp.status_code == 200 + + +def test_stateless_with_timeout_does_not_raise(tmp_path): + # session_timeout must be forced to None in stateless mode (the SDK raises + # RuntimeError otherwise). Building + a request should just work. + app = serve_mod._build_http_app( + _graph_file(tmp_path), stateless=True, session_timeout=3600, json_response=True + ) + with _client(app) as client: + assert client.post("/mcp", headers=_MCP_HEADERS, json=_INIT_BODY).status_code == 200 + + +def test_session_timeout_zero_disables(tmp_path): + # 0 / non-positive must disable reaping without tripping the SDK's validation. + app = serve_mod._build_http_app(_graph_file(tmp_path), session_timeout=0, json_response=True) + with _client(app) as client: + assert client.post("/mcp", headers=_MCP_HEADERS, json=_INIT_BODY).status_code == 200 + + +# --- CLI argument parsing ------------------------------------------------- + +def test_cli_defaults_to_stdio(monkeypatch): + calls = {} + monkeypatch.setattr(serve_mod, "serve", lambda gp: calls.setdefault("stdio", gp)) + monkeypatch.setattr( + serve_mod, "serve_http", lambda *a, **k: calls.setdefault("http", (a, k)) + ) + serve_mod._main(["graphify-out/graph.json"]) + assert calls.get("stdio") == "graphify-out/graph.json" + assert "http" not in calls + + +def test_cli_http_passes_flags(monkeypatch): + captured = {} + monkeypatch.setattr(serve_mod, "serve", lambda gp: captured.setdefault("stdio", gp)) + monkeypatch.setattr( + serve_mod, "serve_http", lambda gp, **k: captured.update(gp=gp, **k) + ) + serve_mod._main([ + "g.json", "--transport", "http", "--host", "0.0.0.0", + "--port", "9000", "--api-key", "k", "--stateless", + ]) + assert captured["gp"] == "g.json" + assert captured["host"] == "0.0.0.0" + assert captured["port"] == 9000 + assert captured["api_key"] == "k" + assert captured["stateless"] is True + + +def test_cli_api_key_from_env(monkeypatch): + captured = {} + monkeypatch.setenv("GRAPHIFY_API_KEY", "from-env") + monkeypatch.setattr(serve_mod, "serve_http", lambda gp, **k: captured.update(**k)) + serve_mod._main(["g.json", "--transport", "http"]) + assert captured["api_key"] == "from-env"