src/mcp_sourcecontrol/http_app.py - master - pickling/mcp-sourcecontrol-python

src/mcp_sourcecontrol/http_app.py

58 строк2 KB

pickling-21

Python-порт mcp-sourcecontrol: 20 инструментов, stdio и Streamable HTTP

29 июл 2026, 02:18

29 июл 2026, 02:18bc8bde4

"""HTTP-режим: Streamable HTTP приложение с per-запросной конфигурацией из заголовков.

Аналог HTTP-части src/index.ts. Вместо реестра сессий (Map + TTL в TS-версии)
используется stateless-режим SDK: заголовки X-SC-Url/X-SC-Token разбираются
middleware'ом на каждом запросе, так что смена окружения работает точно так же.
"""

from __future__ import annotations

from mcp.server.fastmcp import FastMCP
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route

from .config import ResolvedConfig, resolve_session_config
from .log import logger
from .runtime import reset_session_config, use_session_config


class ScHeaderMiddleware:
    """ASGI-middleware: X-SC-Url/X-SC-Token каждого запроса → конфигурация сессии."""

    def __init__(self, app, default_config: ResolvedConfig) -> None:
        self.app = app
        self.default_config = default_config

    async def __call__(self, scope, receive, send) -> None:
        if scope["type"] != "http":
            await self.app(scope, receive, send)
            return

        headers = {
            key.decode("latin-1").lower(): value.decode("latin-1")
            for key, value in scope.get("headers") or []
        }
        session_config = resolve_session_config(headers, self.default_config)
        if session_config is not self.default_config:
            logger.debug(
                "session config resolved from headers",
                {"apiUrl": session_config.api_url, "hasToken": session_config.token is not None},
            )

        token = use_session_config(session_config)
        try:
            await self.app(scope, receive, send)
        finally:
            reset_session_config(token)


async def _health(_request: Request) -> JSONResponse:
    return JSONResponse({"status": "ok", "mode": "http"})


def build_http_app(server: FastMCP, default_config: ResolvedConfig):
    """Starlette-приложение SDK (/mcp) + /health, обёрнутые в header-middleware."""
    app = server.streamable_http_app()
    app.router.routes.append(Route("/health", _health, methods=["GET"]))
    return ScHeaderMiddleware(app, default_config)