Connecter un MCP local à ChatGPT pour travailler sur les fichiers de son PC… SANS limite de temps.

py -3.11 -m pip install "mcp>=2,<3"

py -m pip install "mcp>=2,<3"

cloudflared tunnel --url http://127.0.0.1:8765

$env:MCP_TOKEN = [Convert]::ToBase64String([Security.Cryptography.RandomNumberGenerator]::GetBytes(32)).TrimEnd('=').Replace('+','-').Replace('/','_')
$env:MCP_PUBLIC_HOST = "mots-aleatoires.trycloudflare.com"
py -3.11 C:\dev\dev_mcp.py

https://mots-aleatoires.trycloudflare.com/VOTRE_JETON/mcp

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Dev MCP local pour ChatGPT. Fichier unique ; dependance: pip install 'mcp>=2,<3'.

Demarrage Windows PowerShell :
    py -m pip install "mcp>=2,<3"
    py dev_mcp.py

Par defaut, les outils ne peuvent agir que sous C:\\dev\\ai-team. Connexion distante :
Secure MCP Tunnel (recommande) ou tunnel HTTPS temporaire avec URL secrete.
Ne communiquez jamais cette URL secrete publiquement.
"""

import ast
import fnmatch
import hashlib
import json
import os
from pathlib import Path
import re
import secrets
import shutil
import subprocess
import sys
from datetime import datetime, timezone

try:
    from mcp.server import MCPServer
    from mcp.server.transport_security import TransportSecuritySettings
except ImportError:
    print('Dependance manquante. Installe-la avec : py -m pip install "mcp>=2,<3"', file=sys.stderr)
    raise SystemExit(1)

# Configuration par variables d'environnement (aucun fichier de config necessaire).
ROOT = Path(r"C:\dev\ai-team" if os.name == "nt" else str(Path.home() / "dev" / "ai-team")).expanduser().resolve()
PORT = int(os.getenv("MCP_PORT", "8765"))
TOKEN = os.getenv("MCP_TOKEN") or secrets.token_urlsafe(30)
if not re.fullmatch(r"[A-Za-z0-9_-]{24,128}", TOKEN):
    raise SystemExit("MCP_TOKEN doit contenir 24 a 128 caracteres URL-safe.")
MCP_PATH = f"/{TOKEN}/mcp"
PUBLIC_HOST = os.getenv("MCP_PUBLIC_HOST", "").strip().lower()
ALLOW_TESTS = os.getenv("MCP_ALLOW_TESTS", "0") == "1"
BACKUP_DIR = ".mcp_backups"

IGNORED_DIRS = {
    ".git", ".svn", ".hg", ".venv", "venv", "node_modules",
    "vendor", "dist", "build", "__pycache__", ".next", BACKUP_DIR,
}
# Fichiers susceptibles de contenir des secrets : interdits aussi en lecture.
SECRET_NAMES = {
    "wp-config.php", ".npmrc", ".pypirc", ".netrc", ".htpasswd",
    "id_rsa", "id_ed25519", "credentials.json", "secrets.json",
    "service-account.json", "application_default_credentials.json",
}
SECRET_SUFFIXES = {".pem", ".key", ".p12", ".pfx", ".jks", ".sqlite", ".db"}
MAX_FILE_BYTES = 2_000_000
MAX_WRITE_BYTES = 1_500_000
ROOT.mkdir(parents=True, exist_ok=True)

mcp = MCPServer(
    "Atelier Dev Local",
    instructions=(
        "Acces UNIQUEMENT au dossier de developpement configure. "
        "Pour modifier du code existant, lis d'abord le fichier, puis utilise replace_text "
        "ou write_file avec expected_sha256. Ne touche pas aux secrets. "
        "Tests de projet desactives par defaut. Les outils ne permettent pas "
        "d'executer des commandes shell arbitraires."
    ),
)


def safe_path(path: str = ".", *, allow_root: bool = True) -> Path:
    if not isinstance(path, str) or not path.strip():
        raise ValueError("Chemin vide.")
    p = Path(path)
    if p.is_absolute():
        # Les chemins absolus sont acceptes uniquement s'ils restent dans ROOT.
        candidate = p.resolve(strict=False)
    else:
        candidate = (ROOT / p).resolve(strict=False)
    if not candidate.is_relative_to(ROOT):
        raise PermissionError("Acces refuse : chemin hors du dossier autorise.")
    if not allow_root and candidate == ROOT:
        raise PermissionError("Operation interdite sur le dossier racine.")
    relative_parts = candidate.relative_to(ROOT).parts
    for component in relative_parts:
        lower = component.lower()
        if lower in IGNORED_DIRS or lower in SECRET_NAMES or lower.startswith(".env"):
            raise PermissionError(f"Chemin protege : {component}")
        if Path(lower).suffix in SECRET_SUFFIXES or "secret" in lower or "credential" in lower:
            raise PermissionError(f"Fichier potentiellement sensible : {component}")
    return candidate


def rel(path: Path) -> str:
    return path.relative_to(ROOT).as_posix() or "."


def sha(path: Path) -> str:
    h = hashlib.sha256()
    with path.open("rb") as f:
        for block in iter(lambda: f.read(65536), b""):
            h.update(block)
    return h.hexdigest()


def read_text(path: Path) -> str:
    if not path.is_file():
        raise FileNotFoundError(rel(path))
    if path.stat().st_size > MAX_FILE_BYTES:
        raise ValueError("Fichier trop volumineux (>2 Mo) ; utiliser un autre outil.")
    try:
        with path.open("r", encoding="utf-8-sig", newline="") as stream:
            return stream.read()
    except UnicodeDecodeError as exc:
        raise ValueError("Fichier non UTF-8 ou binaire.") from exc


def backup(path: Path) -> str | None:
    if not path.is_file():
        return None
    stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%S")
    dest = ROOT / BACKUP_DIR / f"{stamp}_{secrets.token_hex(5)}" / path.relative_to(ROOT)
    dest.parent.mkdir(parents=True, exist_ok=True)
    shutil.copy2(path, dest)
    return dest.relative_to(ROOT / BACKUP_DIR).as_posix()


def save_text(path: Path, text: str, expected_sha256: str = "") -> dict:
    if not isinstance(text, str):
        raise ValueError("Le contenu doit etre du texte UTF-8.")
    if len(text.encode("utf-8")) > MAX_WRITE_BYTES:
        raise ValueError("Ecriture limitee a 1,5 Mo par appel.")
    if path.exists() and not path.is_file():
        raise IsADirectoryError(rel(path))
    if expected_sha256 and (not path.exists() or sha(path) != expected_sha256):
        raise ValueError("Le fichier a change : empreinte SHA256 differente. Relis-le.")
    previous_backup = backup(path)
    path.parent.mkdir(parents=True, exist_ok=True)
    # Remplacement atomique lorsque possible, a cote du fichier de destination.
    temp = path.with_name(f".{path.name}.mcp-{secrets.token_hex(6)}.tmp")
    try:
        temp.write_text(text, encoding="utf-8", newline="")
        temp.replace(path)
    finally:
        temp.unlink(missing_ok=True)
    return {"path": rel(path), "bytes": path.stat().st_size,
            "sha256": sha(path), "backup_id": previous_backup}


def walk_files(start: Path):
    for folder, dirs, files in os.walk(start, followlinks=False):
        d = Path(folder)
        dirs[:] = [name for name in sorted(dirs, key=str.lower)
                   if not name.startswith(".mcp-") and
                   (lambda p: _allowed_for_walk(p))(d / name)]
        for name in sorted(files, key=str.lower):
            p = d / name
            if _allowed_for_walk(p) and p.is_file():
                yield p


def _allowed_for_walk(path: Path) -> bool:
    try:
        safe_path(str(path))
        return True
    except (PermissionError, ValueError, OSError):
        return False


def run_process(args: list[str], cwd: Path, timeout: int = 25) -> dict:
    try:
        result = subprocess.run(args, cwd=cwd, capture_output=True, text=True,
                                encoding="utf-8", errors="replace", timeout=timeout,
                                shell=False, stdin=subprocess.DEVNULL)
    except FileNotFoundError:
        return {"error": f"Programme absent : {args[0]}"}
    except subprocess.TimeoutExpired:
        return {"error": f"Delai depasse : {timeout} secondes"}
    return {"exit_code": result.returncode,
            "stdout": result.stdout[-24000:], "stderr": result.stderr[-10000:]}


def _check_git_root(folder: Path) -> None:
    """Interdit de consulter un depot Git situe au-dessus de C:/dev."""
    result = run_process(["git", "rev-parse", "--show-toplevel"], folder)
    if result.get("exit_code") != 0:
        raise ValueError("Aucun depot Git dans ce repertoire.")
    repo = Path(result["stdout"].strip()).resolve()
    if not repo.is_relative_to(ROOT):
        raise PermissionError("Depot Git hors du dossier autorise.")


@mcp.tool()
def atelier_info() -> dict:
    """Affiche la racine autorisee, les projets de premier niveau et les options de securite."""
    projects = [x.name for x in ROOT.iterdir() if x.is_dir() and _allowed_for_walk(x)]
    return {"root": str(ROOT), "projects": sorted(projects)[:150],
            "tests_enabled": ALLOW_TESTS, "port": PORT,
            "note": "Aucun fichier hors de cette racine n'est accessible."}


@mcp.tool()
def list_files(path: str = ".", depth: int = 2, limit: int = 180) -> dict:
    """Liste dossiers et fichiers, avec profondeur et volume limites. Ignore les dependances."""
    start = safe_path(path)
    if not start.exists():
        raise FileNotFoundError(rel(start))
    if start.is_file():
        return {"entries": [rel(start)]}
    depth = max(0, min(depth, 6))
    limit = max(1, min(limit, 300))
    entries = []
    for current, dirs, files in os.walk(start, followlinks=False):
        base = Path(current)
        nesting = len(base.relative_to(start).parts)
        dirs[:] = [d for d in sorted(dirs, key=str.lower)
                   if _allowed_for_walk(base / d) and nesting < depth]
        for d in dirs:
            entries.append(rel(base / d) + "/")
            if len(entries) >= limit:
                return {"entries": entries, "truncated": True}
        for filename in sorted(files, key=str.lower):
            fp = base / filename
            if _allowed_for_walk(fp):
                entries.append(rel(fp))
                if len(entries) >= limit:
                    return {"entries": entries, "truncated": True}
    return {"entries": entries, "truncated": False}


@mcp.tool()
def read_file(path: str, start_line: int = 1, max_lines: int = 250) -> str:
    """Lit un fichier UTF-8, renvoie du TEXTE (lignes numerotees + SHA256).

    Retour en texte brut pour eviter les problemes de conversion de la cle
    'content' dans certaines passerelles MCP / ChatGPT.
    """
    p = safe_path(path, allow_root=False)
    data = read_text(p)
    lines = data.splitlines(keepends=True)
    start = max(1, start_line)
    end = min(len(lines), start + max(1, min(max_lines, 450)) - 1)
    chunk = "".join(f"{i+1}: {lines[i]}" for i in range(start-1, end))
    truncated = end < len(lines) or len(chunk) > 48000
    header = (f"PATH: {rel(p)}\n"
              f"SHA256: {sha(p)}\n"
              f"TOTAL_LINES: {len(lines)}\n"
              f"FROM_LINE: {start}\n"
              f"TO_LINE: {end}\n"
              f"TRUNCATED: {str(truncated).lower()}\n"
              "--- FILE CONTENT ---\n")
    return header + chunk[:48000]


@mcp.tool()
def search_code(query: str, path: str = ".", file_glob: str = "*", regex: bool = False,
                max_results: int = 60) -> dict:
    """Cherche du texte ou une regex dans un projet. Retourne fichier, ligne et extrait."""
    if not query or len(query) > 250:
        raise ValueError("Recherche vide ou trop longue.")
    start = safe_path(path)
    if not start.is_dir():
        raise NotADirectoryError(rel(start))
    pattern = re.compile(query, re.IGNORECASE) if regex else None
    max_results = max(1, min(max_results, 120))
    matches = []
    scanned = 0
    for p in walk_files(start):
        if not fnmatch.fnmatch(p.name, file_glob) or p.stat().st_size > 500_000:
            continue
        scanned += 1
        if scanned > 3000:
            break
        try:
            with p.open(encoding="utf-8-sig") as f:
                for n, line in enumerate(f, 1):
                    hit = bool(pattern.search(line)) if pattern else query.lower() in line.lower()
                    if hit:
                        matches.append({"path": rel(p), "line": n, "text": line.strip()[:250]})
                        if len(matches) >= max_results:
                            return {"matches": matches, "files_scanned": scanned, "truncated": True}
        except (UnicodeError, OSError):
            continue
    return {"matches": matches, "files_scanned": scanned,
            "truncated": scanned > 3000}


@mcp.tool()
def create_directory(path: str) -> dict:
    """Cree un dossier (et ses parents) dans la racine autorisee seulement."""
    p = safe_path(path, allow_root=False)
    p.mkdir(parents=True, exist_ok=True)
    return {"created_or_exists": rel(p)}


@mcp.tool()
def write_file(path: str, content: str, expected_sha256: str = "") -> dict:
    """Cree ou remplace un fichier texte UTF-8. Sauvegarde automatique avant remplacement.
    Si l'empreinte SHA256 est fournie, refuse de modifier si le fichier a change.
    """
    return save_text(safe_path(path, allow_root=False), content, expected_sha256)


@mcp.tool()
def write_files(files: list[dict[str, str]]) -> dict:
    """Cree ou modifie plusieurs fichiers texte en un seul appel (maximum 25).
    Chaque element contient path, content et facultativement expected_sha256.
    Valide tous les chemins et empreintes avant de commencer, puis sauvegarde.
    """
    if not files or len(files) > 25:
        raise ValueError("Fournir de 1 a 25 fichiers.")
    prepared = []
    seen = set()
    total_bytes = 0
    for item in files:
        if not isinstance(item, dict) or not isinstance(item.get("content"), str):
            raise ValueError("Chaque entree doit contenir path et content texte.")
        p = safe_path(item.get("path", ""), allow_root=False)
        if p in seen:
            raise ValueError(f"Chemin duplique : {rel(p)}")
        seen.add(p)
        expected = item.get("expected_sha256", "")
        if p.exists() and not p.is_file():
            raise IsADirectoryError(rel(p))
        if expected and (not p.exists() or sha(p) != expected):
            raise ValueError(f"Empreinte SHA256 differente pour {rel(p)}")
        data = item["content"]
        data_bytes = len(data.encode("utf-8"))
        if data_bytes > MAX_WRITE_BYTES:
            raise ValueError(f"Fichier trop grand : {rel(p)}")
        total_bytes += data_bytes
        prepared.append((p, data, expected))
    if total_bytes > 3_000_000:
        raise ValueError("Lot limite a 3 Mo.")
    results = []
    for p, content, expected in prepared:
        # Une erreur disque reste possible apres quelques ecritures : non transactionnel.
        results.append(save_text(p, content, expected))
    return {"written": len(results), "files": results}


@mcp.tool()
def replace_text(path: str, old: str, new: str, expected_count: int = 1) -> dict:
    """Remplace exactement une portion de texte dans un fichier existant, avec sauvegarde.
    Refuse si le nombre d'occurrences ne correspond pas (1 par defaut).
    """
    if not old:
        raise ValueError("Texte 'old' vide interdit.")
    p = safe_path(path, allow_root=False)
    text = read_text(p)
    found = text.count(old)
    if found != expected_count or not (1 <= expected_count <= 1000):
        raise ValueError(f"Remplacement refuse : {found} occurrence(s), {expected_count} attendue(s).")
    result = save_text(p, text.replace(old, new), sha(p))
    return {**result, "replacements": found}


@mcp.tool()
def append_file(path: str, content: str) -> dict:
    """Ajoute du texte a la fin d'un fichier, sauvegarde l'ancienne version si presente."""
    p = safe_path(path, allow_root=False)
    previous = read_text(p) if p.exists() else ""
    return save_text(p, previous + content, sha(p) if p.exists() else "")


@mcp.tool()
def move_file(source: str, destination: str) -> dict:
    """Deplace un fichier vers un chemin inexistant dans le dossier autorise."""
    src = safe_path(source, allow_root=False)
    dst = safe_path(destination, allow_root=False)
    if not src.is_file():
        raise FileNotFoundError(rel(src))
    if dst.exists():
        raise FileExistsError(rel(dst))
    dst.parent.mkdir(parents=True, exist_ok=True)
    src.rename(dst)
    return {"from": rel(src), "to": rel(dst)}


@mcp.tool()
def delete_file(path: str, confirm: str = "") -> dict:
    """Supprime uniquement un fichier (jamais un dossier), apres sauvegarde.
    Pour confirmer, fournir confirm='SUPPRIMER'.
    """
    if confirm != "SUPPRIMER":
        raise PermissionError("Confirmation explicite requise : SUPPRIMER")
    p = safe_path(path, allow_root=False)
    if not p.is_file():
        raise FileNotFoundError(rel(p))
    b = backup(p)
    p.unlink()
    return {"deleted": rel(p), "backup_id": b}


@mcp.tool()
def list_backups(limit: int = 35) -> dict:
    """Liste les sauvegardes automatiques creees avant modification ou suppression."""
    base = ROOT / BACKUP_DIR
    if not base.exists():
        return {"backups": []}
    backups = []
    for p in base.glob("**/*"):
        if p.is_file():
            backups.append(p.relative_to(base).as_posix())
    return {"backups": sorted(backups, reverse=True)[:max(1, min(limit, 150))]}


@mcp.tool()
def restore_backup(backup_id: str, confirm: str = "") -> dict:
    """Restaure une sauvegarde automatique. Exige confirm='RESTAURER'."""
    if confirm != "RESTAURER":
        raise PermissionError("Confirmation explicite requise : RESTAURER")
    base = (ROOT / BACKUP_DIR).resolve()
    source = (base / backup_id).resolve()
    if not source.is_relative_to(base) or not source.is_file():
        raise PermissionError("Sauvegarde introuvable ou chemin invalide.")
    parts = source.relative_to(base).parts
    if len(parts) < 2:
        raise ValueError("Identifiant de sauvegarde invalide.")
    target = safe_path(str(Path(*parts[1:])), allow_root=False)
    existing_backup = backup(target)
    target.parent.mkdir(parents=True, exist_ok=True)
    shutil.copy2(source, target)
    return {"restored": rel(target), "previous_version_backup": existing_backup}


@mcp.tool()
def git_status(path: str = ".") -> dict:
    """Retourne le statut Git (lecture seule) d'un projet situe dans la racine autorisee."""
    folder = safe_path(path)
    if not folder.is_dir():
        raise NotADirectoryError(rel(folder))
    _check_git_root(folder)
    return run_process(["git", "status", "--short", "--branch"], folder)


@mcp.tool()
def git_diff(path: str = ".", staged: bool = False) -> dict:
    """Affiche les modifications Git non commitees (lecture seule, sortie limitee)."""
    folder = safe_path(path)
    if not folder.is_dir():
        raise NotADirectoryError(rel(folder))
    _check_git_root(folder)
    args = ["git", "diff", "--no-ext-diff"]
    if staged:
        args.append("--cached")
    return run_process(args, folder)


@mcp.tool()
def check_syntax(path: str) -> dict:
    """Verifie la syntaxe Python, JSON, PHP ou JS sans executer les fichiers source.
    Pour PHP et JS, php / node doivent etre installes localement.
    """
    p = safe_path(path, allow_root=False)
    if not p.is_file():
        raise FileNotFoundError(rel(p))
    ext = p.suffix.lower()
    if ext == ".py":
        try:
            ast.parse(read_text(p), filename=rel(p))
            return {"valid": True, "path": rel(p), "language": "python"}
        except SyntaxError as exc:
            return {"valid": False, "path": rel(p), "error": str(exc)}
    if ext == ".json":
        try:
            json.loads(read_text(p))
            return {"valid": True, "path": rel(p), "language": "json"}
        except json.JSONDecodeError as exc:
            return {"valid": False, "path": rel(p), "error": str(exc)}
    if ext == ".php":
        return run_process(["php", "-l", str(p)], p.parent)
    if ext in {".js", ".mjs", ".cjs"}:
        return run_process(["node", "--check", str(p)], p.parent)
    raise ValueError("Extensions supportees : .py, .json, .php, .js, .mjs, .cjs")


@mcp.tool()
def run_tests(path: str = ".", test_runner: str = "pytest") -> dict:
    """Lance UNIQUEMENT pytest ou npm test, si MCP_ALLOW_TESTS=1.
    Attention : les tests d'un projet peuvent executer du code et modifier des donnees.
    """
    if not ALLOW_TESTS:
        raise PermissionError("Tests desactives. Demarrer avec MCP_ALLOW_TESTS=1 pour les autoriser.")
    folder = safe_path(path)
    if not folder.is_dir():
        raise NotADirectoryError(rel(folder))
    commands = {"pytest": [sys.executable, "-m", "pytest", "-q"],
                "npm": ["npm", "test"]}
    if test_runner not in commands:
        raise ValueError("Test runner autorise : pytest ou npm")
    if test_runner == "npm" and not (folder / "package.json").is_file():
        raise FileNotFoundError("package.json absent dans le dossier cible.")
    return run_process(commands[test_runner], folder, timeout=60)


if __name__ == "__main__":
    allowed_hosts = ["127.0.0.1:*", "localhost:*", "[::1]:*"]
    if PUBLIC_HOST:
        if not re.fullmatch(r"[a-z0-9][a-z0-9.-]{1,250}[a-z0-9]", PUBLIC_HOST):
            raise SystemExit("MCP_PUBLIC_HOST invalide : indiquer seulement le domaine, sans https://")
        allowed_hosts += [PUBLIC_HOST, PUBLIC_HOST + ":*"]
    security = TransportSecuritySettings(
        allowed_hosts=allowed_hosts,
        allowed_origins=["http://127.0.0.1:*", "http://localhost:*",
                         "https://chatgpt.com", "https://chat.openai.com"],
    )
    print("\n=== ATELIER DEV MCP ===")
    print(f"Dossier autorise : {ROOT}")
    print(f"URL locale : http://127.0.0.1:{PORT}{MCP_PATH}")
    print(f"Tests avec execution : {'ACTIFS' if ALLOW_TESTS else 'DESACTIVES'}")
    print("Ne partagez pas le token present dans l'URL.")
    print("Pour ChatGPT : utiliser le Secure MCP Tunnel (recommande).\n", flush=True)
    mcp.run(transport="streamable-http", host="127.0.0.1", port=PORT,
            streamable_http_path=MCP_PATH, stateless_http=True,
            json_response=True, transport_security=security)