#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ mcp_sliding.py — Serveur MCP Sliding Automation · Chantier C8 ============================================================= Même squelette que mcp_server.py (Context Continuity) : Starlette pur, JSON-RPC 2.0, Bearer token, compatible Python 3.9. Transport : Streamable HTTP sur /mcp — port 8767 (Trilium : 8766). Auth : mêmes variables .env (API_KEY_CLAUDE / API_KEY_LECHAT). Particularité vs Trilium : le tool get_slide_image renvoie un content de type IMAGE (base64) — Claude/Le Chat affichent la slide dans la conversation et peuvent la critiquer. Usage : python3 mcp_sliding.py → écoute sur 127.0.0.1:8767 Reverse proxy DSM : mcp-sliding.bertha-cloud.fr → 127.0.0.1:8767 """ import json import os from dotenv import load_dotenv from starlette.applications import Starlette from starlette.requests import Request from starlette.responses import JSONResponse, Response from starlette.routing import Route import sliding_api as api from mcp_oauth import OAuthProvider load_dotenv() API_KEY_CLAUDE = os.getenv("API_KEY_CLAUDE", "") API_KEY_LECHAT = os.getenv("API_KEY_LECHAT", "") VALID_TOKENS = {k: v for k, v in ((API_KEY_CLAUDE, "Claude"), (API_KEY_LECHAT, "LeChat")) if k} # OAuth 2.1 (Claude Desktop) — module partagé, état dédié à Sliding oauth = OAuthProvider( base_url=os.getenv("OAUTH_BASE_URL", "https://mcp-sliding.bertha-cloud.fr"), state_file=os.path.expanduser( "~/App/Sliding/python-pptx/oauth_state.json"), security_word=os.getenv("SECURITY_WORD", ""), realm="Sliding Automation") PROTOCOL_VERSION = "2025-06-18" SERVER_INFO = {"name": "Sliding Automation", "version": "1.0.0"} # --------------------------------------------------------------------------- # Tools # --------------------------------------------------------------------------- TOOLS = [ {"name": "list_projects", "description": "Liste les projets Sliding existants (slug, plan " "présent, dernier PPTX).", "inputSchema": {"type": "object", "properties": {}, "required": []}}, {"name": "create_project", "description": "Crée un projet Sliding (dossiers inputs/outputs/" "assets).", "inputSchema": {"type": "object", "properties": { "nom": {"type": "string"}}, "required": ["nom"]}}, {"name": "get_project", "description": "État complet d'un projet : plan compact, derniers " "artefacts, documents, assets, journal récent. À " "appeler en début de session pour reprendre le " "travail exactement où il en était.", "inputSchema": {"type": "object", "properties": { "slug": {"type": "string"}}, "required": ["slug"]}}, {"name": "set_plan", "description": "Persiste le plan compact du projet (source de " "vérité, rechargée à l'identique). À appeler après " "chaque évolution structurelle validée du plan.", "inputSchema": {"type": "object", "properties": { "slug": {"type": "string"}, "plan_compact": {"type": "string"}}, "required": ["slug", "plan_compact"]}}, {"name": "add_document", "description": "Dépose un document (brief, notes) dans inputs/ du " "projet. encoding: text (défaut) ou base64.", "inputSchema": {"type": "object", "properties": { "slug": {"type": "string"}, "filename": {"type": "string"}, "content": {"type": "string"}, "encoding": {"type": "string", "enum": ["text", "base64"]}}, "required": ["slug", "filename", "content"]}}, {"name": "list_assets", "description": "Images disponibles dans assets/ du projet (pour " "les layouts image_split/image_full et le freeform).", "inputSchema": {"type": "object", "properties": { "slug": {"type": "string"}}, "required": ["slug"]}}, {"name": "formalise", "description": "Chaîne complète : Designer (choix des layouts) → " "Encoder structuré → validation → rendu PPTX. " "plan_markdown = la formalisation slide par slide " "produite dans la conversation (titres + contenus). " "Long (1-3 min).", "inputSchema": {"type": "object", "properties": { "slug": {"type": "string"}, "plan_markdown": {"type": "string"}}, "required": ["slug", "plan_markdown"]}}, {"name": "revise_slides", "description": "Révision ciblée : retravaille les slides indiquées" " puis fusionne et re-rend le deck COMPLET.", "inputSchema": {"type": "object", "properties": { "slug": {"type": "string"}, "positions": {"type": "array", "items": {"type": "integer"}}, "instruction": {"type": "string"}}, "required": ["slug", "positions", "instruction"]}}, {"name": "render_yaml", "description": "Rendu direct d'un YAML complet fourni (équivalent " "--render), pour les cas avancés/freeform.", "inputSchema": {"type": "object", "properties": { "slug": {"type": "string"}, "yaml": {"type": "string"}}, "required": ["slug", "yaml"]}}, {"name": "preview", "description": "Génère les aperçus PNG du deck (bloquant, " "quelques minutes sur le NAS). Retourne la liste " "des slides ; utiliser ensuite get_slide_image.", "inputSchema": {"type": "object", "properties": { "slug": {"type": "string"}, "pptx": {"type": "string", "description": "Chemin optionnel (défaut : dernier " "PPTX du projet)"}}, "required": ["slug"]}}, {"name": "get_slide_image", "description": "Renvoie l'image PNG d'une slide (après preview) — " "affichée directement dans la conversation pour " "critique visuelle.", "inputSchema": {"type": "object", "properties": { "slug": {"type": "string"}, "position": {"type": "integer"}, "pptx": {"type": "string"}}, "required": ["slug", "position"]}}, ] DISPATCH = { "list_projects": lambda: api.list_projects(), "create_project": api.create_project, "get_project": api.get_project, "set_plan": api.set_plan, "add_document": api.add_document, "list_assets": api.list_project_assets, "formalise": api.formalise, "revise_slides": lambda slug, positions, instruction: api.revise(slug, positions, instruction), "render_yaml": lambda slug, yaml: api.render_yaml(slug, yaml), "preview": api.preview, "get_slide_image": api.get_slide_image, } # --------------------------------------------------------------------------- # Handler JSON-RPC MCP # --------------------------------------------------------------------------- def check_auth(request: Request): auth = request.headers.get("authorization", "") token = auth.replace("Bearer ", "").strip() # 1) Tokens statiques (Le Chat, clé Claude historique) if token in VALID_TOKENS: return VALID_TOKENS[token] # 2) Tokens OAuth émis (Claude Desktop) if oauth.check_token(token): return "Claude-OAuth" return None def _result_content(tool_name, result): """Contenu MCP : image pour get_slide_image, texte JSON sinon.""" if (tool_name == "get_slide_image" and isinstance(result, dict) and result.get("b64")): return [ {"type": "image", "data": result["b64"], "mimeType": result.get("mime", "image/png")}, {"type": "text", "text": json.dumps({"filename": result["filename"]}, ensure_ascii=False)}, ] return [{"type": "text", "text": json.dumps(result, ensure_ascii=False, indent=2)}] async def mcp_endpoint(request: Request): 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", {}) if method == "initialize": return JSONResponse({ "jsonrpc": "2.0", "id": req_id, "result": {"protocolVersion": PROTOCOL_VERSION, "capabilities": {"tools": {}}, "serverInfo": SERVER_INFO}}) if method == "notifications/initialized": return Response(status_code=202) if method == "tools/list": return JSONResponse({"jsonrpc": "2.0", "id": req_id, "result": {"tools": TOOLS}}) 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": f"Tool inconnu : {tool_name}"}}) try: result = fn(**args) return JSONResponse({ "jsonrpc": "2.0", "id": req_id, "result": {"content": _result_content(tool_name, result)}}) except Exception as e: return JSONResponse({ "jsonrpc": "2.0", "id": req_id, "error": {"code": -32603, "message": str(e)}}) return JSONResponse({ "jsonrpc": "2.0", "id": req_id, "error": {"code": -32601, "message": f"Méthode inconnue : {method}"}}) async def health(request: Request): return JSONResponse({"status": "ok", "server": "mcp-sliding", "version": "1.0.0"}) app = Starlette(routes=[ Route("/mcp", mcp_endpoint, methods=["POST"]), Route("/", mcp_endpoint, methods=["POST"]), Route("/health", health, methods=["GET"]), ] + oauth.routes()) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=8767)