Build & publish an MCP app

uvx fastapi-new myapp
cd myapp
uv add fastapi-mcp httpx pyjwt

"""Save as app/mcp_store_auth.py"""

from __future__ import annotations

import os
from dataclasses import dataclass
from functools import lru_cache
from uuid import UUID

import httpx
import jwt
from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from jwt import PyJWKClient

_bearer = HTTPBearer(auto_error=False)

APP_SLUG = os.environ["APP_SLUG"]
STORE_URL = os.environ["STORE_URL"].rstrip("/")
SUPABASE_URL = os.environ["SUPABASE_URL"].rstrip("/")


@dataclass
class StoreUser:
    id: UUID
    email: str | None


@lru_cache
def _jwks_client() -> PyJWKClient:
    return PyJWKClient(f"{SUPABASE_URL}/auth/v1/.well-known/jwks.json")


def verify_supabase_jwt(token: str) -> StoreUser:
    try:
        signing_key = _jwks_client().get_signing_key_from_jwt(token)
        payload = jwt.decode(
            token,
            signing_key.key,
            algorithms=["ES256", "HS256"],
            audience="authenticated",
        )
    except jwt.PyJWTError as exc:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token.") from exc

    sub = payload.get("sub")
    if not sub:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token subject.")
    return StoreUser(id=UUID(sub), email=payload.get("email"))


async def verify_install(token: str) -> StoreUser:
    user = verify_supabase_jwt(token)
    async with httpx.AsyncClient(timeout=10.0) as client:
        response = await client.get(
            f"{STORE_URL}/api/v1/auth/verify",
            params={"app": APP_SLUG},
            headers={"Authorization": f"Bearer {token}"},
        )
    if response.status_code == status.HTTP_401_UNAUTHORIZED:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token.")
    if response.status_code != status.HTTP_200_OK:
        raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="Store verify unavailable.")

    data = response.json()
    if not data.get("authorized"):
        raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail=data.get("reason", "not_installed"))
    return user


async def require_installed_user(
    credentials: HTTPAuthorizationCredentials | None = Depends(_bearer),
) -> StoreUser:
    if credentials is None or credentials.scheme.lower() != "bearer":
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Missing bearer token.",
            headers={"WWW-Authenticate": "Bearer"},
        )
    return await verify_install(credentials.credentials)

"""Save as app/mcp_transport.py — required for FastAPI Cloud production."""

from __future__ import annotations

import asyncio
import logging

from fastapi import APIRouter, FastAPI
from fastapi_mcp.server import FastApiMCP
from fastapi_mcp.transport.http import FastApiHttpSessionManager
from mcp.server.streamable_http_manager import StreamableHTTPSessionManager

logger = logging.getLogger(__name__)


class StatelessFastApiHttpSessionManager(FastApiHttpSessionManager):
    async def _ensure_session_manager_started(self) -> None:
        if self._manager_started:
            return

        async with self._startup_lock:
            if self._manager_started:
                return

            logger.info("Starting stateless StreamableHTTP session manager")
            self._session_manager = StreamableHTTPSessionManager(
                app=self.mcp_server,
                event_store=self.event_store,
                json_response=self.json_response,
                stateless=True,
                security_settings=self.security_settings,
            )

            async def run_session_manager():
                try:
                    async with self._session_manager.run():
                        logger.info("Stateless StreamableHTTP session manager is running")
                        await asyncio.Event().wait()
                except asyncio.CancelledError:
                    logger.info("Stateless StreamableHTTP session manager is shutting down")
                    raise
                except Exception:
                    logger.exception("Error in StreamableHTTP session manager")
                    raise

            self._manager_task = asyncio.create_task(run_session_manager())
            self._manager_started = True
            await asyncio.sleep(0.1)


def mount_stateless_http(
    mcp: FastApiMCP,
    router: FastAPI | APIRouter | None = None,
    mount_path: str = "/mcp",
) -> None:
    if not mount_path.startswith("/"):
        mount_path = f"/{mount_path}"
    if mount_path.endswith("/"):
        mount_path = mount_path[:-1]

    if router is None:
        router = mcp.fastapi

    assert isinstance(router, (FastAPI, APIRouter))

    http_transport = StatelessFastApiHttpSessionManager(mcp_server=mcp.server)
    dependencies = mcp._auth_config.dependencies if mcp._auth_config else None

    mcp._register_mcp_endpoints_http(router, http_transport, mount_path, dependencies)
    mcp._setup_auth()
    mcp._http_transport = http_transport

    if isinstance(router, APIRouter):
        mcp.fastapi.include_router(router)

    logger.info(f"MCP stateless HTTP server listening at {mount_path}")

import os

from fastapi import Depends, FastAPI
from fastapi_mcp import AuthConfig, FastApiMCP

from app.mcp_store_auth import StoreUser, require_installed_user
from app.mcp_transport import mount_stateless_http

APP_SLUG = os.environ["APP_SLUG"]
STORE_URL = os.environ["STORE_URL"].rstrip("/")

app = FastAPI(title="My App", version="0.1.0")


@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok", "app": APP_SLUG}


@app.get("/my_tool", operation_id="my_tool", summary="My first tool")
async def my_tool(user: StoreUser = Depends(require_installed_user)) -> dict[str, str]:
    return {"message": f"Hello {user.email or user.id}"}


mcp = FastApiMCP(
    app,
    name="My App",
    exclude_operations=["health_health_get"],
    auth_config=AuthConfig(
        issuer=STORE_URL,
        authorize_url=f"{STORE_URL}/oauth/authorize?app={APP_SLUG}",
        oauth_metadata_url=f"{STORE_URL}/.well-known/oauth-authorization-server",
        client_id="mcp-store",
        client_secret="local-dev-secret",
        dependencies=[Depends(require_installed_user)],
        setup_proxies=True,
        setup_fake_dynamic_registration=True,
    ),
)
mount_stateless_http(mcp)  # → /mcp — do not use mcp.mount_http() in production

APP_SLUG=my-app
STORE_URL=http://localhost:8002
SUPABASE_URL=http://127.0.0.1:54321

APP_SLUG=my-app
STORE_URL=https://mcp-store.fastapicloud.dev
SUPABASE_URL=https://prttkffibedzpunjxuqv.supabase.co
APP_ENV=production

# Create .env.local in project root (see env vars section)
fastapi dev

# Health:  http://127.0.0.1:8000/health
# MCP:     http://127.0.0.1:8000/mcp

# Creates your FastAPI Cloud app on first run
fastapi deploy
# → https://myapp.fastapicloud.dev

# Set production env vars, then redeploy
fastapi cloud env set APP_SLUG my-app
fastapi cloud env set STORE_URL https://mcp-store.fastapicloud.dev
fastapi cloud env set SUPABASE_URL https://prttkffibedzpunjxuqv.supabase.co
fastapi cloud env set APP_ENV production

fastapi deploy

# Catalog mcp_url:
# https://myapp.fastapicloud.dev/mcp

---
name: my-app
description: When and how agents should use this MCP app. Be specific.
---

# My App

## When to use

Describe the scenarios where an agent should reach for this app.

## Tools

### my_tool

What it does, required inputs, and example prompts.

### get_config_url (optional)

If you ship a secure config console for Courier Scout (and similar hosts), document it here so clients can show the console chrome. Returns a short-lived `config_url` — never put the Store JWT in that URL.

## Setup

1. Install from the MCP Store catalog
2. Connect your MCP client to the app's MCP URL
3. Add this SKILL.md to your agent (Cursor skills folder)