CrazyClaude/mcp/qwen3-mcp/server.py

#!/usr/bin/env -S venv/bin/python3
"""
MCP server stdio exposant un outil `task`.
Lance aichat en mode serveur au démarrage et gère la boucle tool_calls.
Les outils disponibles pour Qwen3 sont chargés depuis llm-functions/functions.json.
Appelé par Claude Code via : claude mcp add --transport stdio qwen3 -- .claude/venv/bin/python3 .claude/mcp/qwen3-mcp/server.py
"""

import atexit
import json
import os
import shutil
import socket
import subprocess
import sys
import time
from pathlib import Path

import requests

QWEN3_MODEL = "LMStudio:qwen/qwen3-coder-next"
MAX_TOOL_ITERATIONS = 10

# Répertoire llm-functions relatif à ce script
_HERE = Path(__file__).parent
_LLM_FUNCTIONS_DIR = _HERE.parent.parent / "llm-functions"
_FUNCTIONS_JSON = _LLM_FUNCTIONS_DIR / "functions.json"
_BIN_DIR = _LLM_FUNCTIONS_DIR / "bin"


_qwen3_tools: list[dict] | None = None


def _load_qwen3_tools() -> list[dict]:
    """Charge les outils depuis llm-functions/functions.json (mise en cache au premier appel)."""
    global _qwen3_tools
    if _qwen3_tools is not None:
        return _qwen3_tools
    if not _FUNCTIONS_JSON.exists():
        print(f"WARN: {_FUNCTIONS_JSON} introuvable — aucun outil disponible",
              file=sys.stderr, flush=True)
        _qwen3_tools = []
        return _qwen3_tools
    raw = json.loads(_FUNCTIONS_JSON.read_text())
    _qwen3_tools = [{"type": "function", "function": tool} for tool in raw]
    return _qwen3_tools


# Outil MCP exposé à Claude Code
MCP_TOOLS = [
    {
        "name": "task",
        "description": "Délègue une tâche de codage à Qwen3-Coder via aichat. Qwen3 dispose d'outils filesystem, shell et web (llm-functions).",
        "inputSchema": {
            "type": "object",
            "properties": {
                "task": {"type": "string", "description": "Description précise de la tâche"},
                "files": {
                    "type": "array",
                    "items": {"type": "string"},
                    "description": "Fichiers à mettre en contexte initial (optionnel)",
                },
            },
            "required": ["task"],
        },
    }
]


# --- Gestion du processus aichat ---

_aichat_proc: subprocess.Popen | None = None
_aichat_url: str | None = None


def _find_free_port() -> int:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind(("127.0.0.1", 0))
        return s.getsockname()[1]


def _start_aichat() -> str:
    global _aichat_proc, _aichat_url

    if not shutil.which("aichat"):
        raise RuntimeError("aichat introuvable dans le PATH")

    port = _find_free_port()
    address = f"127.0.0.1:{port}"

    env = os.environ.copy()
    if _LLM_FUNCTIONS_DIR.exists():
        env["AICHAT_FUNCTIONS_DIR"] = str(_LLM_FUNCTIONS_DIR)

    _aichat_proc = subprocess.Popen(
        ["aichat", "--serve", address],
        stdin=subprocess.DEVNULL,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
        env=env,
    )
    atexit.register(_stop_aichat)

    url = f"http://{address}"
    for _ in range(30):
        try:
            requests.get(f"{url}/v1/models", timeout=1)
            _aichat_url = url
            return url
        except requests.RequestException:
            time.sleep(0.3)

    _aichat_proc.kill()
    raise RuntimeError(f"aichat n'a pas démarré sur {address}")


def _stop_aichat():
    if _aichat_proc and _aichat_proc.poll() is None:
        _aichat_proc.terminate()
        try:
            _aichat_proc.wait(timeout=5)
        except subprocess.TimeoutExpired:
            _aichat_proc.kill()


# --- Exécution des outils via llm-functions/bin/ ---

def _execute_tool(name: str, args: dict) -> str:
    """Exécute un outil llm-functions via son binaire dans bin/.
    Les binaires attendent le JSON des arguments comme premier argument positionnel.
    """
    bin_path = _BIN_DIR / name
    if not bin_path.exists():
        return f"ERROR: outil '{name}' introuvable dans {_BIN_DIR}"
    try:
        result = subprocess.run(
            [str(bin_path), json.dumps(args)],
            capture_output=True,
            text=True,
            timeout=60,
        )
        output = (result.stdout + result.stderr).strip()
        return output or f"(exit code {result.returncode})"
    except subprocess.TimeoutExpired:
        return f"ERROR: timeout lors de l'exécution de '{name}'"
    except Exception as e:
        return f"ERROR: {e}"


# --- Boucle tool_calls → Qwen3 ---

def call_qwen3(task: str, files: list[str]) -> str:
    if _aichat_url is None:
        raise RuntimeError("aichat non démarré")

    qwen3_tools = _load_qwen3_tools()

    context = ""
    for path in files:
        try:
            with open(path, encoding="utf-8") as f:
                context += f"--- {path} ---\n{f.read()}\n\n"
        except OSError as e:
            context += f"--- {path} --- ERREUR: {e}\n\n"

    user_content = f"TÂCHE: {task}"
    if context:
        user_content += f"\n\nFICHIERS:\n{context}"

    system = (
        "Tu es un assistant de développement logiciel. "
        "Pour toute tâche impliquant la création ou modification de fichiers, "
        "utilise TOUJOURS les outils fs_write ou fs_patch — ne jamais afficher "
        "le contenu dans ta réponse. "
        "Utilise execute_command pour vérifier le résultat si nécessaire. "
        "Réponds en français, de façon concise."
    )
    messages = [
        {"role": "system", "content": system},
        {"role": "user", "content": user_content},
    ]
    payload: dict = {
        "model": QWEN3_MODEL,
        "messages": messages,
        "temperature": 0.2,
        "max_tokens": 4096,
    }
    if qwen3_tools:
        payload["tools"] = qwen3_tools

    for _ in range(MAX_TOOL_ITERATIONS):
        resp = requests.post(
            f"{_aichat_url}/v1/chat/completions",
            json=payload,
            timeout=120,
        )
        resp.raise_for_status()
        data = resp.json()
        choice = data["choices"][0]
        message = choice["message"]
        payload["messages"].append(message)

        if choice["finish_reason"] != "tool_calls":
            return message.get("content") or ""

        for tc in message.get("tool_calls", []):
            fn = tc["function"]
            args = json.loads(fn["arguments"])
            result = _execute_tool(fn["name"], args)
            payload["messages"].append({
                "role": "tool",
                "tool_call_id": tc["id"],
                "content": result,
            })

    return "ERROR: nombre maximum d'itérations tool_calls atteint"


# --- Protocole MCP stdio ---

def send(obj: dict):
    print(json.dumps(obj), flush=True)


def handle(req: dict):
    method = req.get("method", "")
    req_id = req.get("id")

    if method == "initialize":
        send({
            "jsonrpc": "2.0",
            "id": req_id,
            "result": {
                "protocolVersion": "2024-11-05",
                "capabilities": {"tools": {}},
                "serverInfo": {"name": "qwen3-mcp", "version": "3.0.0"},
            },
        })
    elif method == "notifications/initialized":
        pass
    elif method == "tools/list":
        send({"jsonrpc": "2.0", "id": req_id, "result": {"tools": MCP_TOOLS}})
    elif method == "tools/call":
        params = req.get("params", {})
        name = params.get("name", "")
        args = params.get("arguments", {})
        if name == "task":
            try:
                text = call_qwen3(args["task"], args.get("files", []))
                send({"jsonrpc": "2.0", "id": req_id, "result": {
                    "content": [{"type": "text", "text": text}]
                }})
            except Exception as e:
                send({"jsonrpc": "2.0", "id": req_id, "result": {
                    "content": [{"type": "text", "text": f"ERROR: {e}"}],
                    "isError": True,
                }})
        else:
            send({"jsonrpc": "2.0", "id": req_id,
                  "error": {"code": -32601, "message": f"Outil inconnu : {name}"}})
    elif req_id is not None:
        send({"jsonrpc": "2.0", "id": req_id,
              "error": {"code": -32601, "message": f"Méthode inconnue : {method}"}})


# --- Démarrage ---

try:
    _start_aichat()
except Exception as e:
    print(json.dumps({"error": f"Impossible de démarrer aichat : {e}"}),
          file=sys.stderr, flush=True)
    sys.exit(1)

for line in sys.stdin:
    line = line.strip()
    if not line:
        continue
    try:
        handle(json.loads(line))
    except Exception as e:
        print(json.dumps({"error": str(e)}), file=sys.stderr, flush=True)