MCP inside FastAPI

from contextlib import asynccontextmanager

from fastapi import FastAPI
from fastmcp import FastMCP
from fastmcp.utilities.lifespan import combine_lifespans

import store

mcp = FastMCP("docs-assistant")
started = {"app": False}


@mcp.tool(annotations={"readOnlyHint": True})
def search_docs(query: str, component: str, top_k: int = 3) -> dict:
    """Search the documentation of one component."""
    return {"query": query, "component": component, "hits": store.search(query, component, top_k)}


@asynccontextmanager
async def app_lifespan(app: FastAPI):
    started["app"] = True          # e.g. open a database pool here
    yield


mcp_app = mcp.http_app(path="/mcp")
app = FastAPI(lifespan=combine_lifespans(app_lifespan, mcp_app.lifespan))


@app.get("/api/health")
def health() -> dict:
    return {"app_lifespan_ran": started["app"]}


app.mount("/", mcp_app)

"""A small REST API whose OpenAPI document feeds the auto-generation examples."""
from fastapi import FastAPI, HTTPException

import store

api = FastAPI(title="docs-api")


@api.get("/components", operation_id="list_components")
def list_components() -> list[str]:
    return store.COMPONENTS


@api.get("/search", operation_id="search_docs")
def search(query: str, component: str, top_k: int = 3) -> list[dict]:
    return store.search(query, component, top_k)


@api.delete("/pages/{component}/{page}", operation_id="delete_page")
def delete_page(component: str, page: str) -> dict:
    """A destructive endpoint that auto-generation would expose as a tool."""
    raise HTTPException(status_code=501, detail="not implemented in the demo")

list_components   (no annotations)
search_docs       (no annotations)
delete_page       (no annotations)

from fastmcp import FastMCP
from fastmcp.server.providers.openapi import MCPType, RouteMap

from rest_api import api

mcp = FastMCP.from_fastapi(
    app=api,
    name="docs-api-mcp",
    route_maps=[RouteMap(methods=["DELETE", "POST", "PUT", "PATCH"], mcp_type=MCPType.EXCLUDE)],  # read-only surface
)

from fastapi_mcp import FastApiMCP

from rest_api import api

mcp = FastApiMCP(api, name="docs-api-mcp", include_operations=["list_components", "search_docs"])
mcp.mount_http()

"""One FastAPI process, one token verifier, one service: REST routes and MCP tools share all three."""
import os
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import Annotated

import anyio
import jwt
from fastapi import Depends, FastAPI, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from jwt import PyJWKClient
from mcp.server import MCPServer
from mcp.server.auth.middleware.auth_context import get_access_token
from mcp.server.auth.provider import AccessToken
from mcp.server.auth.settings import AuthSettings
from mcp.types import ToolAnnotations
from pydantic import BaseModel

from service import DocsService, docs_service

ISSUER = os.environ["MCP_ISSUER"]
AUDIENCE = os.environ["MCP_AUDIENCE"]
JWKS_URI = os.environ["MCP_JWKS_URI"]


class JwtVerifier:
    """The single place that decides whether a bearer token is acceptable."""

    def __init__(self) -> None:
        self._jwks = PyJWKClient(JWKS_URI)

    async def verify_token(self, token: str) -> AccessToken | None:
        try:
            # PyJWKClient fetches keys with blocking HTTP: run it off the event loop so a JWKS refresh
            # does not stall every REST and MCP request in this process.
            signing_key = await anyio.to_thread.run_sync(self._jwks.get_signing_key_from_jwt, token)
            key = signing_key.key
            claims = jwt.decode(token, key, algorithms=["RS256", "ES256"], issuer=ISSUER, audience=AUDIENCE,
                                options={"require": ["exp", "iss", "aud"]})
        except jwt.PyJWTError:
            return None
        return AccessToken(token=token, client_id=str(claims.get("azp", "unknown")),
                           scopes=str(claims.get("scope", "")).split(), expires_at=int(claims["exp"]),
                           resource=AUDIENCE, subject=claims.get("sub"))


verifier = JwtVerifier()
bearer = HTTPBearer(auto_error=False)


# --- REST side: a FastAPI dependency wrapping the shared verifier -------------------------------------------
async def current_token(creds: Annotated[HTTPAuthorizationCredentials | None, Depends(bearer)]) -> AccessToken:
    token = await verifier.verify_token(creds.credentials) if creds else None
    if token is None:
        raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid or missing token",
                            headers={"WWW-Authenticate": "Bearer"})
    if "docs:read" not in token.scopes:
        raise HTTPException(status.HTTP_403_FORBIDDEN, "insufficient scope")
    return token


def service() -> DocsService:
    return docs_service


# --- MCP side: the same verifier, the same service ----------------------------------------------------------
class Hit(BaseModel):
    component: str
    page: str
    title: str
    score: int
    snippet: str


class SearchResult(BaseModel):
    query: str
    component: str
    caller: str | None
    hits: list[Hit]


mcp = MCPServer(
    "docs-assistant",
    version="1.0.0",
    token_verifier=verifier,
    auth=AuthSettings(issuer_url=ISSUER, resource_server_url=AUDIENCE, required_scopes=["docs:read"],
                      validate_token_resource=True),
)


@mcp.tool(annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=False))
async def search_docs(query: str, component: str, top_k: int = 3) -> SearchResult:
    """Search the documentation of one component."""
    token = get_access_token()                       # the token the MCP middleware already verified
    hits = docs_service.search(query, component, top_k)
    return SearchResult(query=query, component=component, caller=token.subject if token else None,
                        hits=[Hit(**h) for h in hits])


mcp_app = mcp.streamable_http_app()


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    async with mcp.session_manager.run():
        yield


app = FastAPI(title="docs-assistant", lifespan=lifespan)


@app.get("/api/search")
def rest_search(query: str, component: str, top_k: int = 3, _: AccessToken = Depends(current_token),
                svc: DocsService = Depends(service)) -> list[dict]:
    return svc.search(query, component, top_k)


app.mount("/", mcp_app)

"""Business logic shared by the REST routes and the MCP tools. Neither transport owns it."""
import store


class DocsService:
    def search(self, query: str, component: str, top_k: int = 3) -> list[dict]:
        if component not in store.COMPONENTS:
            raise ValueError(f"Unknown component '{component}'")
        return store.search(query, component, top_k)


docs_service = DocsService()