125 lines
6.0 KiB
Python
Executable File
125 lines
6.0 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""Serveur MCP stdio autonome pour Hermes, sans installation pip.
|
||
|
||
Protocole JSON-RPC 2.0 (MCP legacy 2025-03/2025-06), messages JSONL.
|
||
Le serveur ne gère pas l'horloge ; il délègue à Alfred sur 127.0.0.1:54321.
|
||
"""
|
||
from __future__ import annotations
|
||
import json
|
||
import os
|
||
import sys
|
||
from urllib import request, error
|
||
|
||
URL = os.getenv('ALFRED_TIMERS_API', 'http://127.0.0.1:54321').rstrip('/')
|
||
|
||
|
||
def tool(name, desc, props=None, required=None):
|
||
return {'name': name, 'description': desc,
|
||
'inputSchema': {'type': 'object', 'properties': props or {}, 'required': required or [], 'additionalProperties': False}}
|
||
|
||
|
||
def seconds_prop(desc):
|
||
return {'type': 'number', 'exclusiveMinimum': 0, 'maximum': 2592000, 'description': desc}
|
||
|
||
|
||
def label_prop():
|
||
return {'type': 'string', 'maxLength': 80, 'description': 'Nom facultatif pour reconnaître le minuteur (ex. pâtes, cuisson).'}
|
||
|
||
|
||
def id_prop():
|
||
return {'type': 'string', 'description': "Identifiant ou nom unique renvoyé par list_timers ou create_timer."}
|
||
|
||
|
||
TOOLS = [
|
||
tool('create_timer', 'Crée un minuteur qui compte à rebours. Convertis les minutes/heures en secondes. Ex : 5 minutes = 300 secondes. Une alerte vocale est déclenchée à zéro.', {'seconds': seconds_prop('Durée demandée en secondes'), 'label': label_prop()}, ['seconds']),
|
||
tool('create_stopwatch', 'Démarre un chronomètre qui compte à partir de zéro ; si l’utilisateur dit « chrono de 10 minutes », fournir 600 secondes pour l’arrêter avec une alerte après 10 minutes. Sinon omettre seconds pour un chrono libre.', {'seconds': seconds_prop('Durée limite facultative, en secondes'), 'label': label_prop()}),
|
||
tool('list_timers', 'Liste minuteurs/chronomètres en cours et le temps restant ou écoulé.'),
|
||
tool('cancel_all_timers', 'Annule TOUS les minuteurs et chronomètres en cours sur demande explicite de l’utilisateur.'),
|
||
tool('pause_timer', 'Met en pause le minuteur ou le chronomètre sélectionné.', {'id': id_prop()}, ['id']),
|
||
tool('resume_timer', 'Reprend un minuteur ou chronomètre en pause.', {'id': id_prop()}, ['id']),
|
||
tool('cancel_timer', 'Annule le minuteur ou le chronomètre sélectionné.', {'id': id_prop()}, ['id']),
|
||
tool('add_time', 'Ajoute du temps à un minuteur ou un chrono avec durée limite. Ex : ajoute 2 minutes = 120 secondes.', {'id': id_prop(), 'seconds': seconds_prop('Temps supplémentaire en secondes')}, ['id','seconds']),
|
||
]
|
||
|
||
|
||
def api_call(action=None, args=None):
|
||
if action is None:
|
||
req = request.Request(URL + '/api/timers', headers={'Accept': 'application/json'})
|
||
else:
|
||
data = json.dumps({'action': action, **(args or {})}).encode('utf-8')
|
||
req = request.Request(URL + '/api/timers/action', data=data, method='POST', headers={'Content-Type': 'application/json', 'X-Alfred-MCP-Local': 'timers'})
|
||
try:
|
||
with request.urlopen(req, timeout=7) as response:
|
||
return json.loads(response.read())
|
||
except error.HTTPError as exc:
|
||
try:
|
||
detail = json.loads(exc.read()).get('error', '')
|
||
except Exception:
|
||
detail = ''
|
||
raise ValueError(detail or f'HTTP {exc.code}') from exc
|
||
except (error.URLError, TimeoutError) as exc:
|
||
raise ValueError(f'Alfred indisponible sur {URL} : {exc}') from exc
|
||
|
||
|
||
def invoke(name, arguments):
|
||
if not isinstance(arguments, dict):
|
||
raise ValueError('Arguments JSON attendus.')
|
||
if name == 'list_timers':
|
||
return api_call()
|
||
if name == 'cancel_all_timers':
|
||
return api_call('cancel_all')
|
||
if name in ('create_timer', 'create_stopwatch'):
|
||
return api_call(name, {k: arguments[k] for k in ('seconds','label') if k in arguments})
|
||
action = {'pause_timer':'pause','resume_timer':'resume','cancel_timer':'cancel','add_time':'add_time'}.get(name)
|
||
if action:
|
||
return api_call(action, {k: arguments[k] for k in ('id','seconds') if k in arguments})
|
||
raise ValueError('Outil inconnu : '+str(name))
|
||
|
||
|
||
def respond(msg):
|
||
sys.stdout.write(json.dumps(msg, ensure_ascii=False, separators=(',', ':'))+'\n')
|
||
sys.stdout.flush()
|
||
|
||
|
||
def handle(obj):
|
||
method = obj.get('method')
|
||
if not isinstance(method, str) or 'id' not in obj:
|
||
return
|
||
req_id = obj['id']
|
||
try:
|
||
if method == 'initialize':
|
||
requested = str((obj.get('params') or {}).get('protocolVersion', '2025-03-26'))
|
||
version = requested if requested in ('2025-03-26', '2025-06-18', '2024-11-05') else '2025-06-18'
|
||
result = {'protocolVersion': version, 'capabilities': {'tools': {'listChanged': False}},
|
||
'serverInfo': {'name': 'alfred-timers', 'version': '1.0.0'},
|
||
'instructions': 'Use these tools for all timers, chronometers and time adjustments on Alfred. Never claim a timer was set without tool success.'}
|
||
elif method == 'tools/list':
|
||
result = {'tools': TOOLS}
|
||
elif method == 'tools/call':
|
||
params = obj.get('params') or {}
|
||
try:
|
||
answer = invoke(str(params.get('name','')), params.get('arguments') or {})
|
||
result = {'content': [{'type': 'text', 'text': json.dumps(answer, ensure_ascii=False)}], 'isError': False}
|
||
except Exception as exc:
|
||
result = {'content': [{'type': 'text', 'text': str(exc)}], 'isError': True}
|
||
elif method == 'ping':
|
||
result = {}
|
||
else:
|
||
respond({'jsonrpc':'2.0','id':req_id,'error':{'code':-32601,'message':'Method not found'}})
|
||
return
|
||
respond({'jsonrpc': '2.0', 'id': req_id, 'result': result})
|
||
except Exception as exc:
|
||
respond({'jsonrpc':'2.0','id':req_id,'error':{'code':-32603,'message':str(exc)}})
|
||
|
||
|
||
if __name__ == '__main__':
|
||
for line in sys.stdin:
|
||
if not line.strip():
|
||
continue
|
||
try:
|
||
obj = json.loads(line)
|
||
if isinstance(obj, dict):
|
||
handle(obj)
|
||
except json.JSONDecodeError as exc:
|
||
print('Erreur JSON MCP :', exc, file=sys.stderr)
|