diff --git a/src/app.py b/src/app.py index 03f6cf5..c584b2d 100644 --- a/src/app.py +++ b/src/app.py @@ -149,6 +149,10 @@ from src.roadmap import db as roadmap_db # noqa: E402 roadmap_db.init_schema() +from src.mcp.account import mcp_account_bp # noqa: E402 + +app.server.register_blueprint(mcp_account_bp) + # robots.txt @app.server.route("/robots.txt") diff --git a/src/pages/_compte_shell.py b/src/pages/_compte_shell.py index 74ac17e..2f845f7 100644 --- a/src/pages/_compte_shell.py +++ b/src/pages/_compte_shell.py @@ -17,6 +17,12 @@ SECTIONS = [ "href": "/compte/roadmap", "require_subscription": True, }, + { + "key": "mcp", + "label": "Connecteur MCP", + "href": "/compte/mcp", + "require_subscription": True, + }, { "key": "abonnement", "label": "Abonnement", diff --git a/src/pages/compte/mcp.py b/src/pages/compte/mcp.py new file mode 100644 index 0000000..df18007 --- /dev/null +++ b/src/pages/compte/mcp.py @@ -0,0 +1,205 @@ +import os + +import dash_bootstrap_components as dbc +from dash import dcc, html, register_page +from flask import session +from flask_login import current_user + +from src.api import tokens_db +from src.pages._compte_shell import account_guard, account_shell + +register_page( + __name__, + path="/compte/mcp", + title="Connecteur MCP | colibre", + name="Connecteur MCP", + description="Connectez votre agent IA aux données colibre via le protocole MCP.", +) + +MCP_ENABLED = os.getenv("DASH_MCP_ENABLED") == "true" +_TOKEN_PLACEHOLDER = "" + + +def _mcp_url() -> str: + base = os.getenv("APP_BASE_URL", "").rstrip("/") + return f"{base}/_mcp" if base else "/_mcp" + + +def _csrf(index: str): + return dcc.Input( + type="hidden", + id={"type": "csrf-input", "index": index}, + name="csrf_token", + ) + + +def client_instructions(url: str, token: str): + """Construit les blocs d'instructions par client (fonction pure, testable).""" + claude = f'claude mcp add colibre --transport http {url} --header "Authorization: Bearer {token}"' + gemini = f'gemini mcp add --transport http --header "Authorization: Bearer {token}" colibre {url}' + return dbc.Accordion( + start_collapsed=True, + always_open=False, + children=[ + dbc.AccordionItem( + title="Claude (Code / Desktop)", + children=html.Pre(html.Code(claude)), + ), + dbc.AccordionItem( + title="Gemini CLI", + children=[ + html.Pre(html.Code(gemini)), + html.P( + "Ou dans ~/.gemini/settings.json : un serveur mcpServers.colibre " + f'avec "httpUrl": "{url}" et "headers": ' + f'{{"Authorization": "Bearer {token}"}}.' + ), + ], + ), + dbc.AccordionItem( + title="Mistral Le Chat", + children=html.P( + "Dans les connecteurs MCP, ajoutez un serveur HTTP d'URL " + f"{url}, authentification « API Token », en-tête " + f"« Authorization » = « Bearer {token} »." + ), + ), + dbc.AccordionItem( + title="ChatGPT", + children=[ + html.P( + "L'app ChatGPT grand public exige le flux OAuth 2.1 et " + "n'accepte pas de jeton statique. En attendant le connecteur " + "OAuth (itération future), utilisez la voie développeur " + "(API / Agents SDK OpenAI), qui accepte un serveur MCP distant " + f"d'URL {url} avec l'en-tête « Authorization: Bearer {token} »." + ), + ], + ), + ], + ) + + +def _token_row(row: dict): + statut = "Révoqué" if row["revoked_at"] else "Actif" + actions = [] + if not row["revoked_at"]: + actions = [ + html.Form( + method="POST", + action=f"/compte/mcp/revoquer/{row['id']}", + children=[ + _csrf(f"revoke-{row['id']}"), + dbc.Button( + "Révoquer", + type="submit", + color="danger", + size="sm", + outline=True, + ), + ], + ) + ] + return html.Tr( + [ + html.Td(row["label"]), + html.Td(row["created_at"]), + html.Td(row["last_used_at"] or "—"), + html.Td(statut), + html.Td(actions), + ] + ) + + +def layout(**_): + guard = account_guard("/compte/mcp", require_subscription=True) + if guard is not None: + return guard + + if not MCP_ENABLED: + contenu = html.Div( + [ + html.H2("Connecteur MCP"), + dbc.Alert( + "Le connecteur MCP sera bientôt disponible. Revenez prochainement.", + color="info", + ), + ] + ) + return account_shell("mcp", contenu) + + url = _mcp_url() + new_token = session.pop("mcp_new_token", None) + + alerts = [] + if new_token: + alerts.append( + dbc.Alert( + [ + html.Strong("Votre nouveau jeton (copiez-le maintenant, "), + html.Strong("il ne sera plus affiché) :"), + html.Pre(html.Code(new_token)), + ], + color="success", + ) + ) + + tokens = tokens_db.list_user_tokens( + os.environ["USERS_DB_PATH"], current_user.id, "mcp" + ) + table = ( + dbc.Table( + [ + html.Thead( + html.Tr( + [ + html.Th("Nom"), + html.Th("Créé le"), + html.Th("Dernière utilisation"), + html.Th("Statut"), + html.Th(""), + ] + ) + ), + html.Tbody([_token_row(dict(r)) for r in tokens]), + ], + striped=True, + bordered=False, + hover=True, + ) + if tokens + else html.P("Aucun jeton pour le moment.") + ) + + create_form = html.Form( + method="POST", + action="/compte/mcp/creer", + className="mb-4", + children=[ + _csrf("mcp-create"), + dbc.Label("Nom du jeton (ex. « Claude sur mon portable »)"), + dbc.Input(type="text", name="label", required=True, className="mb-2"), + dbc.Button("Générer un jeton", type="submit", color="primary"), + ], + ) + + snippet_token = new_token or _TOKEN_PLACEHOLDER + contenu = html.Div( + [ + html.H2("Connecteur MCP"), + html.P( + "Générez un jeton pour connecter votre agent IA (Claude, Gemini, " + "Mistral…) aux données colibre via le protocole MCP. Le jeton vaut " + "votre identité : gardez-le secret. Un abonnement actif est requis." + ), + *alerts, + html.H4("Générer un jeton", className="mt-3"), + create_form, + html.H4("Mes jetons", className="mt-4"), + table, + html.H4("Connecter un client", className="mt-4"), + html.P(f"URL du serveur MCP : {url}"), + client_instructions(url, snippet_token), + ] + ) + return account_shell("mcp", contenu) diff --git a/tests/mcp/test_account_page.py b/tests/mcp/test_account_page.py new file mode 100644 index 0000000..5708325 --- /dev/null +++ b/tests/mcp/test_account_page.py @@ -0,0 +1,26 @@ +def test_section_connecteur_mcp_present(): + from src.pages._compte_shell import SECTIONS + + keys = {s["key"]: s for s in SECTIONS} + assert "mcp" in keys + assert keys["mcp"]["label"] == "Connecteur MCP" + assert keys["mcp"]["href"] == "/compte/mcp" + assert keys["mcp"]["require_subscription"] is True + + +def test_page_module_registers_and_builds_client_instructions(): + # importe l'app pour la découverte use_pages en contexte propre + from src.app import app # noqa: F401 + from src.pages.compte import mcp as mcp_page + + # helper pur : construit les 4 blocs d'instructions clients + blocks = mcp_page.client_instructions( + "https://colibre.fr/_mcp", "colibre_TESTTOKEN" + ) + text = str(blocks) + assert "colibre_TESTTOKEN" in text + assert "https://colibre.fr/_mcp" in text + for client_name in ("Claude", "Gemini", "Mistral", "ChatGPT"): + assert client_name in text + # caveat ChatGPT (OAuth / itération future) + assert "OAuth" in text