Streamable HTTP support

Author: 3dgiordano

README.md

@@ -157,6 +157,23 @@ When using custom CA certificate bundles, you must configure both:
 
 ---
 
+## Transports
+
+Perfecto MCP runs over **stdio** by default. It can also serve **streamable HTTP**, where credentials are
+resolved per request from an `Authorization: Bearer` header and the target cloud from a `Perfecto-Cloud-Name`
+header, so one server can serve several users and clouds.
+
+```bash
+perfecto-mcp --mcp http
+```
+
+Transport resolution precedence: **CLI `--mcp` > `PERFECTO_MCP_TRANSPORT` > stdio**.
+
+See [docs/hosted-http.md](docs/hosted-http.md) for client configuration, auth behavior, health probes and
+environment variables.
+
+---
+
 ## OpenTelemetry
 
 Perfecto MCP reports traces and metrics for MCP tool calls using [OpenTelemetry](https://opentelemetry.io/). This gives you visibility into which tools are used, how long they take, and when errors occur.

build.py

@@ -123,6 +123,9 @@ def run_pyinstaller(name: str, icon: str):
         '--hidden-import=opentelemetry.propagate',
         '--collect-submodules=opentelemetry',
         '--collect-all=grpc',
+        # Streamable HTTP transport: uvicorn resolves its loop/protocol
+        # implementations by name at runtime, so PyInstaller cannot see them.
+        '--collect-submodules=uvicorn',
     ])
 
 

config/auth.py

@@ -0,0 +1,168 @@
+"""Per-request authentication for the streamable HTTP transport."""
+from __future__ import annotations
+
+import os
+from typing import Optional, Protocol, runtime_checkable
+
+from mcp.server.fastmcp import Context, FastMCP
+from starlette.requests import Request
+from starlette.responses import JSONResponse
+from starlette.types import ASGIApp, Receive, Scope, Send
+
+from config.perfecto import PERFECTO_CLOUD_NAME_ENV_NAME
+from config.token import PerfectoToken, PerfectoTokenError
+
+PERFECTO_TOKEN_STATE_ATTR = "token"
+PERFECTO_USER_CONFIG_STATE_ATTR = "user_config"
+PERFECTO_CLOUD_NAME_HEADER = "perfecto-cloud-name"
+
+# Unauthenticated probe paths for orchestrators / load balancers.
+HEALTH_PATHS = frozenset({"/health", "/healthz"})
+
+
+class AuthError(Exception):
+    """Raised when Authorization cannot be parsed into credentials."""
+
+
+@runtime_checkable
+class AuthPort(Protocol):
+    """Resolves the Perfecto security token for the current tool invocation."""
+
+    def get_token(self, ctx: Context) -> Optional[PerfectoToken]:
+        ...
+
+
+class StdioAuthProvider:
+    """Process-lifetime token from env / token file / Docker secrets."""
+
+    def __init__(self, token: Optional[PerfectoToken]):
+        self._token = token
+
+    def get_token(self, ctx: Context) -> Optional[PerfectoToken]:
+        return self._token
+
+
+class HttpAuthProvider:
+    """Per-request token attached by Bearer auth middleware to request.state."""
+
+    def get_token(self, ctx: Context) -> Optional[PerfectoToken]:
+        request = ctx.request_context.request
+        if request is None:
+            return None
+        return getattr(request.state, PERFECTO_TOKEN_STATE_ATTR, None)
+
+
+def resolve_cloud_name(header_value: Optional[str] = None) -> Optional[str]:
+    """
+    Resolve the Perfecto cloud for a request.
+
+    Precedence: ``Perfecto-Cloud-Name`` header > PERFECTO_CLOUD_NAME env var.
+    """
+    candidate = (header_value or "").strip()
+    if candidate:
+        return candidate
+    return os.getenv(PERFECTO_CLOUD_NAME_ENV_NAME, "").strip() or None
+
+
+def parse_authorization_header(value: Optional[str], cloud_name: Optional[str] = None) -> PerfectoToken:
+    """
+    Parse ``Authorization: Bearer <security-token>`` into a PerfectoToken.
+
+    The cloud name is not carried in the credentials; it comes from the
+    ``Perfecto-Cloud-Name`` header or PERFECTO_CLOUD_NAME. Does not call the
+    Perfecto API — parse only.
+    """
+    if not value or not value.strip():
+        raise AuthError("Missing Authorization header")
+
+    scheme, _, credentials = value.strip().partition(" ")
+    if scheme.lower() != "bearer" or not credentials.strip():
+        raise AuthError("Authorization header must use Bearer scheme")
+
+    try:
+        return PerfectoToken.from_bearer_credentials(credentials.strip(), cloud_name)
+    except PerfectoTokenError as exc:
+        raise AuthError("Unparseable Bearer credentials") from exc
+
+
+class BearerAuthMiddleware:
+    """
+    HTTP gate: require a parseable Bearer token on every request.
+
+    Attaches PerfectoToken to ``request.state``; does not validate against Perfecto.
+    A missing cloud name is not rejected here — tools surface it as a
+    configuration error, the same way stdio does.
+    """
+
+    def __init__(self, app: ASGIApp):
+        self.app = app
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+
+        if scope.get("method") == "OPTIONS":
+            await self.app(scope, receive, send)
+            return
+
+        path = scope.get("path", "") or ""
+        if path in HEALTH_PATHS:
+            await self.app(scope, receive, send)
+            return
+
+        request = Request(scope, receive)
+        cloud_name = resolve_cloud_name(request.headers.get(PERFECTO_CLOUD_NAME_HEADER))
+        try:
+            token = parse_authorization_header(
+                request.headers.get("authorization"),
+                cloud_name,
+            )
+        except AuthError:
+            response = JSONResponse(
+                {"error": "Unauthorized"},
+                status_code=401,
+                headers={"WWW-Authenticate": "Bearer"},
+            )
+            await response(scope, receive, send)
+            return
+
+        setattr(request.state, PERFECTO_TOKEN_STATE_ATTR, token)
+        setattr(
+            request.state,
+            PERFECTO_USER_CONFIG_STATE_ATTR,
+            {"token": token, "cloud_name": token.cloud_name},
+        )
+        await self.app(scope, receive, send)
+
+
+def register_health_routes(mcp: FastMCP) -> None:
+    """Register unauthenticated health probes on the FastMCP ASGI app."""
+
+    @mcp.custom_route("/health", methods=["GET"])
+    async def health(_request: Request) -> JSONResponse:
+        return JSONResponse({"status": "ok"})
+
+    @mcp.custom_route("/healthz", methods=["GET"])
+    async def healthz(_request: Request) -> JSONResponse:
+        return JSONResponse({"status": "ok"})
+
+
+def run_streamable_http(mcp: FastMCP) -> None:
+    """Serve FastMCP over streamable HTTP with Bearer auth middleware."""
+    import anyio
+    import uvicorn
+
+    register_health_routes(mcp)
+
+    async def _serve() -> None:
+        app = BearerAuthMiddleware(mcp.streamable_http_app())
+        config = uvicorn.Config(
+            app,
+            host=mcp.settings.host,
+            port=mcp.settings.port,
+            log_level=mcp.settings.log_level.lower(),
+        )
+        await uvicorn.Server(config).serve()
+
+    anyio.run(_serve)

config/context_resolution.py

@@ -0,0 +1,36 @@
+"""Helpers to read the per-session user config carried by the MCP context."""
+from typing import Any
+
+from config.auth import PERFECTO_TOKEN_STATE_ATTR, PERFECTO_USER_CONFIG_STATE_ATTR
+
+
+def get_request_context(ctx: Any) -> Any:
+    return getattr(ctx, "request_context", None)
+
+
+def get_request_state(ctx: Any) -> Any:
+    request_context = get_request_context(ctx)
+    request = getattr(request_context, "request", None)
+    return getattr(request, "state", None)
+
+
+def resolve_ctx_user_config(ctx: Any) -> dict[str, Any]:
+    request_context = get_request_context(ctx)
+    request_state = get_request_state(ctx)
+
+    request_context_config = getattr(request_context, PERFECTO_USER_CONFIG_STATE_ATTR, None)
+    if isinstance(request_context_config, dict):
+        return request_context_config
+
+    request_state_config = getattr(request_state, PERFECTO_USER_CONFIG_STATE_ATTR, None)
+    if isinstance(request_state_config, dict):
+        return request_state_config
+
+    return {}
+
+
+def resolve_ctx_token(ctx: Any) -> Any:
+    user_config = resolve_ctx_user_config(ctx)
+    request_state = get_request_state(ctx)
+    request_state_token = getattr(request_state, PERFECTO_TOKEN_STATE_ATTR, None)
+    return user_config.get("token") or request_state_token

config/perfecto.py

@@ -8,8 +8,8 @@
 SECURITY_TOKEN_ENV_NAME: str = "PERFECTO_SECURITY_TOKEN"
 PERFECTO_CLOUD_NAME_ENV_NAME: str = 'PERFECTO_CLOUD_NAME'
 
-SECURITY_TOKEN_NOT_SET_MESSAGE: str = f"Perfecto Security Token not set. Set environment variable {SECURITY_TOKEN_FILE_ENV_NAME} or {SECURITY_TOKEN_ENV_NAME}"
-PERFECTO_CLOUD_NAME_NOT_SET_MESSAGE: str = f"Perfecto Environment Cloud Name not set. Set environment variable {PERFECTO_CLOUD_NAME_ENV_NAME}"
+SECURITY_TOKEN_NOT_SET_MESSAGE: str = f"Perfecto Security Token not set. Set environment variable {SECURITY_TOKEN_FILE_ENV_NAME} or {SECURITY_TOKEN_ENV_NAME}, or send it as 'Authorization: Bearer <security-token>' when connecting over HTTP"
+PERFECTO_CLOUD_NAME_NOT_SET_MESSAGE: str = f"Perfecto Environment Cloud Name not set. Set environment variable {PERFECTO_CLOUD_NAME_ENV_NAME}, or send the 'Perfecto-Cloud-Name' header when connecting over HTTP"
 
 HELP_TOC_URL = "https://help.perfecto.io/perfecto-help/Data/Tocs/"
 HELP_INDEX_URL = f"{HELP_TOC_URL}perfecto_help.js"

config/runtime.py

@@ -0,0 +1,102 @@
+"""Process-level runtime wiring shared by all tool registrations."""
+from dataclasses import dataclass
+from typing import Any, Literal, Optional
+
+from config.auth import (
+    AuthPort,
+    PERFECTO_USER_CONFIG_STATE_ATTR,
+    HttpAuthProvider,
+    StdioAuthProvider,
+)
+from config.token import PerfectoToken
+
+Transport = Literal["stdio", "streamable-http"]
+
+
+@dataclass(frozen=True)
+class AppRuntime:
+    """Process-level collaborators shared by tool registrations."""
+
+    transport: Transport
+    auth: AuthPort
+    user_config: dict[str, Any]
+
+    def resolve_user_config(self, ctx: Any) -> dict[str, Any]:
+        user_config = dict(self.user_config)
+        user_config.update(_read_ctx_user_config(ctx))
+        token = self.auth.get_token(ctx)
+        if token is not None:
+            user_config["token"] = token
+        return user_config
+
+    def configure_context(self, ctx: Any) -> dict[str, Any]:
+        user_config = self.resolve_user_config(ctx)
+        _hydrate_ctx_user_config(ctx, user_config)
+        return user_config
+
+
+def _read_ctx_user_config(ctx: Any) -> dict[str, Any]:
+    if ctx is None:
+        return {}
+
+    user_config: dict[str, Any] = {}
+    request_context = getattr(ctx, "request_context", None)
+    request = getattr(request_context, "request", None)
+    request_state = getattr(request, "state", None)
+
+    for target, attr_name in (
+        (ctx, "user_config"),
+        (request_context, PERFECTO_USER_CONFIG_STATE_ATTR),
+        (request_state, PERFECTO_USER_CONFIG_STATE_ATTR),
+    ):
+        request_config = getattr(target, attr_name, None)
+        if isinstance(request_config, dict):
+            user_config.update(request_config)
+
+    return user_config
+
+
+def _hydrate_ctx_user_config(ctx: Any, user_config: dict[str, Any]) -> None:
+    if ctx is None:
+        return
+
+    config_copy = dict(user_config)
+    request_context = getattr(ctx, "request_context", None)
+    request = getattr(request_context, "request", None)
+    request_state = getattr(request, "state", None)
+
+    for target in (request_context, request_state):
+        if target is not None:
+            setattr(target, PERFECTO_USER_CONFIG_STATE_ATTR, dict(config_copy))
+
+
+def build_runtime(
+        transport: Transport,
+        startup_token: Optional[PerfectoToken] = None,
+) -> AppRuntime:
+    """
+    Compose auth for the selected transport.
+
+    - stdio: process-lifetime ``startup_token``.
+    - streamable-http: request-scoped Bearer auth.
+    """
+    if transport == "stdio":
+        stdio_user_config = {
+            "startup_token": startup_token,
+            "token": startup_token,
+            "cloud_name": startup_token.cloud_name if startup_token else None,
+        }
+        return AppRuntime(
+            transport=transport,
+            auth=StdioAuthProvider(startup_token),
+            user_config=stdio_user_config,
+        )
+
+    if transport == "streamable-http":
+        return AppRuntime(
+            transport=transport,
+            auth=HttpAuthProvider(),
+            user_config={},
+        )
+
+    raise ValueError(f"Unknown transport: {transport}")

config/token.py

@@ -1,6 +1,6 @@
 from functools import lru_cache
 from pathlib import Path
-from typing import Union
+from typing import Optional, Union
 
 from config.perfecto import SECURITY_TOKEN_NOT_SET_MESSAGE, PERFECTO_CLOUD_NAME_NOT_SET_MESSAGE
 
@@ -52,5 +52,22 @@ def from_file(cls, path: Union[str, Path], cloud_name: str) -> "PerfectoToken":
 
         return cls(token=token_val, cloud_name=cloud_name)
 
+    @classmethod
+    def from_bearer_credentials(cls, credentials: str, cloud_name: Optional[str] = None) -> "PerfectoToken":
+        """
+        Parse Bearer credential material into a PerfectoToken.
+
+        Perfecto credentials are a single security token, so the cloud name is
+        not part of them: it is supplied by the caller from the
+        ``Perfecto-Cloud-Name`` header (falling back to PERFECTO_CLOUD_NAME).
+        Does not call the Perfecto API.
+        """
+        raw = (credentials or "").strip()
+        if not raw:
+            raise PerfectoTokenError("Empty bearer credentials")
+
+        normalized_cloud_name = (cloud_name or "").strip() or None
+        return cls(token=raw, cloud_name=normalized_cloud_name)
+
     def __repr__(self):
         return "<PerfectoToken cloud_name=******** token=********>"