server.py

#!/usr/bin/env python3
"""
Francis - MCP Server spécialisé

Serveur MCP (Model Context Protocol) qui expose des outils aux LLM.
Communication via JSON-RPC sur stdin/stdout.
"""

import sys
import json
from typing import Any

# Registre des outils disponibles
TOOLS: dict[str, dict] = {}


def register_tool(name: str, description: str, parameters: dict):
    """Décorateur pour enregistrer un outil."""
    def decorator(func):
        TOOLS[name] = {
            "name": name,
            "description": description,
            "inputSchema": {
                "type": "object",
                "properties": parameters,
                "required": list(parameters.keys()),
            },
            "handler": func,
        }
        return func
    return decorator


# === OUTILS ===

@register_tool(
    name="ping",
    description="Test de connexion - répond pong",
    parameters={}
)
def tool_ping(**kwargs) -> str:
    return "pong"


@register_tool(
    name="echo",
    description="Répète le message",
    parameters={
        "message": {"type": "string", "description": "Message à répéter"}
    }
)
def tool_echo(message: str, **kwargs) -> str:
    return f"Francis dit: {message}"


@register_tool(
    name="stm32_info",
    description="Retourne des infos sur un MCU STM32",
    parameters={
        "mcu": {"type": "string", "description": "Référence du MCU (ex: STM32F411RE)"}
    }
)
def tool_stm32_info(mcu: str, **kwargs) -> str:
    # TODO: Implémenter vraiment avec parsing des datasheets
    mcu = mcu.upper()

    # Infos basiques hardcodées pour la démo
    known_mcus = {
        "STM32F411RE": {
            "core": "Cortex-M4",
            "flash": "512 KB",
            "ram": "128 KB",
            "freq": "100 MHz",
            "package": "LQFP64",
        },
        "STM32F103C8": {
            "core": "Cortex-M3",
            "flash": "64 KB",
            "ram": "20 KB",
            "freq": "72 MHz",
            "package": "LQFP48",
        },
    }

    if mcu in known_mcus:
        info = known_mcus[mcu]
        return f"{mcu}: {info['core']}, {info['flash']} Flash, {info['ram']} RAM, {info['freq']} max"
    else:
        return f"MCU {mcu} non trouvé dans la base. Ajoute-le à tool_stm32_info()."


# === PROTOCOLE MCP ===

def handle_initialize(params: dict) -> dict:
    """Répond à la requête d'initialisation."""
    return {
        "protocolVersion": "2024-11-05",
        "capabilities": {
            "tools": {},
        },
        "serverInfo": {
            "name": "francis",
            "version": "0.1.0",
        },
    }


def handle_tools_list(params: dict) -> dict:
    """Liste les outils disponibles."""
    tools_list = []
    for name, tool in TOOLS.items():
        tools_list.append({
            "name": tool["name"],
            "description": tool["description"],
            "inputSchema": tool["inputSchema"],
        })
    return {"tools": tools_list}


def handle_tools_call(params: dict) -> dict:
    """Exécute un outil."""
    tool_name = params.get("name")
    arguments = params.get("arguments", {})

    if tool_name not in TOOLS:
        return {
            "content": [{"type": "text", "text": f"Outil inconnu: {tool_name}"}],
            "isError": True,
        }

    try:
        handler = TOOLS[tool_name]["handler"]
        result = handler(**arguments)
        return {
            "content": [{"type": "text", "text": str(result)}],
            "isError": False,
        }
    except Exception as e:
        return {
            "content": [{"type": "text", "text": f"Erreur: {str(e)}"}],
            "isError": True,
        }


def handle_request(request: dict) -> dict | None:
    """Traite une requête JSON-RPC."""
    method = request.get("method", "")
    params = request.get("params", {})
    req_id = request.get("id")

    # Notifications (pas de réponse attendue)
    if method == "notifications/initialized":
        return None

    # Méthodes avec réponse
    handlers = {
        "initialize": handle_initialize,
        "tools/list": handle_tools_list,
        "tools/call": handle_tools_call,
    }

    if method in handlers:
        result = handlers[method](params)
        return {
            "jsonrpc": "2.0",
            "id": req_id,
            "result": result,
        }
    else:
        return {
            "jsonrpc": "2.0",
            "id": req_id,
            "error": {
                "code": -32601,
                "message": f"Méthode inconnue: {method}",
            },
        }


def main():
    """Boucle principale - lit stdin, répond sur stdout."""
    # Log sur stderr (pas stdout qui est réservé au protocole)
    print("Francis MCP Server démarré", file=sys.stderr)
    print(f"Outils disponibles: {', '.join(TOOLS.keys())}", file=sys.stderr)

    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue

        try:
            request = json.loads(line)
            response = handle_request(request)

            if response is not None:
                print(json.dumps(response), flush=True)

        except json.JSONDecodeError as e:
            print(f"Erreur JSON: {e}", file=sys.stderr)
        except Exception as e:
            print(f"Erreur: {e}", file=sys.stderr)


if __name__ == "__main__":
    main()