dmslanes/lanes
Lanes - team task boards with roles, history and an MCP endpoint for Claude and ChatGPT
A remix template · 0 remixes · Remix dmslanes/lanes
app/main.py
"""Lanes: one FastAPI app, two doors (web routes and /mcp) over one set of use cases.
Run locally: DEV_ALLOW_ANON=1 LANES_STORAGE=memory uvicorn app.main:app --reload
Deployed: K3_BUCKET + DODIL_SERVICE_ACCOUNT_ID/SECRET from .dodil/deploy.yaml
"""
from __future__ import annotations
import json
import os
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
from fastapi.staticfiles import StaticFiles
from mcp.server.transport_security import TransportSecuritySettings
from starlette.routing import Route
from app.application.use_cases import BoardService
from app.domain.model import DomainError
from app.infrastructure.identity import Unauthenticated, dev_mode, email_from_headers
from app.mcp.tools import build_mcp
from app.web.routes import STATIC, build_router
STATUS = {"not_found": 404, "forbidden": 403, "conflict": 409, "invalid": 422}
MCP_PATH = "/mcp"
METADATA_PATH = "/.well-known/oauth-protected-resource"
def make_repo():
if os.environ.get("LANES_STORAGE") == "memory":
from app.infrastructure.memory import MemoryRepository
return MemoryRepository()
from app.infrastructure.k3 import K3Repository
return K3Repository()
def make_ai():
if os.environ.get("LANES_STORAGE") == "memory" and not os.environ.get("DODIL_TOKEN"):
return None
from app.infrastructure.models import DodilModels
return DodilModels()
def _public_base(scope) -> str:
explicit = os.environ.get("PUBLIC_BASE_URL", "").rstrip("/")
if explicit:
return explicit
h = {k.decode().lower(): v.decode() for k, v in scope.get("headers", [])}
proto = h.get("x-forwarded-proto", scope.get("scheme", "http")).split(",")[0].strip()
return f"{proto}://{h.get('x-forwarded-host', h.get('host', 'localhost')).split(',')[0].strip()}"
def _pool_issuer() -> str | None:
"""Where a person signs in: the app's appid pool (DODIL_USER_POOL = <org>/<pool>)."""
if os.environ.get("AUTH_ISSUER"):
return os.environ["AUTH_ISSUER"].rstrip("/")
pool = os.environ.get("DODIL_USER_POOL", "").strip("/")
return f"{os.environ.get('APPID_PUBLIC_URL', 'https://appid.dodil.io').rstrip('/')}/{pool}" if pool.count("/") == 1 else None
class SignInGate:
"""No person -> HTTP 401 with the RFC 9728 pointer, which is what makes Claude/ChatGPT start sign-in."""
def __init__(self):
self.mcp_app = None
async def __call__(self, scope, receive, send):
if scope["type"] == "http":
headers = {k.decode(): v.decode() for k, v in scope.get("headers", [])}
try:
email_from_headers(headers)
except Unauthenticated as e:
challenge = "Bearer"
if _pool_issuer():
challenge += f' resource_metadata="{_public_base(scope)}{METADATA_PATH}{MCP_PATH}"'
body = json.dumps({"detail": str(e)}).encode()
await send({"type": "http.response.start", "status": 401,
"headers": [(b"content-type", b"application/json"),
(b"www-authenticate", challenge.encode()),
(b"content-length", str(len(body)).encode())]})
await send({"type": "http.response.body", "body": body})
return
scope = {**scope, "path": MCP_PATH}
await self.mcp_app(scope, receive, send)
def _transport_security() -> TransportSecuritySettings:
# Behind the gateway the Host is the app's public domain; every call carries identity only the
# gateway mints, so DNS-rebinding protection adds nothing there. Locally, keep it on.
if dev_mode():
return TransportSecuritySettings(enable_dns_rebinding_protection=True,
allowed_hosts=["127.0.0.1:*", "localhost:*", "localhost", "127.0.0.1"],
allowed_origins=["http://127.0.0.1:*", "http://localhost:*"])
return TransportSecuritySettings(enable_dns_rebinding_protection=False)
def create_app(svc: BoardService | None = None) -> FastAPI:
svc = svc or BoardService(make_repo(), ai=make_ai())
gate = SignInGate()
@asynccontextmanager
async def lifespan(_app):
mcp_app = build_mcp(svc).streamable_http_app(
streamable_http_path=MCP_PATH, stateless_http=True, json_response=True,
transport_security=_transport_security())
gate.mcp_app = mcp_app
async with mcp_app.router.lifespan_context(mcp_app):
yield
app = FastAPI(title="Lanes", version="1.0.0", lifespan=lifespan)
app.state.svc = svc
@app.exception_handler(DomainError)
async def domain_error(_req, exc: DomainError):
return JSONResponse({"detail": str(exc)}, status_code=STATUS.get(exc.code, 422))
@app.exception_handler(Unauthenticated)
async def unauthenticated(_req, exc):
return JSONResponse({"detail": str(exc)}, status_code=401)
@app.get("/healthz", include_in_schema=False)
def healthz():
return {"ok": True}
@app.post("/", include_in_schema=False)
def self_check():
"""`dodil ignite invoke` target: proves the app reaches its own bucket with its own credential."""
check = getattr(svc.repo, "self_check", None)
return {"ok": True, "rows": check() if check else "memory storage"}
def metadata(request: Request):
issuer = _pool_issuer()
if not issuer:
return JSONResponse({"detail": "no user pool configured (DODIL_USER_POOL)"}, status_code=404)
return JSONResponse({"resource": f"{_public_base(request.scope)}{MCP_PATH}",
"authorization_servers": [issuer], "bearer_methods_supported": ["header"],
"resource_name": "Lanes"})
for path in (METADATA_PATH + MCP_PATH, METADATA_PATH):
app.add_api_route(path, metadata, methods=["GET"], include_in_schema=False)
for path in (MCP_PATH, MCP_PATH + "/"):
app.router.routes.append(Route(path, endpoint=gate, methods=["GET", "POST", "DELETE"]))
app.include_router(build_router(svc))
app.mount("/static", StaticFiles(directory=STATIC), name="static")
return app
app = create_app() if os.environ.get("LANES_NO_AUTOAPP") != "1" else None