Files
prompts/src/personal_mcp/app.py
T
2026-08-29 20:10:30 -05:00

95 lines
2.6 KiB
Python

from contextlib import asynccontextmanager
from importlib.resources import as_file
from importlib.resources import files
from pathlib import Path
from fastapi import FastAPI
from fastapi import Response
from fastapi import status
from fastapi.responses import RedirectResponse
from fastapi.staticfiles import StaticFiles
from .config import Settings
from .config import get_settings
from .mcp import create_mcp
def create_app(settings: Settings | None = None) -> FastAPI:
runtime_settings = settings if settings is not None else get_settings()
mcp_app = create_mcp().http_app(
json_response=True,
stateless_http=True,
transport="http",
)
app = FastAPI(
debug=runtime_settings.debug,
docs_url=None,
redoc_url=None,
openapi_url=None,
lifespan=app_lifespan,
)
app.state.settings = runtime_settings
app.mount(runtime_settings.mounts.mcp, mcp_app, name="mcp")
return app
@asynccontextmanager
async def app_lifespan(app: FastAPI):
from . import __name__ as package_root_name
site_resource = files(package_root_name).joinpath("site")
with as_file(site_resource) as site_dir:
mount_docs(
app,
docs_route=app.state.settings.mounts.docs,
site_dir=site_dir,
)
yield
def mount_docs(app: FastAPI, *, docs_route: str, site_dir: Path) -> None:
"""Mount the pre-built static docs site, or expose a clear missing-build response."""
normalized_route = docs_route.rstrip("/") or "/docs"
docs_root = f"{normalized_route}/"
async def redirect_to_docs_root() -> RedirectResponse:
return RedirectResponse(
url=docs_root,
status_code=status.HTTP_307_TEMPORARY_REDIRECT,
)
app.add_api_route(
normalized_route,
redirect_to_docs_root,
methods=["GET", "HEAD"],
include_in_schema=False,
)
if site_dir.is_dir():
app.mount(
normalized_route,
StaticFiles(directory=site_dir, html=True),
name="docs",
)
return
async def docs_not_built() -> Response:
return Response(
content=("Static docs have not been built yet. Run `uv run zensical build` before using this route."),
media_type="text/plain",
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
)
app.add_api_route(
normalized_route,
docs_not_built,
methods=["GET"],
include_in_schema=False,
)
app.add_api_route(
f"{normalized_route}/{{path:path}}",
docs_not_built,
methods=["GET"],
include_in_schema=False,
)