import os import uuid import time import base64 import httpx import markdown as md_lib import weasyprint from mcp.server.fastmcp import FastMCP from mcp.server.transport_security import TransportSecuritySettings from starlette.requests import Request from starlette.responses import JSONResponse, RedirectResponse import uvicorn # ───────────────────────────────────────────── # Configuration # ───────────────────────────────────────────── YOUSIGN_API_URL = "https://api-sandbox.yousign.app/v3" # En production : https://api.yousign.app/v3 YOUSIGN_API_KEY = os.environ.get("YOUSIGN_API_KEY") registered_clients: dict = {} issued_tokens: dict = {} # ───────────────────────────────────────────── # Serveur MCP # ───────────────────────────────────────────── mcp = FastMCP( "yousign-mcp", stateless_http=True, transport_security=TransportSecuritySettings( enable_dns_rebinding_protection=False ), ) # ───────────────────────────────────────────── # OAuth 2.0 # ───────────────────────────────────────────── @mcp.custom_route("/.well-known/oauth-authorization-server", methods=["GET"]) async def oauth_metadata(request: Request) -> JSONResponse: public_host = os.environ.get("PUBLIC_HOST", "") base = f"https://{public_host}" if public_host else str(request.base_url).rstrip("/") return JSONResponse({ "issuer": base, "authorization_endpoint": f"{base}/oauth/authorize", "token_endpoint": f"{base}/oauth/token", "registration_endpoint": f"{base}/oauth/register", "response_types_supported": ["code"], "grant_types_supported": ["authorization_code", "client_credentials", "refresh_token"], "code_challenge_methods_supported": ["S256"], "token_endpoint_auth_methods_supported": ["none"], }) @mcp.custom_route("/oauth/register", methods=["POST"]) async def oauth_register(request: Request) -> JSONResponse: body = await request.json() client_id = str(uuid.uuid4()) client_secret = str(uuid.uuid4()) registered_clients[client_id] = { "client_secret": client_secret, "redirect_uris": body.get("redirect_uris", []), "created_at": int(time.time()), } return JSONResponse({ "client_id": client_id, "client_secret": client_secret, "client_id_issued_at": int(time.time()), "redirect_uris": body.get("redirect_uris", []), "token_endpoint_auth_method": "none", "grant_types": ["authorization_code", "refresh_token"], "response_types": ["code"], }, status_code=201) @mcp.custom_route("/oauth/authorize", methods=["GET"]) async def oauth_authorize(request: Request) -> RedirectResponse: redirect_uri = request.query_params.get("redirect_uri", "") state = request.query_params.get("state", "") code = str(uuid.uuid4()) separator = "&" if "?" in redirect_uri else "?" return RedirectResponse(url=f"{redirect_uri}{separator}code={code}&state={state}", status_code=302) @mcp.custom_route("/oauth/token", methods=["POST"]) async def oauth_token(request: Request) -> JSONResponse: token = "internal-" + str(uuid.uuid4()).replace("-", "") issued_tokens[token] = {"issued_at": int(time.time())} return JSONResponse({"access_token": token, "token_type": "bearer", "expires_in": 3600, "scope": ""}) @mcp.custom_route("/health", methods=["GET"]) async def health(request: Request) -> JSONResponse: return JSONResponse({"status": "ok"}) # ───────────────────────────────────────────── # Helper : Markdown → PDF bytes # ───────────────────────────────────────────── def _markdown_to_pdf_bytes(markdown_text: str, title: str) -> bytes: converter = md_lib.Markdown(extensions=["extra", "tables"]) html_body = converter.convert(markdown_text) html_doc = f""" {title} {html_body} """ return weasyprint.HTML(string=html_doc).write_pdf() # ───────────────────────────────────────────── # Helper : compter les pages d'un PDF # ───────────────────────────────────────────── def _count_pdf_pages(pdf_bytes: bytes) -> int: """Compte le nombre de pages dans un PDF en cherchant /Type /Page.""" count = pdf_bytes.count(b"/Type /Page") if count == 0: count = pdf_bytes.count(b"/Type/Page") return max(count, 1) # ───────────────────────────────────────────── # OUTIL 1 : Conversion Markdown → PDF base64 # ───────────────────────────────────────────── @mcp.tool() async def markdown_to_pdf_base64(markdown_text: str, title: str = "NDA - Accord de Confidentialité") -> str: """ Convertit un texte Markdown en PDF encodé en base64. À utiliser pour préparer le NDA avant envoi à Yousign. Args: markdown_text: Le contenu du NDA en Markdown (avec les variables déjà remplacées) title: Titre du document PDF """ pdf_bytes = _markdown_to_pdf_bytes(markdown_text, title) return base64.b64encode(pdf_bytes).decode("utf-8") # ───────────────────────────────────────────── # OUTIL 2 : Envoi NDA Markdown à Yousign (tout-en-un) # ───────────────────────────────────────────── @mcp.tool() async def yousign_create_signature_request( document_markdown: str, document_name: str, signer_client_first_name: str, signer_client_last_name: str, signer_client_email: str, signer_internal_first_name: str = "Pierre-Yves", signer_internal_last_name: str = "Tachon", signer_internal_email: str = "pierre-yves.tachon@seizeconsulting.com", document_title: str = "NDA - Accord de Confidentialité", ) -> str: """ Convertit un NDA en Markdown en PDF puis crée une demande de signature Yousign avec deux signataires : le client et le signataire interne Seize Consulting. Tout est fait en une seule étape sans passer par le base64. Args: document_markdown: Le contenu du NDA en Markdown (avec les variables déjà remplacées) document_name: Nom du fichier PDF, ex: NDA_ClientX.pdf signer_client_first_name: Prénom du signataire client signer_client_last_name: Nom du signataire client signer_client_email: Email du signataire client signer_internal_first_name: Prénom du signataire interne (défaut: Pierre-Yves) signer_internal_last_name: Nom du signataire interne (défaut: Tachon) signer_internal_email: Email du signataire interne (défaut: pierre-yves.tachon@seizeconsulting.com) document_title: Titre du document PDF (défaut: NDA - Accord de Confidentialité) """ # Étape 0 : Convertir Markdown → PDF en mémoire pdf_bytes = _markdown_to_pdf_bytes(document_markdown, document_title) last_page = _count_pdf_pages(pdf_bytes) auth_headers = {"Authorization": f"Bearer {YOUSIGN_API_KEY}"} json_headers = {**auth_headers, "Content-Type": "application/json"} async with httpx.AsyncClient(timeout=60.0) as client: # Étape 1 : Créer la signature request avec les signataires sr_payload = { "name": document_name.replace(".pdf", ""), "delivery_mode": "email", "signers": [ { "info": { "first_name": signer_client_first_name, "last_name": signer_client_last_name, "email": signer_client_email, "locale": "fr", }, "signature_level": "electronic_signature", "signature_authentication_mode": "no_otp", }, { "info": { "first_name": signer_internal_first_name, "last_name": signer_internal_last_name, "email": signer_internal_email, "locale": "fr", }, "signature_level": "electronic_signature", "signature_authentication_mode": "no_otp", }, ], } sr_res = await client.post( f"{YOUSIGN_API_URL}/signature_requests", headers=json_headers, json=sr_payload, ) if sr_res.status_code not in (200, 201): return f"Erreur création signature request (HTTP {sr_res.status_code}): {sr_res.text}" sr = sr_res.json() sr_id = sr["id"] signer_client_id = sr["signers"][0]["id"] signer_internal_id = sr["signers"][1]["id"] # Étape 2 : Upload du PDF en multipart/form-data (bytes directs) doc_res = await client.post( f"{YOUSIGN_API_URL}/signature_requests/{sr_id}/documents", headers=auth_headers, data={"nature": "signable_document"}, files={"file": (document_name, pdf_bytes, "application/pdf")}, ) if doc_res.status_code not in (200, 201): return f"Erreur upload document (HTTP {doc_res.status_code}): {doc_res.text}" doc_id = doc_res.json()["id"] # Étape 3 : Ajouter les champs de signature (un par signataire, dernière page) # Signataire client : colonne gauche field_client = { "type": "signature", "page": last_page, "x": 35, "y": 650, "width": 200, "height": 50, "signer_id": signer_client_id, } fc_res = await client.post( f"{YOUSIGN_API_URL}/signature_requests/{sr_id}/documents/{doc_id}/fields", headers=json_headers, json=field_client, ) if fc_res.status_code not in (200, 201): return f"Erreur ajout champ signature client (HTTP {fc_res.status_code}): {fc_res.text}" # Signataire interne : colonne droite field_internal = { "type": "signature", "page": last_page, "x": 310, "y": 650, "width": 200, "height": 50, "signer_id": signer_internal_id, } fi_res = await client.post( f"{YOUSIGN_API_URL}/signature_requests/{sr_id}/documents/{doc_id}/fields", headers=json_headers, json=field_internal, ) if fi_res.status_code not in (200, 201): return f"Erreur ajout champ signature interne (HTTP {fi_res.status_code}): {fi_res.text}" # Étape 4 : Activer la demande de signature activate_res = await client.post( f"{YOUSIGN_API_URL}/signature_requests/{sr_id}/activate", headers=json_headers, ) if activate_res.status_code not in (200, 201): return f"Erreur activation (HTTP {activate_res.status_code}): {activate_res.text}" activated = activate_res.json() return ( f"Demande de signature créée et envoyée avec succès !\n" f"- ID Yousign : {sr_id}\n" f"- Statut : {activated.get('status', 'N/A')}\n" f"- Pages du PDF : {last_page}\n" f"- Email client : {signer_client_email}\n" f"- Email interne : {signer_internal_email}" ) # ───────────────────────────────────────────── # Point d'entrée # ───────────────────────────────────────────── if __name__ == "__main__": port = int(os.environ.get("PORT", 3000)) app = mcp.streamable_http_app() uvicorn.run(app, host="0.0.0.0", port=port, proxy_headers=True, forwarded_allow_ips="*")