Python API

def create_app(config: AdaptConfig) -> FastAPI:
    """Create and configure the FastAPI application.

    Args:
        config: The application configuration.

    Returns:
        The configured FastAPI app instance.
    """
    logger.debug("Creating FastAPI app with config: %s", config)
    engine, lock_manager, resources = _init_infrastructure(config)

    app = FastAPI(title="Adapt", version=config.version, lifespan=lifespan, docs_url=None, redoc_url=None, openapi_url=None)
    app.state.config = config
    app.state.db_engine = engine
    app.state.use_tls = bool(config.tls_cert and config.tls_key)
    app.state.lock_manager = lock_manager
    app.state.resources = resources
    app.state.resource_registry = build_resource_registry(resources, config)

    allowed_hosts = build_allowed_hosts(config.host)
    if allowed_hosts != ["*"]:
        app.add_middleware(TrustedHostMiddleware, allowed_hosts=allowed_hosts)

    # Set up Jinja2 templates
    templates_dir = Path(__file__).parent / "templates"
    templates = Jinja2Templates(directory=str(templates_dir))
    # Search snippets hold raw docroot text; this escapes them and restores only
    # the <mark> highlights. See routes_search.safe_snippet.
    templates.env.filters["safe_snippet"] = safe_snippet
    app.state.templates = templates

    # Mount static files
    static_dir = Path(__file__).parent / "static"
    app.mount("/static", StaticFiles(directory=str(static_dir)), name="static")

    # Authentication middleware
    @app.middleware("http")
    async def security_middleware(request: Request, call_next):
        """Middleware for CSRF validation and response security headers."""
        csrf_token = request.cookies.get("adapt_csrf")
        should_set_csrf_cookie = False
        if not csrf_token:
            csrf_token = generate_csrf_token()
            should_set_csrf_cookie = True

        request.state.csrf_token = csrf_token

        csrf_error = await validate_csrf(request)
        if csrf_error:
            apply_security_headers(csrf_error, use_tls=request.app.state.use_tls)
            if should_set_csrf_cookie:
                set_csrf_cookie(csrf_error, csrf_token, secure=request.app.state.config.secure_cookies)
            return csrf_error

        response = await call_next(request)
        apply_security_headers(response, use_tls=request.app.state.use_tls)
        if should_set_csrf_cookie:
            set_csrf_cookie(response, csrf_token, secure=request.app.state.config.secure_cookies)
        return response

    @app.middleware("http")
    async def auth_middleware(request: Request, call_next):
        """Middleware to handle user authentication via session cookies."""
        token = request.cookies.get("adapt_session")
        request.state.user = None
        if token:
            with Session(engine) as db:
                sess = get_session(db, token)  # Uses fixed function with expiration check
                if sess:
                    user = db.get(User, sess.user_id)
                    request.state.user = user
                    logger.debug("Authenticated user %s for request %s", user.username if user else None, request.url)
                else:
                    logger.debug("Invalid or expired session token for request %s", request.url)
        else:
            logger.debug("No session token in request %s", request.url)
        response = await call_next(request)
        return response

    # Mount authentication routes
    app.include_router(auth_router, prefix="", tags=["auth"])

    # Mount admin routes
    app.include_router(admin_router)

    # Mount search routes
    app.include_router(search_router)

    # Generate and mount routes
    generate_routes(app, app.state.resource_registry)

    # Mount the MCP server, exposing resources as agent-facing tools
    if config.mcp_enabled:
        mcp_server = build_mcp_server(config)
        mcp_app = mcp_server.streamable_http_app()
        # A mounted sub-app's `request.app` is itself, not the main app (see
        # lifespan() docstring for the related lifespan gotcha) — mirror the
        # slice of state every tool/helper needs onto it.
        mcp_app.state.db_engine = engine
        mcp_app.state.resources = resources
        mcp_app.state.config = config
        mcp_app.state.lock_manager = lock_manager
        mcp_app.state.resource_registry = app.state.resource_registry
        # The outer app's own middleware still runs for /mcp requests (it wraps
        # routing, including the Mount), and by the time it inspects
        # `request.app.state` post-routing, `scope["app"]` has already been
        # overwritten to `mcp_app` — so anything that middleware reads off
        # app.state must be mirrored here too, not just what the tools need.
        mcp_app.state.use_tls = app.state.use_tls
        app.state.mcp_server = mcp_server
        app.mount("/mcp", mcp_app)

    @app.get("/openapi.json", include_in_schema=False)
    def openapi_schema(request: Request):
        """Return an OpenAPI document filtered to the current user's visible routes."""
        user = get_current_user(request)
        return JSONResponse(_build_openapi_schema(app, request, user))

    @app.get("/docs", include_in_schema=False)
    @app.get("/docs/", include_in_schema=False)
    def swagger_ui():
        """Render Swagger UI against the request-filtered OpenAPI schema."""
        return get_swagger_ui_html(
            openapi_url="/openapi.json",
            title=f"{app.title} - API Docs",
            oauth2_redirect_url="/docs/oauth2-redirect",
        )

    @app.get("/docs/oauth2-redirect", include_in_schema=False)
    def swagger_ui_redirect():
        """Serve the Swagger UI OAuth redirect helper."""
        return get_swagger_ui_oauth2_redirect_html()

    @app.get("/health", tags=["system"])
    async def health(request: Request, user=Depends(get_current_user)):
        """
        Health check endpoint.
        - Unauthenticated: returns minimal info (status, version, timestamp)
        - Authenticated: adds uptime, cache size, and endpoint count
        """
        info = {
            "status": "ok",
            "version": getattr(config, "version", "unknown"),
            "timestamp": datetime.now(timezone.utc).isoformat() + "Z"
        }
        if user:
            # Add extra info for authenticated users
            uptime = time.time() - _START_TIME
            # Try to get cache size if available
            cache_size = None
            try:
                conn = cache._get_conn()
                cursor = conn.cursor()
                cursor.execute(f"SELECT COUNT(*) FROM {cache.CACHE_TABLE}")
                cache_size = cursor.fetchone()[0]
                conn.close()
            except Exception:
                pass
            # Count endpoints
            endpoint_count = len(app.routes)
            info.update({
                "uptime_seconds": int(uptime),
                "cache_size": cache_size,
                "endpoint_count": endpoint_count
            })
        return JSONResponse(info)


    # Media gallery route
    @app.get("/ui/media")
    def media_gallery(request: Request):
        """Render the media gallery UI for authenticated users."""
        user = get_current_user(request)
        if not user:
            logger.debug("Unauthenticated access to media gallery, redirecting to login")
            return RedirectResponse(url=login_redirect_url("/ui/media"), status_code=302)

        all_media = [r for r in request.app.state.resources if r.resource_type == "media"]
        if getattr(user, "is_superuser", False):
            permitted_media = all_media
        else:
            with Session(engine) as db:
                checker = PermissionChecker(db)
                permitted_media = [
                    r for r in all_media
                    if checker.has_permission(user, r.relative_path.with_suffix("").as_posix(), "read")
                ]
        if not permitted_media and not getattr(user, "is_superuser", False):
            logger.warning("Permission denied for user %s: no accessible media resources", user.username)
            raise HTTPException(status_code=403, detail="No accessible media resources")

        media_items = []
        for r in permitted_media:
            media_items.append({
                "name": r.path.name,
                "relative_path": r.relative_path.as_posix(),
                "media_type": r.metadata.get("media_type", "unknown"),
                "file_size": r.metadata.get("file_size", 0),
                "duration": r.metadata.get("duration"),
                "bitrate": r.metadata.get("bitrate"),
                "title": r.metadata.get("title"),
                "artist": r.metadata.get("artist"),
                "album": r.metadata.get("album"),
                "genre": r.metadata.get("genre"),
                "thumbnail": r.metadata.get("thumbnail"),
            })
        accessible_resources = build_accessible_ui_links(request, user)
        context = {
            "media_items": media_items,
            "user": user,
            "ui_links": accessible_resources,
            "is_superuser": user and getattr(user, "is_superuser", False)
        }
        logger.debug("Rendering media gallery for user %s with %d items", user.username, len(media_items))
        return request.app.state.templates.TemplateResponse(request, "media_gallery.html", context)

    # Debug root route
    @app.get("/")
    def root(request: Request):
        """Handle root requests, rendering HTML landing page or JSON API response."""
        accept = request.headers.get("accept", "")
        if "text/html" in accept:
            # Render landing page
            user = get_current_user(request)
            accessible_resources = build_accessible_ui_links(request, user)

            # Add media gallery link only if the user can access at least one media file
            if any(link["type"] == "media" for link in accessible_resources):
                accessible_resources.append({"name": "Media Gallery", "url": "/ui/media", "type": "media"})

            context = {
                "user": user,
                "ui_links": accessible_resources,
                "is_superuser": user and getattr(user, "is_superuser", False)
            }
            logger.debug("Rendering HTML landing page for user %s", user.username if user else None)
            return request.app.state.templates.TemplateResponse(request, "landing.html", context)
        else:
            # JSON API response
            user = get_current_user(request)
            resources = _visible_resource_paths(request, user)
            logger.debug("Returning JSON API response with %d resources", len(resources))
            return {"resources": resources}

    # Exception handler for auth redirects
    @app.exception_handler(HTTPException)
    async def auth_exception_handler(request: Request, exc: HTTPException):
        """Handle HTTP exceptions, redirecting to login for 401 errors in HTML requests."""
        if exc.status_code == 401:
            accept = request.headers.get("accept", "")
            if "text/html" in accept:
                logger.debug("Redirecting unauthenticated request to login for %s", request.url)
                return RedirectResponse(url=login_redirect_url(request.url.path), status_code=302)

        logger.warning("HTTP exception %d: %s for request %s", exc.status_code, exc.detail, request.url)
        return JSONResponse(
            status_code=exc.status_code,
            content={"detail": exc.detail},
        )

    return app

def discover_resources(root: Path, config: AdaptConfig) -> list[DatasetResource]:
    """Discover dataset resources in the root directory.

    Args:
        root: The root directory to search.
        config: The Adapt configuration.

    Returns:
        A list of discovered DatasetResource objects.
    """
    logger.info(f"Discovering resources in {root}")
    resources: list[DatasetResource] = []
    supported = {ext for ext in config.plugin_registry}
    adapt_dir = root / ".adapt"

    for path in root.rglob("*"):
        if path.is_dir() or should_ignore(path):
            continue

        ext = path.suffix.lower()
        if ext not in supported:
            continue

        logger.debug(f"Processing file: {path}")
        plugin_cls = config.get_plugin_factory(ext)
        plugin: Plugin = plugin_cls()

        if not plugin.detect(path):
            logger.debug(
                "Plugin %s rejected file during detection: %s",
                plugin_cls.__name__,
                path,
            )
            continue

        loaded = plugin.load(path)
        if isinstance(loaded, ResourceDescriptor):
            descriptors = [loaded]
        else:
            descriptors = loaded

        for descriptor in descriptors:
            sub_namespace = descriptor.metadata.get("sub_namespace", "")
            suffix = f".{sub_namespace}" if sub_namespace else ""
            base_path = adapt_dir / path.relative_to(root)
            schema_path = base_path.with_suffix(f"{suffix}.schema.json")
            ui_path = base_path.with_suffix(f"{suffix}.index.html")
            options_path = base_path.with_suffix(f"{suffix}.options.json")

            descriptor.schema_path = schema_path
            descriptor.ui_path = ui_path
            descriptor.options_path = options_path

            # Options must be applied before companion files are generated, so a
            # schema derived from an overridden header row is written out correctly.
            descriptor.metadata["options"] = read_resource_options(options_path)
            plugin.apply_options(descriptor)

            plugin.generate_companion_files(descriptor)

            resource = DatasetResource(
                path=path,
                relative_path=path.relative_to(root),
                resource_type=descriptor.resource_type,
                schema_path=schema_path,
                ui_path=ui_path,
                options_path=options_path,
                plugin_name=plugin_cls.__name__,
                metadata=descriptor.metadata,
            )
            resources.append(resource)

    logger.info(f"Discovered {len(resources)} resources")
    return resources

def build_mcp_server(config: AdaptConfig) -> FastMCP:
    """Construct the FastMCP server and register its tools.

    `streamable_http_path` is set to `/` because the outer app already mounts
    this server's ASGI app at `/mcp` (`app.mount("/mcp", mcp_app)`); leaving
    the SDK's own default of `/mcp` here would double up to `/mcp/mcp`.
    """
    mcp = FastMCP(
        name="adapt",
        instructions=(
            "Adapt exposes file-backed datasets, documents, and media as "
            "permission-filtered tools. Call list_resources first to see what "
            "you can read; search works across everything you're permitted to see."
        ),
        streamable_http_path="/",
    )

    @mcp.tool()
    async def list_resources(ctx: Context) -> dict:
        """List every resource namespace the caller may read, with its type."""
        request = ctx.request_context.request
        user = await _authenticated_user(ctx)
        registry: dict[str, ResourceRegistryEntry] = request.app.state.resource_registry

        if getattr(user, "is_superuser", False):
            readable = None
        else:
            with Session(request.app.state.db_engine) as db:
                readable = PermissionChecker(db).readable_resources(user)

        resources = [
            {"resource": namespace, "type": entry.resource.resource_type}
            for namespace, entry in registry.items()
            if readable is None or namespace in readable
        ]
        return {"resources": resources}

    @mcp.tool()
    async def get_schema(resource: str, ctx: Context) -> dict:
        """Return the schema for a dataset resource (columns and types)."""
        request = ctx.request_context.request
        user = await _authenticated_user(ctx)
        entry = _authorized_entry(request, user, resource, "read")
        return entry.plugin.schema(entry.descriptor)

    @mcp.tool()
    async def read_resource(
        resource: str,
        ctx: Context,
        limit: int | None = None,
        offset: int = 0,
        sort: SortParam = None,
        order: OrderParam = "asc",
        filter: FilterParam = None,
    ) -> Any:
        """Read a resource's content, permission-checked like the REST API.

        For dataset resources, `sort` is the column name and `order` must be
        `asc` or `desc`.
        """
        request = ctx.request_context.request
        user = await _authenticated_user(ctx)
        entry = _authorized_entry(request, user, resource, "read")
        request.state.user = user

        if entry.resource.resource_type in DATASET_TYPES:
            query_params = QueryParams(limit=limit, offset=offset, sort=sort, order=order, filter=filter)
            return entry.plugin.read(entry.descriptor, request, query_params)
        return entry.plugin.read(entry.descriptor, request)

    @mcp.tool()
    async def write_resource(
        resource: str,
        action: Literal["create", "update", "delete"],
        data: Any,
        ctx: Context,
    ) -> dict:
        """Create, update, or delete rows in a writable (dataset) resource."""
        request = ctx.request_context.request
        user = await _authenticated_user(ctx)
        entry = _authorized_entry(request, user, resource, "write")

        if request.app.state.config.readonly:
            raise ToolError("Server is in read-only mode")

        request.state.user = user
        context = PluginContext(
            engine=request.app.state.db_engine,
            root=request.app.state.config.root,
            readonly=request.app.state.config.readonly,
            lock_manager=request.app.state.lock_manager,
        )
        try:
            return entry.plugin.write(entry.descriptor, {"action": action, "data": data}, request, context)
        except NotImplementedError:
            raise ToolError(f"Resource type {entry.resource.resource_type!r} does not support write operations")
        except HTTPException as exc:
            raise ToolError(str(exc.detail))

    @mcp.tool()
    async def search(
        q: str,
        ctx: Context,
        limit: int = 20,
        offset: int = 0,
        resource_type: str | None = None,
    ) -> dict:
        """Full-text search across every resource the caller may read."""
        request = ctx.request_context.request
        user = await _authenticated_user(ctx)
        return _run_search(request, q, limit, offset, resource_type, user)

    return mcp