devnet: FastAPI app, routes, health

src/devnet/main.py

@@ -0,0 +1,120 @@
+"""devnet - one interface to the places builders use.
+
+REST under /v1, MCP at /mcp, health at /health.
+"""
+from __future__ import annotations
+
+import os
+import secrets
+
+from fastapi import Depends, FastAPI, Header, HTTPException
+
+from .core import AuthRequired, RateLimited, UpstreamError
+from .sources import github, hackernews, pypi, rss
+
+app = FastAPI(title="devnet", version="0.1.0", summary="Developer internet for AI agents.")
+
+# ponytail: one shared key from env, no accounts and no database. Ceiling: cannot
+# rotate per customer or attribute usage to one. Upgrade: DB-backed keys at billing.
+_KEYS = tuple(k for k in os.environ.get("DEVNET_KEYS", "").replace(",", " ").split() if k)
+_OPEN = os.environ.get("DEVNET_OPEN") == "1"
+
+
+def require_key(x_key: str = Header(default="")) -> None:
+    if _OPEN or not _KEYS:
+        return
+    if not any(secrets.compare_digest(x_key, k) for k in _KEYS):
+        raise HTTPException(401, "invalid or missing X-Key header")
+
+
+key = Depends(require_key)
+
+
+def _run(fn, *args):
+    """Map the four upstream states onto HTTP status codes."""
+    try:
+        return fn(*args)
+    except RateLimited as exc:
+        raise HTTPException(429, str(exc)) from exc
+    except AuthRequired as exc:
+        raise HTTPException(401, str(exc)) from exc
+    except UpstreamError as exc:
+        raise HTTPException(502, str(exc)) from exc
+
+
+@app.get("/", include_in_schema=False)
+def root() -> dict:
+    return {
+        "service": "devnet",
+        "mcp": "/mcp",
+        "rest": [
+            "/v1/pypi/package",
+            "/v1/github/repo",
+            "/v1/github/issues",
+            "/v1/hackernews/search",
+            "/v1/rss/read",
+        ],
+        "health": "/health",
+        "docs": "/docs",
+        "auth": "open" if (_OPEN or not _KEYS) else "X-Key header",
+    }
+
+
+@app.get("/health")
+def health() -> dict:
+    """One cheap cached probe per source. Four states: ok/error/rate_limited/auth_required."""
+    probes = {
+        "pypi": lambda: pypi.package("requests"),
+        "github": lambda: github.repository("psf", "requests"),
+        "hackernews": lambda: hackernews.search("python", limit=1),
+        "rss": lambda: rss.read("https://huggingface.co/blog/feed.xml", limit=1),
+    }
+    checks: dict[str, dict] = {}
+    for name, probe in probes.items():
+        try:
+            probe()
+            checks[name] = {"status": "ok"}
+        except UpstreamError as exc:
+            checks[name] = {"status": exc.kind, "detail": str(exc)[:200]}
+        except Exception as exc:  # health must never crash
+            checks[name] = {"status": "error", "detail": f"{type(exc).__name__}: {exc}"[:200]}
+
+    worst = "ok"
+    for candidate in ("error", "rate_limited", "auth_required"):
+        if any(c["status"] == candidate for c in checks.values()):
+            worst = candidate
+            break
+    return {"status": worst, "checks": checks}
+
+
+@app.get("/v1/pypi/package", dependencies=[key])
+def pypi_package(name: str) -> dict:
+    return _run(pypi.package, name)
+
+
+@app.get("/v1/github/repo", dependencies=[key])
+def github_repo(owner: str, name: str) -> dict:
+    return _run(github.repository, owner, name)
+
+
+@app.get("/v1/github/issues", dependencies=[key])
+def github_issues(owner: str, name: str, state: str = "open", limit: int = 20) -> dict:
+    return _run(github.issues, owner, name, state, limit)
+
+
+@app.get("/v1/hackernews/search", dependencies=[key])
+def hn_search(query: str, limit: int = 20, kind: str = "all") -> dict:
+    return _run(hackernews.search, query, limit, kind)
+
+
+@app.get("/v1/rss/read", dependencies=[key])
+def rss_read(url: str, limit: int = 20) -> dict:
+    return _run(rss.read, url, limit)
+
+
+try:
+    from .mcp_server import mcp_app
+
+    app.mount("/mcp", mcp_app)
+except ImportError:  # REST still serves if the mcp package is unavailable
+    pass