Quick start

[project]
name = "team-app"
version = "0.1.0"
description = "A small example of a team's app for Theorem: an MCP server that trusts Theorem's token."
requires-python = ">=3.11"
dependencies = [
    "mcp>=1.20,<2",
    "pyjwt[crypto]>=2.10",
    "uvicorn>=0.34",
]

[dependency-groups]
dev = ["pytest>=8", "httpx>=0.28"]

[tool.uv]
package = false

[tool.pytest.ini_options]
testpaths = ["."]

"""A small example of a team's app for Theorem: an MCP server that trusts Theorem's token.

Every request must carry `Authorization: Bearer <token>`. The token is a JWT signed by
Theorem (EdDSA). This server checks its signature against Theorem's public keys, that
Theorem issued it, and that it was made for this app's ID. It then knows who is calling
and which of the team's roles they hold, and decides for itself what each role may do.

Settings, from the environment:
  APP_ID      this app's ID at Theorem (`app_...`), shown where the app was added
  APP_HOST    the host name people reach this server by
  APP_PORT    the port to listen on, on this machine only (default 8787)
  THEOREM     Theorem's address (default https://theorem.sa)
  THEOREM_KEYS_FILE  Theorem's public keys saved to a file, for a server that cannot reach Theorem
"""

import contextvars
import os

import jwt
import uvicorn
from mcp.server.fastmcp import FastMCP
from mcp.server.transport_security import TransportSecuritySettings
from mcp.types import ToolAnnotations
from starlette.responses import JSONResponse

APP_ID = os.environ["APP_ID"]
APP_HOST = os.environ["APP_HOST"]
PORT = int(os.environ.get("APP_PORT", "8787"))
THEOREM = os.environ.get("THEOREM", "https://theorem.sa")

keys = jwt.PyJWKClient(f"{THEOREM}/api/identity/jwks")
# For a server that cannot reach Theorem: the same keys, saved to a file.
KEYS_FILE = os.environ.get("THEOREM_KEYS_FILE")
caller = contextvars.ContextVar("caller")

HOURS = {}


def checked(token):
    """The token's claims, or an error: Theorem's signature, issuer, this app, not ended."""
    if KEYS_FILE:
        kid = jwt.get_unverified_header(token)["kid"]
        key = jwt.PyJWKSet.from_json(open(KEYS_FILE).read())[kid]
    else:
        key = keys.get_signing_key_from_jwt(token)
    return jwt.decode(
        token,
        key.key,
        algorithms=["EdDSA"],
        audience=APP_ID,
        issuer=THEOREM,
        options={"require": ["exp", "iss", "aud", "sub"]},
    )


class TheoremToken:
    """Lets a request in only with a token Theorem made for this app."""

    def __init__(self, app):
        self.app = app

    async def __call__(self, scope, receive, send):
        if scope["type"] == "lifespan":
            return await self.app(scope, receive, send)
        if scope["type"] != "http":
            return  # this app serves HTTP only; anything else would get past the check
        header = dict(scope["headers"]).get(b"authorization", b"").decode()
        try:
            kind, _, token = header.partition(" ")
            if kind.lower() != "bearer" or not token:
                raise jwt.InvalidTokenError("no bearer token")
            claims = checked(token)
        except Exception as error:  # every failure is the same answer: not let in
            print(f"refused: {type(error).__name__}: {error}", flush=True)
            response = JSONResponse({"error": "A token from Theorem for this app is needed."}, status_code=401,
                                    headers={"WWW-Authenticate": "Bearer"})
            return await response(scope, receive, send)
        marker = caller.set(claims)
        try:
            print(f"let in: {claims['email']} roles={[r['name'] for r in claims['team']['roles']]} "
                  f"leader={claims['team']['leader']}", flush=True)
            await self.app(scope, receive, send)
        finally:
            caller.reset(marker)


mcp = FastMCP(
    "Attendance",
    stateless_http=True,
    json_response=True,
    transport_security=TransportSecuritySettings(
        enable_dns_rebinding_protection=True, allowed_hosts=[APP_HOST], allowed_origins=[f"https://{APP_HOST}"]
    ),
)


@mcp.tool(title="My hours", annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=False))
def my_hours() -> str:
    """The hours the person has clocked in this week."""
    me = caller.get()
    return f"{me.get('name') or me['email']} has {HOURS.get(me['sub'], 0)} hours this week."


@mcp.tool(title="Clock in", annotations=ToolAnnotations(readOnlyHint=False, destructiveHint=False, openWorldHint=True))
def clock_in(hours: int) -> str:
    """Adds hours worked today for the person."""
    me = caller.get()
    HOURS[me["sub"]] = HOURS.get(me["sub"], 0) + hours
    return f"Added {hours} hours. The week's total is {HOURS[me['sub']]}."


@mcp.tool(title="Team hours", annotations=ToolAnnotations(readOnlyHint=True, openWorldHint=False))
def team_hours() -> str:
    """Everyone's hours this week. Only for the team's leaders and the Finance role."""
    team = caller.get()["team"]
    if not team["leader"] and "Finance" not in [role["name"] for role in team["roles"]]:
        return "Only the team's leaders and people in Finance may see everyone's hours."
    return f"{len(HOURS)} people have clocked in, {sum(HOURS.values())} hours in all, in the team {team['name']}."


if __name__ == "__main__":
    uvicorn.run(TheoremToken(mcp.streamable_http_app()), host="127.0.0.1", port=PORT, log_level="warning")