app/backend/fluksio/mcp/http.py

"""Mounting the MCP endpoint, and deciding whose tokens it accepts.

The transport is streamable HTTP, stateless, one JSON response per call: no
sticky sessions to route and nothing for a proxy to buffer. The SDK serves both
``/mcp`` and the protected-resource metadata beside it, which is why the sub-app
is mounted at the root rather than under ``/mcp``.
"""

from __future__ import annotations

import logging
from urllib.parse import urlparse

import httpx
import jwt
from fastapi.concurrency import run_in_threadpool
from mcp.server.auth.provider import AccessToken
from mcp.server.auth.settings import AuthSettings
from mcp.server.transport_security import TransportSecuritySettings
from pydantic import AnyHttpUrl
from starlette.applications import Starlette

from fluksio.api.deps import oauth_client_lives
from fluksio.core import security
from fluksio.core.config import settings
from fluksio.mcp import server

logger = logging.getLogger(__name__)

#: The tool calls never leave the process, so the host is a label, not a route.
_INTERNAL_BASE = "http://fluksio-mcp.internal"


class _JWTVerifier:
    """Accept only tokens minted for the MCP channel, by agents still registered.

    A perfectly valid browser token is refused: it was issued for a person's
    session, and honouring it here would make agent traffic indistinguishable
    from theirs.

    The client row is looked up because the token itself cannot be withdrawn:
    it is stateless and good until ``MCP_TOKEN_EXPIRE_MINUTES`` runs out, so
    the registration it names is the thing revoking an agent actually removes.
    Refusing at the door rather than leaving it to the API the tools call means
    a revoked agent gets the 401 that sends it back to authorize, instead of a
    tool listing that works and a tool call that does not.
    """

    async def verify_token(self, token: str) -> AccessToken | None:
        try:
            payload = security.decode_oauth_token(token)
        except jwt.InvalidTokenError as exc:
            logger.debug("MCP token rejected: %s", exc)
            return None
        if payload.get("mcp") is not True:
            logger.debug("MCP token rejected: not an MCP-channel token")
            return None
        client_id = str(payload.get("client_id", ""))
        # Off the event loop: the lookup is SQLite, like every other read this
        # process makes, and the session it opens is blocking.
        if not await run_in_threadpool(oauth_client_lives, client_id):
            logger.debug(
                "MCP token rejected: agent %s is no longer registered", client_id
            )
            return None
        return AccessToken(
            token=token,
            client_id=client_id,
            scopes=[security.MCP_SCOPE],
            subject=str(payload.get("sub", "")),
        )


def _transport_security() -> TransportSecuritySettings:
    """Which Host headers to trust.

    The SDK defaults to localhost only and answers 421 to anything arriving
    through a reverse proxy under the real hostname, so the deployment's own
    host is named here rather than the protection being switched off.
    """
    host = urlparse(settings.oauth_issuer).netloc
    allowed_hosts = [host, f"{host}:*"]
    allowed_origins = [f"https://{host}", f"http://{host}"]
    if settings.ENVIRONMENT == "local":
        # The test client and a direct uvicorn run do not go through Traefik.
        allowed_hosts += [
            "localhost",
            "localhost:*",
            "127.0.0.1",
            "127.0.0.1:*",
            "testserver",
        ]
        allowed_origins += ["http://localhost", "http://127.0.0.1"]
    return TransportSecuritySettings(
        enable_dns_rebinding_protection=True,
        allowed_hosts=allowed_hosts,
        allowed_origins=allowed_origins,
    )


def build_http_app(app: object) -> Starlette:
    """The MCP sub-app, wired to this app's API and its OAuth server."""
    server.mcp.settings.stateless_http = True
    server.mcp.settings.json_response = True
    server.mcp.settings.auth = AuthSettings(
        issuer_url=AnyHttpUrl(settings.oauth_issuer),
        resource_server_url=AnyHttpUrl(settings.mcp_resource),
        required_scopes=None,
    )
    server.mcp.settings.transport_security = _transport_security()
    # The SDK only takes a verifier through its constructor, and the instance is
    # module-level so the tools can be declared at import time.
    server.mcp._token_verifier = _JWTVerifier()  # noqa: SLF001

    server.set_client(
        httpx.AsyncClient(
            transport=httpx.ASGITransport(app=app),  # type: ignore[arg-type]
            base_url=_INTERNAL_BASE,
            # Running a flow is synchronous work on a threadpool, so a slow
            # flow must not look like a dead endpoint.
            timeout=120.0,
            event_hooks={"request": [server.forward_caller_auth]},
        )
    )
    return server.mcp.streamable_http_app()


async def aclose() -> None:
    if server._client is not None:  # noqa: SLF001
        await server._client.aclose()  # noqa: SLF001
        server.set_client(None)