mobiledev/taskflow

TaskFlow, a team task board you run from the web or from Claude and ChatGPT. FastAPI + DataK3 + appid + Models + MCP.

• dodil-challenge
• dodil-challenge-boards

A remix template · 0 remixes · Remix mobiledev/taskflow

app/taskflow/main.py

"""Wires the layers together: one FastAPI app with the web routes, /mcp, and the sign-in
metadata Claude.ai and ChatGPT use to find where a person signs in (RFC 9728)."""
import json
import os
from urllib.parse import urlsplit
from contextlib import asynccontextmanager

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse, StreamingResponse
from mcp.server.transport_security import TransportSecuritySettings
from starlette.routing import Route

from taskflow.application.use_cases import TaskFlow
from taskflow.doors.identity import person_from
from taskflow.doors.mcp import build_mcp
from taskflow.doors.web import build_router

APPID = os.environ.get("APPID_PUBLIC_URL", "https://appid.dodil.io").rstrip("/")
POOL = os.environ.get("DODIL_USER_POOL", "")  # "<org>/<pool>", set by .dodil/deploy.yaml
METADATA = "/.well-known/oauth-protected-resource"
WRITES = {"POST", "PUT", "PATCH", "DELETE"}


def create_app(app_service: TaskFlow, self_check=None) -> FastAPI:
    """`self_check()` returns row counts per table; `dodil ignite invoke` calls POST / (with no
    person) to prove the deployed app reaches its own bucket with its own credential. It returns
    counts only, never rows."""
    gate = {"mcp": None}

    @asynccontextmanager
    async def lifespan(_app):
        mcp_app = build_mcp(app_service).streamable_http_app(
            streamable_http_path="/mcp", stateless_http=True, json_response=True,
            # Every call carries identity only the gateway can mint, so host checks add nothing.
            transport_security=TransportSecuritySettings(enable_dns_rebinding_protection=False))
        gate["mcp"] = mcp_app
        async with mcp_app.router.lifespan_context(mcp_app):
            yield

    app = FastAPI(title="TaskFlow", lifespan=lifespan)

    def base(request: Request) -> str:
        h = request.headers
        return f"{h.get('x-forwarded-proto', 'https')}://{h.get('x-forwarded-host', h.get('host'))}"

    @app.middleware("http")
    async def same_origin_writes(request: Request, call_next):
        """The page signs in with a cookie, so a change must come from the page itself. Every
        Ignite app shares *.ignite.dodil.cloud, so "same-site" is not enough: another org's app
        is same-site. Browsers say where a request comes from (Sec-Fetch-Site, Origin); a write
        from anywhere but this origin is refused. Clients that send neither (Claude, scripts)
        authenticate with a bearer token, not the cookie, and pass."""
        if request.method in WRITES and request.url.path.startswith("/api/"):
            site = request.headers.get("sec-fetch-site")
            origin = request.headers.get("origin")
            own_host = request.headers.get("x-forwarded-host", request.headers.get("host"))
            if (site and site not in ("same-origin", "none")) or (origin and urlsplit(origin).netloc != own_host):
                return JSONResponse({"detail": "cross-site request refused"}, status_code=403)
        return await call_next(request)

    @app.get("/healthz", include_in_schema=False)
    def healthz():
        return {"ok": True}

    @app.post("/", include_in_schema=False)
    def check():
        return {"ok": True, "tables": self_check() if self_check else {}}

    def resource(request: Request) -> dict:
        return {"resource": base(request) + "/mcp", "authorization_servers": [f"{APPID}/{POOL}"],
                "bearer_methods_supported": ["header"], "resource_name": "TaskFlow"}

    @app.get(METADATA, include_in_schema=False)
    def metadata(request: Request):
        return resource(request)

    @app.get(METADATA + "/mcp", include_in_schema=False)
    def metadata_mcp(request: Request):
        # The Ignite gateway relabels any reply under a path ending in /mcp as chunked without
        # re-encoding it, which breaks a reply that carries a Content-Length. Streamed, the app
        # chunks it itself and it arrives intact. (The 401 below points at METADATA instead.)
        body = json.dumps(resource(request)).encode()
        return StreamingResponse(iter([body]), media_type="application/json")

    class McpDoor:
        """No person: 401 with the metadata link, which is what starts Claude's sign-in.
        (A class, so Starlette hands it the raw ASGI call instead of a Request.)"""

        async def __call__(self, scope, receive, send):
            headers = {k.decode(): v.decode() for k, v in scope.get("headers", [])}
            if person_from(headers) is None:
                body = json.dumps({"detail": "sign in first"}).encode()
                link = f'Bearer resource_metadata="{base(Request(scope))}{METADATA}"'
                await send({"type": "http.response.start", "status": 401, "headers": [
                    (b"content-type", b"application/json"), (b"www-authenticate", link.encode())]})
                await send({"type": "http.response.body", "body": body})
                return
            await gate["mcp"]({**scope, "path": "/mcp"}, receive, send)

    mcp_door = McpDoor()
    for path in ("/mcp", "/mcp/"):
        app.router.routes.append(Route(path, endpoint=mcp_door, methods=["GET", "POST", "DELETE"]))
    app.include_router(build_router(app_service))
    return app


def production_app() -> FastAPI:
    # Imported here so tests never need Dodil credentials or a database driver.
    from taskflow.infrastructure.dodil import ModelsAssistant
    from taskflow.infrastructure.repository import SqlBoards
    repo = SqlBoards()
    return create_app(TaskFlow(repo, ModelsAssistant()), self_check=repo.table_counts)