320 lines
14 KiB
Python
320 lines
14 KiB
Python
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"""<!DOCTYPE html>
|
|
<html lang="fr">
|
|
<head>
|
|
<meta charset="UTF-8">
|
|
<title>{title}</title>
|
|
<style>
|
|
@page {{ margin: 2.5cm 2cm; }}
|
|
body {{ font-family: "Liberation Serif", Georgia, serif; font-size: 11pt; line-height: 1.7; color: #222; }}
|
|
h1 {{ font-size: 16pt; text-align: center; text-transform: uppercase; letter-spacing: 1px; border-bottom: 2px solid #222; padding-bottom: 8px; margin-bottom: 20px; }}
|
|
h2 {{ font-size: 12pt; text-transform: uppercase; margin-top: 24px; margin-bottom: 8px; border-bottom: 1px solid #aaa; padding-bottom: 4px; }}
|
|
h3 {{ font-size: 11pt; font-weight: bold; margin-top: 16px; }}
|
|
p {{ margin: 8px 0; text-align: justify; }}
|
|
table {{ border-collapse: collapse; width: 100%; margin: 16px 0; }}
|
|
th, td {{ border: 1px solid #999; padding: 8px 12px; text-align: left; vertical-align: top; }}
|
|
th {{ background-color: #f0f0f0; font-weight: bold; }}
|
|
hr {{ border: none; border-top: 1px solid #ccc; margin: 20px 0; }}
|
|
strong {{ font-weight: bold; }}
|
|
em {{ font-style: italic; }}
|
|
ol, ul {{ margin: 8px 0; padding-left: 24px; }}
|
|
li {{ margin: 4px 0; }}
|
|
</style>
|
|
</head>
|
|
<body>
|
|
{html_body}
|
|
</body>
|
|
</html>"""
|
|
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="*")
|