commit f18e9ad96f13ce8008308e48176702131d845123 Author: Bastien Gourdon Date: Mon Jul 20 17:27:02 2026 +0200 feat: serveur MCP Forgejo (4 tools lecture) + demarrage auto diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..a2518d8 --- /dev/null +++ b/.gitignore @@ -0,0 +1,7 @@ +.env +venv/ +__pycache__/ +*.pyc +*.log +*.pid +oauth_state_forgejo.json diff --git a/mcp_forgejo_server.py b/mcp_forgejo_server.py new file mode 100644 index 0000000..4a3ae78 --- /dev/null +++ b/mcp_forgejo_server.py @@ -0,0 +1,547 @@ +#!/usr/bin/env python3 +""" +mcp_forgejo_server.py — Serveur MCP pur Starlette (compatible Python 3.9) +Acces en lecture aux depots Forgejo via le protocole MCP (JSON-RPC 2.0). + +Calque sur ~/App/Context_continuity/mcp_server.py : +transport Streamable HTTP sur /mcp, auth Bearer (tokens statiques + OAuth 2.1 +avec PKCE), etat OAuth isole dans oauth_state_forgejo.json. + +Usage : python3 mcp_forgejo_server.py -> ecoute sur 127.0.0.1:8770 + +Variables .env requises : + FORGEJO_URL (defaut http://localhost:3000) + FORGEJO_TOKEN token d'API Forgejo, permissions lecture repo + API_KEY_CLAUDE token statique Claude + API_KEY_LECHAT token statique Le Chat + OAUTH_BASE_URL defaut https://mcp-forgejo.bertha-cloud.fr + SECURITY_WORD mot de securite de la page /authorize +""" +import os +import json +import secrets +import hashlib +import base64 +import time +import requests +from dotenv import load_dotenv +from starlette.applications import Starlette +from starlette.routing import Route +from starlette.requests import Request +from starlette.responses import JSONResponse, Response, HTMLResponse, RedirectResponse + +load_dotenv() + +# --- Configuration Forgejo --- +FORGEJO_URL = os.getenv("FORGEJO_URL", "http://localhost:3000") +FORGEJO_TOKEN = os.getenv("FORGEJO_TOKEN", "") + +if not FORGEJO_TOKEN: + raise ValueError("FORGEJO_TOKEN doit etre defini (token d'API Forgejo, acces lecture).") + +# --- Auth MCP --- +API_KEY_CLAUDE = os.getenv("API_KEY_CLAUDE", "") +API_KEY_LECHAT = os.getenv("API_KEY_LECHAT", "") +VALID_TOKENS = {} +if API_KEY_CLAUDE: + VALID_TOKENS[API_KEY_CLAUDE] = "Claude" +if API_KEY_LECHAT: + VALID_TOKENS[API_KEY_LECHAT] = "LeChat" + +PROTOCOL_VERSION = "2025-06-18" +SERVER_INFO = {"name": "Forgejo Code Access", "version": "1.0.0"} + +# --- Config OAuth 2.1 (etat isole du serveur Trilium) --- +OAUTH_BASE_URL = os.getenv("OAUTH_BASE_URL", "https://mcp-forgejo.bertha-cloud.fr") +SECURITY_WORD = os.getenv("SECURITY_WORD", "") +OAUTH_STATE_FILE = os.path.expanduser("~/App/Code_versioning/oauth_state_forgejo.json") + + +def load_oauth_state(): + if not os.path.exists(OAUTH_STATE_FILE): + return {"clients": {}, "auth_codes": {}, "tokens": {}} + with open(OAUTH_STATE_FILE) as f: + state = json.load(f) + state.setdefault("clients", {}) + state.setdefault("auth_codes", {}) + state.setdefault("tokens", {}) + return state + + +def save_oauth_state(state): + with open(OAUTH_STATE_FILE, "w") as f: + json.dump(state, f, indent=2) + + +# --------------------------------------------------------------------------- +# Helper API Forgejo +# --------------------------------------------------------------------------- + +def forgejo_api(method, endpoint, **kwargs): + """Appel generique a l'API Forgejo.""" + url = "%s/%s" % (FORGEJO_URL.rstrip("/"), endpoint.lstrip("/")) + headers = {"Authorization": "token %s" % FORGEJO_TOKEN} + response = requests.request(method, url, headers=headers, timeout=30, **kwargs) + response.raise_for_status() + return response.json() if response.content else None + + +# --------------------------------------------------------------------------- +# Definition des tools (schema JSON) +# --------------------------------------------------------------------------- + +TOOLS = [ + { + "name": "list_repos", + "description": "Liste tous les depots Forgejo accessibles avec le token (nom complet, description, branche par defaut, visibilite).", + "inputSchema": { + "type": "object", + "properties": {}, + "required": [], + }, + }, + { + "name": "list_files", + "description": "Liste les fichiers et dossiers d un depot a un chemin donne. repo au format proprietaire/depot (ex: Master/sliding-automation). path vide = racine.", + "inputSchema": { + "type": "object", + "properties": { + "repo": {"type": "string", "description": "proprietaire/depot"}, + "path": {"type": "string", "description": "Chemin dans le depot (defaut: racine)"}, + "ref": {"type": "string", "description": "Branche, tag ou commit (defaut: branche par defaut)"}, + }, + "required": ["repo"], + }, + }, + { + "name": "read_file", + "description": "Lit le contenu texte d un fichier d un depot. repo au format proprietaire/depot. Refuse les fichiers binaires et ceux depassant la taille limite.", + "inputSchema": { + "type": "object", + "properties": { + "repo": {"type": "string", "description": "proprietaire/depot"}, + "path": {"type": "string", "description": "Chemin du fichier dans le depot"}, + "ref": {"type": "string", "description": "Branche, tag ou commit (defaut: branche par defaut)"}, + }, + "required": ["repo", "path"], + }, + }, + { + "name": "get_metadata", + "description": "Metadonnees d un fichier (taille, sha, type, urls) sans telecharger le contenu. Utile avant de lire un gros fichier.", + "inputSchema": { + "type": "object", + "properties": { + "repo": {"type": "string", "description": "proprietaire/depot"}, + "path": {"type": "string", "description": "Chemin du fichier dans le depot"}, + "ref": {"type": "string", "description": "Branche, tag ou commit (defaut: branche par defaut)"}, + }, + "required": ["repo", "path"], + }, + }, +] + +# --------------------------------------------------------------------------- +# Implementation des tools +# --------------------------------------------------------------------------- + +MAX_FILE_BYTES = 200000 # garde-fou : au-dela, on renvoie les metadonnees + + +def _valider_repo(repo): + """Le format attendu est proprietaire/depot. Evite les traversees de chemin.""" + if not repo or repo.count("/") != 1: + return {"error": "repo doit etre au format proprietaire/depot (ex: Master/sliding-automation)"} + if ".." in repo: + return {"error": "repo invalide"} + return None + + +def _valider_path(path): + if path and ".." in path: + return {"error": "path invalide (traversee de repertoire)"} + return None + + +def tool_list_repos(): + repos = forgejo_api("GET", "/api/v1/user/repos") or [] + return [{ + "full_name": r.get("full_name"), + "description": r.get("description") or "", + "default_branch": r.get("default_branch"), + "private": r.get("private"), + "updated_at": r.get("updated_at"), + "size_kb": r.get("size"), + } for r in repos] + + +def tool_list_files(repo, path="", ref=None): + err = _valider_repo(repo) or _valider_path(path) + if err: + return err + params = {"ref": ref} if ref else None + data = forgejo_api("GET", "/api/v1/repos/%s/contents/%s" % (repo, path.lstrip("/")), params=params) + if isinstance(data, dict): + data = [data] + return [{ + "name": e.get("name"), + "path": e.get("path"), + "type": e.get("type"), + "size": e.get("size"), + } for e in (data or [])] + + +def tool_read_file(repo, path, ref=None): + err = _valider_repo(repo) or _valider_path(path) + if err: + return err + params = {"ref": ref} if ref else None + data = forgejo_api("GET", "/api/v1/repos/%s/contents/%s" % (repo, path.lstrip("/")), params=params) + if not data or data.get("type") != "file": + return {"error": "Fichier introuvable ou chemin non fichier : %s" % path} + size = data.get("size") or 0 + if size > MAX_FILE_BYTES: + return {"error": "Fichier trop volumineux (%d octets, limite %d). Utiliser get_metadata." % (size, MAX_FILE_BYTES), + "path": path, "repo": repo, "size": size} + encoded = data.get("content") + if encoded and data.get("encoding") == "base64": + raw = base64.b64decode(encoded) + else: + dl = requests.get(data["download_url"], + headers={"Authorization": "token %s" % FORGEJO_TOKEN}, + timeout=30) + dl.raise_for_status() + raw = dl.content + try: + content = raw.decode("utf-8") + except UnicodeDecodeError: + return {"error": "Fichier binaire, lecture texte impossible", "path": path, "repo": repo} + return {"repo": repo, "path": path, "ref": ref or data.get("ref"), + "size": size, "sha": data.get("sha"), "content": content} + + +def tool_get_metadata(repo, path, ref=None): + err = _valider_repo(repo) or _valider_path(path) + if err: + return err + params = {"ref": ref} if ref else None + data = forgejo_api("GET", "/api/v1/repos/%s/contents/%s" % (repo, path.lstrip("/")), params=params) + if not data: + return {"error": "Introuvable : %s" % path} + if isinstance(data, list): + return {"repo": repo, "path": path, "type": "dir", "entries": len(data)} + return { + "repo": repo, + "path": data.get("path"), + "type": data.get("type"), + "size": data.get("size"), + "sha": data.get("sha"), + "encoding": data.get("encoding"), + "html_url": data.get("html_url"), + "download_url": data.get("download_url"), + } + + +DISPATCH = { + "list_repos": tool_list_repos, + "list_files": tool_list_files, + "read_file": tool_read_file, + "get_metadata": tool_get_metadata, +} + +# --------------------------------------------------------------------------- +# Handler JSON-RPC MCP +# --------------------------------------------------------------------------- + + +def check_auth(request): + auth = request.headers.get("authorization", "") + token = auth.replace("Bearer ", "").strip() + # 1) Tokens statiques (Le Chat, cle Claude historique) + if token and token in VALID_TOKENS: + return VALID_TOKENS[token] + # 2) Tokens OAuth emis (Claude Desktop) + if token.startswith("at_"): + oauth = load_oauth_state() + meta = oauth["tokens"].get(token) + if meta: + return "Claude-OAuth" + return None + + +async def mcp_endpoint(request: Request): + # Auth + llm = check_auth(request) + if not llm: + return JSONResponse( + {"error": "invalid_token", "error_description": "Authentication required"}, + status_code=401, + headers={"WWW-Authenticate": "Bearer"}, + ) + + body = await request.json() + method = body.get("method") + req_id = body.get("id") + params = body.get("params", {}) + + # initialize + if method == "initialize": + return JSONResponse({ + "jsonrpc": "2.0", "id": req_id, + "result": { + "protocolVersion": PROTOCOL_VERSION, + "capabilities": {"tools": {}}, + "serverInfo": SERVER_INFO, + }, + }) + + # notifications/initialized (pas de reponse attendue) + if method == "notifications/initialized": + return Response(status_code=202) + + # tools/list + if method == "tools/list": + return JSONResponse({ + "jsonrpc": "2.0", "id": req_id, + "result": {"tools": TOOLS}, + }) + + # tools/call + if method == "tools/call": + tool_name = params.get("name") + args = params.get("arguments", {}) + fn = DISPATCH.get(tool_name) + if not fn: + return JSONResponse({ + "jsonrpc": "2.0", "id": req_id, + "error": {"code": -32601, "message": "Tool inconnu : %s" % tool_name}, + }) + try: + result = fn(**args) + return JSONResponse({ + "jsonrpc": "2.0", "id": req_id, + "result": { + "content": [{"type": "text", + "text": json.dumps(result, ensure_ascii=False, indent=2)}], + }, + }) + except requests.HTTPError as e: + status = e.response.status_code if e.response is not None else "?" + return JSONResponse({ + "jsonrpc": "2.0", "id": req_id, + "error": {"code": -32603, "message": "Forgejo HTTP %s : %s" % (status, str(e))}, + }) + except Exception as e: + return JSONResponse({ + "jsonrpc": "2.0", "id": req_id, + "error": {"code": -32603, "message": str(e)}, + }) + + # methode inconnue + return JSONResponse({ + "jsonrpc": "2.0", "id": req_id, + "error": {"code": -32601, "message": "Methode inconnue : %s" % method}, + }) + + +async def health(request: Request): + return JSONResponse({"status": "ok", "server": "mcp-forgejo", "version": "1.0.0"}) + + +async def well_known_protected_resource(request: Request): + return JSONResponse({ + "resource": OAUTH_BASE_URL + "/mcp", + "authorization_servers": [OAUTH_BASE_URL], + }) + + +async def well_known_authorization_server(request: Request): + return JSONResponse({ + "issuer": OAUTH_BASE_URL, + "authorization_endpoint": OAUTH_BASE_URL + "/authorize", + "token_endpoint": OAUTH_BASE_URL + "/token", + "registration_endpoint": OAUTH_BASE_URL + "/register", + "response_types_supported": ["code"], + "grant_types_supported": ["authorization_code", "refresh_token"], + "code_challenge_methods_supported": ["S256"], + "token_endpoint_auth_methods_supported": ["none"], + }) + + +async def register(request: Request): + """Dynamic Client Registration (RFC 7591).""" + try: + body = await request.json() + except Exception: + body = {} + client_id = "client_" + secrets.token_urlsafe(16) + state = load_oauth_state() + state["clients"][client_id] = { + "client_id": client_id, + "redirect_uris": body.get("redirect_uris", []), + "client_name": body.get("client_name", "unknown"), + "created_at": int(time.time()), + } + save_oauth_state(state) + return JSONResponse({ + "client_id": client_id, + "redirect_uris": body.get("redirect_uris", []), + "client_name": body.get("client_name", "unknown"), + "token_endpoint_auth_method": "none", + "grant_types": ["authorization_code", "refresh_token"], + "response_types": ["code"], + }, status_code=201) + + +AUTHORIZE_PAGE = """ +Autorisation Forgejo MCP + +

Autorisation Forgejo MCP

+

Saisissez le mot de securite pour autoriser l acces.

+{error} +
+ + + + + + + +
""" + + +async def authorize_get(request: Request): + q = request.query_params + page = AUTHORIZE_PAGE.format( + error="", + client_id=q.get("client_id", ""), + redirect_uri=q.get("redirect_uri", ""), + state=q.get("state", ""), + code_challenge=q.get("code_challenge", ""), + code_challenge_method=q.get("code_challenge_method", "S256"), + ) + return HTMLResponse(page) + + +async def authorize_post(request: Request): + form = await request.form() + word = form.get("security_word", "") + client_id = form.get("client_id", "") + redirect_uri = form.get("redirect_uri", "") + state_param = form.get("state", "") + code_challenge = form.get("code_challenge", "") + code_challenge_method = form.get("code_challenge_method", "S256") + + if not SECURITY_WORD or word != SECURITY_WORD: + page = AUTHORIZE_PAGE.format( + error='

Mot de securite incorrect.

', + client_id=client_id, redirect_uri=redirect_uri, + state=state_param, code_challenge=code_challenge, + code_challenge_method=code_challenge_method, + ) + return HTMLResponse(page, status_code=401) + + oauth = load_oauth_state() + if client_id not in oauth["clients"]: + return JSONResponse({"error": "client inconnu"}, status_code=400) + + code = "code_" + secrets.token_urlsafe(24) + oauth["auth_codes"][code] = { + "client_id": client_id, + "redirect_uri": redirect_uri, + "code_challenge": code_challenge, + "code_challenge_method": code_challenge_method, + "created_at": int(time.time()), + "used": False, + } + save_oauth_state(oauth) + + sep = "&" if "?" in redirect_uri else "?" + location = redirect_uri + sep + "code=" + code + if state_param: + location += "&state=" + state_param + return RedirectResponse(location, status_code=302) + + +def verify_pkce(code_verifier, code_challenge): + """Verifie PKCE S256 : base64url(sha256(verifier)) == challenge.""" + digest = hashlib.sha256(code_verifier.encode("ascii")).digest() + computed = base64.urlsafe_b64encode(digest).decode("ascii").rstrip("=") + return computed == code_challenge + + +async def token(request: Request): + form = await request.form() + grant_type = form.get("grant_type", "") + oauth = load_oauth_state() + + if grant_type == "authorization_code": + code = form.get("code", "") + code_verifier = form.get("code_verifier", "") + entry = oauth["auth_codes"].get(code) + if not entry: + return JSONResponse({"error": "invalid_grant"}, status_code=400) + if entry.get("used"): + return JSONResponse({"error": "invalid_grant", "error_description": "code deja utilise"}, status_code=400) + challenge = entry.get("code_challenge", "") + if challenge: + if not code_verifier or not verify_pkce(code_verifier, challenge): + return JSONResponse({"error": "invalid_grant", "error_description": "PKCE echec"}, status_code=400) + access_token = "at_" + secrets.token_urlsafe(32) + refresh_token = "rt_" + secrets.token_urlsafe(32) + oauth["tokens"][access_token] = { + "client_id": entry["client_id"], + "refresh_token": refresh_token, + "created_at": int(time.time()), + "type": "access", + } + oauth["auth_codes"][code]["used"] = True + save_oauth_state(oauth) + return JSONResponse({ + "access_token": access_token, + "token_type": "Bearer", + "expires_in": 31536000, + "refresh_token": refresh_token, + }) + + elif grant_type == "refresh_token": + rt = form.get("refresh_token", "") + for at, meta in list(oauth["tokens"].items()): + if meta.get("refresh_token") == rt: + new_at = "at_" + secrets.token_urlsafe(32) + oauth["tokens"][new_at] = { + "client_id": meta["client_id"], + "refresh_token": rt, + "created_at": int(time.time()), + "type": "access", + } + save_oauth_state(oauth) + return JSONResponse({ + "access_token": new_at, + "token_type": "Bearer", + "expires_in": 31536000, + "refresh_token": rt, + }) + return JSONResponse({"error": "invalid_grant"}, status_code=400) + + return JSONResponse({"error": "unsupported_grant_type"}, status_code=400) + + +app = Starlette(routes=[ + Route("/mcp", mcp_endpoint, methods=["POST"]), + Route("/", mcp_endpoint, methods=["POST"]), + Route("/health", health, methods=["GET"]), + Route("/.well-known/oauth-protected-resource", well_known_protected_resource, methods=["GET"]), + Route("/.well-known/oauth-authorization-server", well_known_authorization_server, methods=["GET"]), + Route("/register", register, methods=["POST"]), + Route("/authorize", authorize_get, methods=["GET"]), + Route("/authorize", authorize_post, methods=["POST"]), + Route("/token", token, methods=["POST"]), +]) + +if __name__ == "__main__": + import uvicorn + uvicorn.run(app, host="127.0.0.1", port=8770) diff --git a/start_mcp_forgejo.sh b/start_mcp_forgejo.sh new file mode 100644 index 0000000..967f711 --- /dev/null +++ b/start_mcp_forgejo.sh @@ -0,0 +1,8 @@ +#!/bin/bash +cd /volume1/homes/Master/App/Code_versioning +# Tuer tout uvicorn mcp_forgejo_server existant (evite les zombies sur le port 8770) +pkill -f "uvicorn mcp_forgejo_server:app" 2>/dev/null +sleep 2 +source venv/bin/activate +uvicorn mcp_forgejo_server:app --host 127.0.0.1 --port 8770 >> /volume1/homes/Master/App/Code_versioning/mcp_forgejo.log 2>&1 & +echo $! > /volume1/homes/Master/App/Code_versioning/mcp_forgejo.pid diff --git a/watchdog.sh b/watchdog.sh new file mode 100644 index 0000000..2fc5313 --- /dev/null +++ b/watchdog.sh @@ -0,0 +1,37 @@ +#!/bin/sh +# Watchdog Code Versioning — verifie le MCP Forgejo (8770), +# relance via start_mcp_forgejo.sh s'il ne repond pas sur /health. +# Idempotent : couvre le boot (rien ne tourne) ET le crash. +# Lance par tache planifiee DSM toutes les 5 min. +BASE=/volume1/homes/Master/App/Code_versioning +LOG="$BASE/watchdog.log" +TS=$(date '+%Y-%m-%d %H:%M:%S') +wait_health() { + URL="$1" + TRIES="$2" + i=0 + while [ "$i" -lt "$TRIES" ]; do + if curl -s -f -m 5 "$URL" > /dev/null 2>&1; then + return 0 + fi + i=$((i + 1)) + sleep 3 + done + return 1 +} +check_and_restart() { + NAME="$1" + URL="$2" + STARTER="$3" + if curl -s -f -m 5 "$URL" > /dev/null 2>&1; then + return 0 + fi + echo "$TS [$NAME] KO sur $URL — relance via $STARTER" >> "$LOG" + sh "$BASE/$STARTER" + if wait_health "$URL" 5; then + echo "$TS [$NAME] relance OK" >> "$LOG" + else + echo "$TS [$NAME] ECHEC relance — toujours muet apres ~15s" >> "$LOG" + fi +} +check_and_restart "mcp-forgejo" "http://127.0.0.1:8770/health" "start_mcp_forgejo.sh"