MOB-52691: prototype remote Streamable HTTP transport (spike)

README.md

@@ -158,6 +158,65 @@ The easiest way to configure your MCP client is using our interactive CLI tool:
 
 ---
 
+### Streamable HTTP Hosting
+
+The default transport remains local `stdio`. To run a remote MCP endpoint, start the server with the
+opt-in `http` transport:
+
+```bash
+FASTMCP_HOST=127.0.0.1 FASTMCP_PORT=8000 uv run python main.py --mcp http
+```
+
+The MCP endpoint is available at `http://127.0.0.1:8000/mcp`. Configure a remote MCP client with the
+endpoint URL and send the caller's API Monitoring token on every request:
+
+```json
+{
+  "mcpServers": {
+    "BlazeMeter API Test MCP": {
+      "url": "https://mcp.example.com/mcp",
+      "headers": {
+        "Authorization": "Bearer <api-monitoring-token>"
+      }
+    }
+  }
+}
+```
+
+For a hosted deployment, do not configure `BZM_API_TEST_TOKEN` or `BZM_API_TEST_TOKEN_FILE` as a
+server-wide credential. HTTP mode resolves the Bearer token from each incoming request so each caller
+operates only on the teams, buckets, and tests available to that token.
+
+`/health` and `/healthz` are available for unauthenticated load-balancer health checks. All MCP traffic
+requires a valid Bearer header. Requests with an `Origin` header are rejected in this MVP. This protects
+the endpoint from browser-based DNS-rebinding attacks while allowing the usual desktop, IDE, and service
+MCP clients, which do not need browser CORS. Browser clients are not supported until an explicit CORS and
+origin allowlist design is added.
+
+The current MVP deliberately uses the caller's API Monitoring token as the Bearer credential and passes it
+to the Runscope API. A future phase may add separate authentication for the MCP server itself, such as an
+OAuth-based identity layer, with a trusted mapping to the caller's API Monitoring credential. The current
+API-key pass-through should therefore be treated as a product-specific MVP decision, not a complete OAuth
+authorization implementation.
+
+The hosted endpoint uses stateless request handling because the current API Test tools do not retain
+application state between calls. Each request is independently authenticated and processed, so server
+restarts do not invalidate an application session and multiple server instances do not require sticky
+session routing. Future features that need MCP session state will require an explicit state-store or a
+return to stateful transport handling.
+
+Use HTTPS and configure the public host and port through your deployment environment:
+
+```bash
+BZM_API_TEST_MCP_TRANSPORT=http \
+FASTMCP_HOST=0.0.0.0 \
+FASTMCP_PORT=8000 \
+FASTMCP_STREAMABLE_HTTP_PATH=/mcp \
+python main.py --mcp
+```
+
+---
+
 **Docker MCP Client Configuration**
 
 ```json

main.py

@@ -8,13 +8,15 @@
 from mcp.server.fastmcp import FastMCP
 
 from src.common.telemetry import init_telemetry
+from src.config.auth import run_streamable_http
 from src.config.token import BzmApimToken, BzmApimTokenError
 from src.config.version import __executable__, __version__
 from src.server import register_tools
 
 BLAZEMETER_APIM_KEY_FILE_PATH = os.getenv("BZM_API_TEST_TOKEN_FILE")
 
 LOG_LEVELS = Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
+MCP_TRANSPORTS = ("stdio", "http")
 
 
 def init_logging(level_name: str) -> None:
@@ -61,14 +63,22 @@ def get_api_token():
     return token
 
 
-def run(log_level: str = "CRITICAL", base_url: str = None):
-    if base_url:
-        import src.config.defaults as defaults
+def resolve_mcp_transport(raw_cli_transport: str) -> str:
+    """Resolve the MCP transport from CLI, environment, then stdio default."""
+    candidate = raw_cli_transport.strip() or os.getenv("BZM_API_TEST_MCP_TRANSPORT", "").strip()
+    if not candidate:
+        return "stdio"
+
+    transport = candidate.lower()
+    if transport not in MCP_TRANSPORTS:
+        allowed = ", ".join(MCP_TRANSPORTS)
+        raise ValueError(f"Invalid MCP transport '{candidate}'. Valid values: {allowed}.")
+    return transport
 
-        defaults.BZM_APIM_BASE_URL = base_url
 
+def build_mcp_server(log_level: str = "CRITICAL", transport: str = "stdio") -> tuple[FastMCP, str]:
+    """Build an API Test MCP server for local stdio or hosted HTTP transport."""
     init_telemetry("mcp-bzm-apitest", __version__)
-    token = get_api_token()
     instructions = """
     # BlazeMeter API Test MCP Server
     This MCP server provides AI assistants with programmatic access to BlazeMeter's
@@ -98,19 +108,51 @@ def run(log_level: str = "CRITICAL", base_url: str = None):
             steps: Test steps belong to a particular test.
             results: Test execution results belong to a particular test.
     """
+    host = "127.0.0.1"
+    port = 8000
+    if transport == "http":
+        host = os.getenv("FASTMCP_HOST", "127.0.0.1").strip() or "127.0.0.1"
+        port = int((os.getenv("FASTMCP_PORT") or os.getenv("PORT") or "8000").strip() or "8000")
+
+    wire_transport = "streamable-http" if transport == "http" else "stdio"
     mcp = FastMCP(
-        "blazemeter-apitest-mcp", instructions=instructions, log_level=cast(LOG_LEVELS, log_level)
+        "blazemeter-apitest-mcp",
+        instructions=instructions,
+        log_level=cast(LOG_LEVELS, log_level),
+        host=host,
+        port=port,
+        streamable_http_path=os.getenv("FASTMCP_STREAMABLE_HTTP_PATH", "/mcp").strip() or "/mcp",
+        stateless_http=True,
     )
-    register_tools(mcp, token)
-    mcp.run(transport="stdio")
+    register_tools(mcp, get_api_token() if wire_transport == "stdio" else None, hosted=transport == "http")
+    return mcp, wire_transport
+
+
+def run(log_level: str = "CRITICAL", base_url: str = None, transport: str = "stdio"):
+    if base_url:
+        import src.config.defaults as defaults
+
+        defaults.BZM_APIM_BASE_URL = base_url
+
+    mcp, wire_transport = build_mcp_server(log_level=log_level, transport=transport)
+    if wire_transport == "stdio":
+        mcp.run(transport="stdio")
+    else:
+        run_streamable_http(mcp)
 
 
 def main():
     parser = argparse.ArgumentParser(prog="mcp-bzm-apitest")
 
     parser.add_argument("--version", action="version", version=f"%(prog)s {__version__}")
 
-    parser.add_argument("--mcp", action="store_true", help="Execute MCP Server")
+    parser.add_argument(
+        "--mcp",
+        nargs="?",
+        const="",
+        metavar="TRANSPORT",
+        help="Execute MCP Server. Optional TRANSPORT values: stdio or http.",
+    )
 
     parser.add_argument(
         "--log-level",
@@ -130,8 +172,12 @@ def main():
     args = parser.parse_args()
     init_logging(args.log_level)
 
-    if args.mcp:
-        run(log_level=args.log_level.upper(), base_url=args.base_url)
+    if args.mcp is not None:
+        run(
+            log_level=args.log_level.upper(),
+            base_url=args.base_url,
+            transport=resolve_mcp_transport(args.mcp),
+        )
     else:
 
         logo_ascii = (

src/config/auth.py

@@ -0,0 +1,109 @@
+import anyio
+import uvicorn
+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 src.config.token import BzmApimToken, BzmApimTokenError
+
+HEALTH_PATHS = frozenset({"/health", "/healthz"})
+
+
+def parse_bearer_token(authorization: str | None) -> BzmApimToken | None:
+    """Parse an API Test token from an HTTP Authorization header."""
+    if not authorization:
+        return None
+
+    scheme, separator, credentials = authorization.strip().partition(" ")
+    if scheme.lower() != "bearer" or not separator or not credentials.strip():
+        return None
+
+    try:
+        return BzmApimToken(credentials.strip())
+    except BzmApimTokenError:
+        return None
+
+
+class TokenResolver:
+    """Resolve API Test credentials for local or hosted tool invocations."""
+
+    def __init__(self, startup_token: BzmApimToken | str | None, hosted: bool = False):
+        # api_request interpolates this into the header, and BzmApimToken.__repr__ masks itself.
+        self._startup_token = (
+            startup_token.token if isinstance(startup_token, BzmApimToken) else startup_token
+        )
+        self._hosted = hosted
+
+    def get_token(self, ctx: Context) -> str | None:
+        if not self._hosted:
+            return self._startup_token
+
+        request_context = getattr(ctx, "request_context", None)
+        request = getattr(request_context, "request", None)
+        headers = getattr(request, "headers", None)
+        authorization = headers.get("authorization") if headers is not None else None
+        token = parse_bearer_token(authorization)
+        return token.token if token is not None else None
+
+
+class HttpSecurityMiddleware:
+    """Reject untrusted browser origins and unauthenticated MCP requests."""
+
+    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
+
+        request = Request(scope, receive)
+        origin = request.headers.get("origin")
+        if origin:
+            response = JSONResponse({"error": "Forbidden origin"}, status_code=403)
+            await response(scope, receive, send)
+            return
+
+        if (
+            scope.get("path") not in HEALTH_PATHS
+            and parse_bearer_token(request.headers.get("authorization")) is None
+        ):
+            response = JSONResponse(
+                {"error": "Unauthorized"},
+                status_code=401,
+                headers={"WWW-Authenticate": "Bearer"},
+            )
+            await response(scope, receive, send)
+            return
+
+        await self.app(scope, receive, send)
+
+
+def register_health_routes(mcp: FastMCP) -> None:
+    """Register unauthenticated health probes for a hosted MCP deployment."""
+
+    @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 behind the HTTP security boundary."""
+    register_health_routes(mcp)
+
+    async def serve() -> None:
+        app = HttpSecurityMiddleware(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)

src/server.py

@@ -1,5 +1,6 @@
 from typing import Optional
 
+from src.config.auth import TokenResolver
 from src.config.token import BzmApimToken
 from src.tools.bucket_manager import register as register_bucket_manager
 from src.tools.environment_manager import register as register_environment_manager
@@ -11,19 +12,20 @@
 from src.tools.version_manager import register as register_version_manager
 
 
-def register_tools(mcp, token: Optional[BzmApimToken]):
+def register_tools(mcp, token: Optional[BzmApimToken], hosted: bool = False):
     """
     Register all available tools with the MCP server.
 
     Args:
             mcp: The MCP server instance
             token: Optional BlazeMeter API Test token (can be None if not configured)
     """
+    token_resolver = TokenResolver(token, hosted=hosted)
     register_version_manager(mcp, token)
-    register_result_manager(mcp, token)
-    register_team_manager(mcp, token)
-    register_bucket_manager(mcp, token)
-    register_test_manager(mcp, token)
-    register_schedule_manager(mcp, token)
-    register_step_manager(mcp, token)
-    register_environment_manager(mcp, token)
+    register_result_manager(mcp, token_resolver)
+    register_team_manager(mcp, token_resolver)
+    register_bucket_manager(mcp, token_resolver)
+    register_test_manager(mcp, token_resolver)
+    register_schedule_manager(mcp, token_resolver)
+    register_step_manager(mcp, token_resolver)
+    register_environment_manager(mcp, token_resolver)

src/tools/bucket_manager.py

@@ -14,6 +14,7 @@
     record_span_error,
     tool_span,
 )
+from src.config.auth import TokenResolver
 from src.config.defaults import BUCKETS_ENDPOINT, TOOLS_PREFIX
 from src.config.token import BzmApimToken
 from src.formatters.bucket import format_buckets
@@ -44,7 +45,7 @@ async def list(self) -> BaseResult:
         return await api_request(self.token, "GET", f"{BUCKETS_ENDPOINT}", result_formatter=format_buckets)
 
 
-def register(mcp, token: Optional[BzmApimToken]):
+def register(mcp, token_resolver: TokenResolver):
     @mcp.tool(
         name=f"{TOOLS_PREFIX}_buckets",
         description="""
@@ -69,7 +70,7 @@ def register(mcp, token: Optional[BzmApimToken]):
         """,
     )
     async def buckets(action: str, args: Dict[str, Any], ctx: Context) -> BaseResult:
-        bucket_manager = BucketManager(token, ctx)
+        bucket_manager = BucketManager(token_resolver.get_token(ctx), ctx)
         meta = get_meta_from_ctx(ctx)
         parent_context = extract_trace_context(meta)
         async with tool_span(f"{TOOLS_PREFIX}_buckets", action, parent_context) as span:

src/tools/environment_manager.py

@@ -14,6 +14,7 @@
     record_span_error,
     tool_span,
 )
+from src.config.auth import TokenResolver
 from src.config.defaults import TEST_ENVIRONMENT_ENDPOINT, TOOLS_PREFIX
 from src.config.token import BzmApimToken
 from src.formatters.environment import format_environments
@@ -46,7 +47,7 @@ async def list(self, bucket_key: str, test_id: str) -> BaseResult:
         )
 
 
-def register(mcp, token: Optional[BzmApimToken]):
+def register(mcp, token_resolver: TokenResolver):
     @mcp.tool(
         name=f"{TOOLS_PREFIX}_environments",
         description="""
@@ -72,7 +73,7 @@ def register(mcp, token: Optional[BzmApimToken]):
         """,
     )
     async def environments(action: str, args: Dict[str, Any], ctx: Context) -> BaseResult:
-        environment_manager = EnvironmentManager(token, ctx)
+        environment_manager = EnvironmentManager(token_resolver.get_token(ctx), ctx)
         meta = get_meta_from_ctx(ctx)
         parent_context = extract_trace_context(meta)
         async with tool_span(f"{TOOLS_PREFIX}_environments", action, parent_context) as span: