#!/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)