255 lines
10 KiB
Python
255 lines
10 KiB
Python
|
#!/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)
|