akif/arclet-boards
Arclet Boards — private Kanban workspaces with AppID, DataK3, shared web and MCP actions.
A remix template · 1 remixes · Remix akif/arclet-boards
app/main.py
"""One same-origin FastAPI application, one genuine MCP endpoint, one use-case layer."""
from contextlib import asynccontextmanager
import json
import os
import re
from urllib.parse import urlparse
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, RedirectResponse
from fastapi.staticfiles import StaticFiles
from mcp.server.transport_security import TransportSecuritySettings
from starlette.routing import Route
from app.application.boards import Boards
from app.domain.model import AppError, Unavailable
from app.infrastructure.dodil import DataK3Store, ModelsAssistant
from app.infrastructure.identity import person_from
from app.mcp.server import build_mcp
from app.web.routes import STATIC, build_router
METADATA = "/.well-known/oauth-protected-resource"
def create_app(boards=None):
boards = boards or Boards(DataK3Store(), ModelsAssistant())
gateway = {}
@asynccontextmanager
async def lifespan(application):
server = build_mcp(boards)
transport = server.streamable_http_app(streamable_http_path="/mcp", stateless_http=True,
json_response=True, transport_security=TransportSecuritySettings(
enable_dns_rebinding_protection=False))
gateway["mcp"] = transport
async with transport.router.lifespan_context(transport):
yield
application = FastAPI(title="Arclet Boards", version="1.0.0", lifespan=lifespan,
docs_url=None, redoc_url=None, openapi_url=None)
application.state.boards = boards
def base(request):
configured = os.environ.get("PUBLIC_URL", "").rstrip("/")
# Ignite forwards to an internal host. Preserve the starter's public
# forwarding-header behavior for OAuth resource URLs and browser Origin.
host = request.headers.get("x-forwarded-host", request.headers.get("host", "localhost"))
if not re.fullmatch(r"[A-Za-z0-9.-]+(?::[0-9]+)?", host):
raise Unavailable("Invalid gateway host")
return configured or f"https://{host}"
@application.exception_handler(AppError)
async def known_error(request, error):
return JSONResponse({"detail": str(error)}, status_code=error.status)
@application.middleware("http")
async def boundary(request, next_handler):
path = request.url.path
person = person_from(request.headers)
request.state.person = person
discovery = path in (METADATA, METADATA + "/mcp")
platform_check = path == "/" and request.method == "POST" and request.headers.get(
"x-dodil-auth-source") == "platform"
if not person and path != "/healthz" and not discovery and not platform_check:
if path == "/":
response = RedirectResponse("/.dodil/auth/login?return_to=/", status_code=303)
else:
headers = {}
if path.rstrip("/") == "/mcp":
headers["WWW-Authenticate"] = f'Bearer resource_metadata="{base(request)}{METADATA}/mcp"'
response = JSONResponse({"detail": "Sign in through AppID first"}, status_code=401,
headers=headers)
elif request.method in ("POST", "PUT", "PATCH", "DELETE") and path.startswith("/api/"):
origin = request.headers.get("origin")
if request.headers.get("sec-fetch-site") == "cross-site" or (
origin and urlparse(origin).netloc != urlparse(base(request)).netloc) or (
request.headers.get("cookie") and not origin):
response = JSONResponse({"detail": "Cross-origin write refused"}, status_code=403)
elif not request.headers.get("content-type", "").lower().startswith("application/json"):
response = JSONResponse({"detail": "Use application/json"}, status_code=415)
else:
response = await next_handler(request)
else:
response = await next_handler(request)
response.headers.update({
"X-Content-Type-Options": "nosniff", "Referrer-Policy": "same-origin",
"X-Frame-Options": "DENY", "Cache-Control": "no-store",
"Content-Security-Policy": "default-src 'self'; script-src 'self'; style-src 'self'; "
"img-src 'self' data:; connect-src 'self'; object-src 'none'; base-uri 'none'; "
"form-action 'self'; frame-ancestors 'none'",
})
return response
@application.get("/healthz")
def healthz():
return {"ok": True, "service": "Arclet Boards"}
@application.post("/")
def self_check(request: Request):
# A platform principal may prove the app reaches its own database; it
# cannot use any end-user action or see row contents through this route.
if request.headers.get("x-dodil-auth-source") != "platform":
return JSONResponse({"detail": "Platform diagnostic only"}, status_code=403)
if not hasattr(boards.store, "self_check"):
raise Unavailable("Database diagnostic is unavailable")
return boards.store.self_check()
@application.get(METADATA)
@application.get(METADATA + "/mcp")
def metadata(request: Request):
pool = os.environ.get("DODIL_USER_POOL", "")
if not pool:
raise Unavailable("AppID pool is not configured")
return {"resource": base(request) + "/mcp",
"authorization_servers": ["https://appid.dodil.io/" + pool],
"bearer_methods_supported": ["header"], "resource_name": "Arclet Boards"}
class McpDoor:
async def __call__(self, scope, receive, send):
await gateway["mcp"]({**scope, "path": "/mcp"}, receive, send)
for path in ("/mcp", "/mcp/"):
application.router.routes.append(Route(path, endpoint=McpDoor(), methods=["GET", "POST", "DELETE"]))
application.include_router(build_router(boards))
application.mount("/static", StaticFiles(directory=STATIC), name="static")
return application
app = create_app()