Run the Server Over HTTP

# Was: mcp = FastMCP("compliance-toolkit")
mcp = FastMCP("compliance-toolkit", stateless_http=True)

if __name__ == "__main__":
    import sys
    transport = sys.argv[1] if len(sys.argv) > 1 else "stdio"
    if transport == "http":
        mcp.run(transport="streamable-http")
    else:
        mcp.run()  # stdio (default)

# stdio mode (what Claude Desktop uses):
python server.py

# HTTP mode (long-lived service):
python server.py http

npx @modelcontextprotocol/inspector

pip install pyjwt

"""Mint a JWT for testing the compliance-toolkit HTTP server.

In production this lives in a server-side mint endpoint with proper user auth,
short TTLs, and scoping. For learning, we'll mint a 1-hour token for a fake user.
"""

import jwt
from datetime import datetime, timedelta, timezone

SECRET = "dev-secret-do-not-use-in-production"  # in prod: env var, KMS, etc.

def mint(user_id: str, allowed_tools: list[str], ttl_minutes: int = 60) -> str:
    now = datetime.now(timezone.utc)
    payload = {
        "sub": user_id,
        "iat": int(now.timestamp()),
        "exp": int((now + timedelta(minutes=ttl_minutes)).timestamp()),
        "scope": {"tools": allowed_tools},
        "purpose": "compliance-toolkit-access",
    }
    return jwt.encode(payload, SECRET, algorithm="HS256")


if __name__ == "__main__":
    # Issue a token that allows all tools
    token = mint(
        user_id="analyst.demo@example.com",
        allowed_tools=["lookup_sanctions_hit", "check_jurisdiction_risk", "summarize_alert"],
    )
    print(token)

python mint_token.py
# eyJhbGciOiJIUzI1NiIs... (copy this)

import jwt
from contextvars import ContextVar
from functools import wraps
from mcp.server.fastmcp import Context

SECRET = "dev-secret-do-not-use-in-production"

# Threaded through the request via ContextVar so tool signatures stay clean.
# FastMCP rejects tool parameters that start with '_', so we can't pass the
# auth identity as a function kwarg. Tools read it via current_auth_user().
_current_auth_user: ContextVar[str] = ContextVar("_current_auth_user", default="anonymous")


def current_auth_user() -> str:
    """Return the authenticated subject ('sub' claim) for the in-flight tool call."""
    return _current_auth_user.get()


def require_auth(tool_name: str):
    """Decorator: validate a Bearer JWT from the request headers and check scope."""
    def deco(fn):
        @wraps(fn)
        def wrapper(*args, ctx: Context = None, **kwargs):
            # FastMCP makes the request available via ctx.request_context
            headers = {}
            if ctx and hasattr(ctx, "request_context"):
                req = ctx.request_context.request
                if req:
                    headers = dict(req.headers)

            auth = headers.get("authorization", "")
            if not auth.startswith("Bearer "):
                return {"error": "missing or malformed Authorization header", "isError": True}

            token = auth.removeprefix("Bearer ").strip()
            try:
                payload = jwt.decode(token, SECRET, algorithms=["HS256"])
            except jwt.ExpiredSignatureError:
                return {"error": "token expired", "isError": True}
            except jwt.InvalidTokenError as e:
                return {"error": f"invalid token: {e}", "isError": True}

            scope_tools = payload.get("scope", {}).get("tools", [])
            if tool_name not in scope_tools:
                return {
                    "error": f"token does not authorize tool '{tool_name}'",
                    "authorized": scope_tools,
                    "isError": True,
                }

            token_ref = _current_auth_user.set(payload["sub"])
            try:
                return fn(*args, **kwargs)
            finally:
                _current_auth_user.reset(token_ref)
        return wrapper
    return deco

# Was: @mcp.tool()
# Was: def lookup_sanctions_hit(name: str) -> dict:
def _do_screening(name: str) -> dict:
    """Existing screening logic — cache, OpenSanctions API call, fallback.
    No changes to the body; just renamed and the @mcp.tool() decorator removed.
    """
    cached = _cache_get(name)
    if cached is not None:
        return {**cached, "from_cache": True}
    # ... rest of your existing cache + API + fallback logic unchanged ...
    # (the body you wrote in the real-API stretch)

@mcp.tool()
@require_auth("lookup_sanctions_hit")
def lookup_sanctions_hit(name: str, ctx: Context = None) -> dict:
    """Screen a person or entity name against the live OpenSanctions database.

    Requires a Bearer JWT in the Authorization header with 'lookup_sanctions_hit'
    in its tool scope. The caller's identity is recorded on every response.
    """
    result = _do_screening(name)
    result["screened_by"] = current_auth_user()  # audit trail!
    return result

# Replace TOKEN with the JWT you minted
TOKEN="eyJhbGc..."

# Should succeed with that token's scope:
curl -X POST http://localhost:8000/mcp \
  -H "Authorization: Bearer $TOKEN" \
  -H "Accept: application/json, text/event-stream" \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"lookup_sanctions_hit","arguments":{"name":"Vladimir Petrov"}}}'

# Should fail with no token (401):
curl -X POST http://localhost:8000/mcp \
  -H "Accept: application/json, text/event-stream" \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"lookup_sanctions_hit","arguments":{"name":"Vladimir Petrov"}}}'