main.py•7.34 kB
import os
import json
import requests
from datetime import datetime
from flask import Flask, request, jsonify
from functools import wraps
app = Flask(__name__)
# Configuration
LS_API_URL = os.getenv('LS_API_URL')
LS_USERNAME = os.getenv('LS_USERNAME')
LS_PASSWORD = os.getenv('LS_PASSWORD')
MCP_VERSION = "2025-06-18"
# Cache de session
session_cache = {'key': None, 'expiry': None}
def get_session_key():
"""Obtenir une clé de session valide"""
now = datetime.now()
if session_cache['key'] and session_cache['expiry'] and now < session_cache['expiry']:
return session_cache['key']
payload = {
"method": "get_session_key",
"params": [LS_USERNAME, LS_PASSWORD],
"id": 1
}
try:
response = requests.post(LS_API_URL, json=payload).json()
if 'result' in response:
session_cache['key'] = response['result']
# Expire dans 55 minutes (3300 secondes)
session_cache['expiry'] = datetime.fromtimestamp(datetime.now().timestamp() + 3300)
return session_cache['key']
else:
raise Exception(f"Erreur d'authentification: {response.get('error', 'Unknown error')}")
except Exception as e:
raise Exception(f"Échec de la connexion à l'API: {str(e)}")
def release_session_key(session_key):
"""Libérer une clé de session"""
payload = {
"method": "release_session_key",
"params": [session_key],
"id": 1
}
requests.post(LS_API_URL, json=payload)
def mcp_endpoint(expected_params=None):
"""Décorateur pour les endpoints MCP"""
def decorator(f):
@wraps(f)
def wrapper(*args, **kwargs):
# Validation du format MCP
if not request.json or 'model' not in request.json or 'action' not in request.json:
return jsonify({
"type": "error",
"content": {
"message": "Format MCP invalide",
"code": "MCP_FORMAT_ERROR"
}
}), 400
# Exécution de la fonction
try:
result = f(*args, **kwargs)
return jsonify({
"type": "success",
"content": result
})
except Exception as e:
return jsonify({
"type": "error",
"content": {
"message": str(e),
"code": "LS_API_ERROR"
}
}), 500
return wrapper
return decorator
@app.route('/mcp', methods=['POST'])
@mcp_endpoint()
def handle_mcp_request():
"""Gestionnaire principal des requêtes MCP"""
data = request.json
action = data['action']
params = data.get('parameters', {})
# Mapping des actions aux méthodes
actions = {
'create_survey': create_survey,
'add_question': add_question,
'activate_survey': activate_survey,
'list_surveys': list_surveys,
'export_responses': export_responses
}
if action not in actions:
raise Exception(f"Action non supportée: {action}")
return actions[action](params)
def call_limesurvey_api(method, params):
"""Appel générique à l'API LimeSurvey"""
session_key = get_session_key()
payload = {
"method": method,
"params": [session_key] + params,
"id": 1
}
response = requests.post(LS_API_URL, json=payload).json()
if 'error' in response and response['error']:
raise Exception(f"Erreur API: {response['error']}")
return response.get('result', {})
def create_survey(params):
"""Créer un nouveau questionnaire"""
required = ['survey_title', 'survey_language']
if not all(k in params for k in required):
raise Exception("Paramètres manquants: survey_title, survey_language")
survey_id = params.get('survey_id', int(datetime.now().timestamp()))
result = call_limesurvey_api('add_survey', [
survey_id,
params['survey_title'],
params['survey_language']
])
return {
"survey_id": survey_id,
"status": "created",
"details": result
}
def add_question(params):
"""Ajouter une question à un questionnaire"""
required = ['survey_id', 'group_id', 'question_text', 'question_type']
if not all(k in params for k in required):
raise Exception("Paramètres manquants: survey_id, group_id, question_text, question_type")
result = call_limesurvey_api('add_question', [
params['survey_id'],
params['group_id'],
params['question_text'],
params['question_type'],
params.get('question_options', {})
])
return {
"question_id": result['qid'],
"survey_id": params['survey_id'],
"status": "added"
}
def activate_survey(params):
"""Activer un questionnaire"""
if 'survey_id' not in params:
raise Exception("Paramètre survey_id manquant")
result = call_limesurvey_api('activate_survey', [params['survey_id']])
return {"status": "activated" if result else "failed"}
def list_surveys(params):
"""Lister les questionnaires"""
result = call_limesurvey_api('list_surveys', [])
return {"surveys": result}
def export_responses(params):
"""Exporter les réponses"""
required = ['survey_id']
if not all(k in params for k in required):
raise Exception("Paramètre survey_id manquant")
result = call_limesurvey_api('export_responses', [
params['survey_id'],
params.get('format', 'json'),
params.get('language', 'fr')
])
return {"responses": json.loads(result)}
@app.route('/spec', methods=['GET'])
def get_mcp_spec():
"""Endpoint pour la spécification MCP"""
return jsonify({
"version": MCP_VERSION,
"name": "LimeSurvey MCP Server",
"description": "Serveur MCP pour l'API LimeSurvey",
"endpoints": [
{
"name": "create_survey",
"description": "Créer un nouveau questionnaire",
"required_params": ["survey_title", "survey_language"],
"optional_params": ["survey_id"]
},
{
"name": "add_question",
"description": "Ajouter une question à un questionnaire",
"required_params": ["survey_id", "group_id", "question_text", "question_type"],
"optional_params": ["question_options"]
},
{
"name": "activate_survey",
"description": "Activer un questionnaire",
"required_params": ["survey_id"]
},
{
"name": "list_surveys",
"description": "Lister les questionnaires"
},
{
"name": "export_responses",
"description": "Exporter les réponses",
"required_params": ["survey_id"],
"optional_params": ["format", "language"]
}
]
})
if __name__ == '__main__':
port = int(os.getenv('MCP_PORT', 8080))
app.run(host='0.0.0.0', port=port, debug=True)