Files
Alfredv2/hermes/MCP_wazuh/vikunja_mcp.py.bak
T
2026-10-26 00:06:24 +01:00

328 lines
7.5 KiB
Python

#!/usr/bin/env python3
import os
from pathlib import Path
import httpx
from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP
# =============================================================================
# Configuration
# =============================================================================
BASE_DIR = Path(__file__).resolve().parent
load_dotenv(BASE_DIR / ".env")
VIKUNJA_URL = os.getenv("VIKUNJA_URL", "").rstrip("/")
VIKUNJA_TOKEN = os.getenv("VIKUNJA_TOKEN", "")
VIKUNJA_VERIFY_SSL = (
os.getenv("VIKUNJA_VERIFY_SSL", "true").lower()
in ("true", "1", "yes")
)
if not VIKUNJA_URL:
raise RuntimeError("VIKUNJA_URL n'est pas défini dans .env")
if not VIKUNJA_TOKEN:
raise RuntimeError("VIKUNJA_TOKEN n'est pas défini dans .env")
API_URL = f"{VIKUNJA_URL}/api/v2"
HEADERS = {
"Authorization": f"Bearer {VIKUNJA_TOKEN}",
"Accept": "application/json",
}
# =============================================================================
# MCP
# =============================================================================
mcp = FastMCP("vikunja")
# =============================================================================
# Client Vikunja
# =============================================================================
async def vikunja_request(
method: str,
endpoint: str,
params: dict | None = None,
json: dict | None = None,
):
"""
Effectue une requête vers l'API v2 de Vikunja.
"""
url = f"{API_URL}{endpoint}"
async with httpx.AsyncClient(
headers=HEADERS,
verify=VIKUNJA_VERIFY_SSL,
timeout=30.0,
) as client:
response = await client.request(
method,
url,
params=params,
json=json,
)
response.raise_for_status()
if not response.content:
return None
return response.json()
async def vikunja_get(
endpoint: str,
params: dict | None = None,
):
return await vikunja_request(
"GET",
endpoint,
params=params,
)
async def vikunja_post(
endpoint: str,
data: dict,
):
return await vikunja_request(
"POST",
endpoint,
json=data,
)
# =============================================================================
# MCP Tools
# =============================================================================
@mcp.tool()
async def vikunja_get_projects() -> dict:
"""
Retourne les projets Vikunja disponibles pour l'utilisateur.
À utiliser notamment avant de créer une tâche lorsque l'utilisateur
ne donne pas directement l'identifiant du projet.
"""
projects = await vikunja_get(
"/projects",
params={
"format": "markdown",
},
)
result = []
for project in projects or []:
result.append(
{
"id": project.get("id"),
"title": project.get("title"),
"description": project.get("description"),
"parent_project_id": project.get("parent_project_id"),
"is_archived": project.get("is_archived"),
"created": project.get("created"),
"updated": project.get("updated"),
}
)
return {
"count": len(result),
"projects": result,
}
@mcp.tool()
async def vikunja_get_project(
project_id: int,
) -> dict:
"""
Retourne les informations détaillées d'un projet Vikunja.
Args:
project_id:
ID du projet.
"""
project = await vikunja_get(
f"/projects/{project_id}",
params={
"format": "markdown",
},
)
return {
"id": project.get("id"),
"title": project.get("title"),
"description": project.get("description"),
"parent_project_id": project.get("parent_project_id"),
"is_archived": project.get("is_archived"),
"created": project.get("created"),
"updated": project.get("updated"),
}
@mcp.tool()
async def vikunja_get_tasks(
project_id: int,
) -> dict:
"""
Retourne les tâches d'un projet Vikunja.
Args:
project_id:
ID du projet Vikunja.
"""
tasks = await vikunja_get(
f"/projects/{project_id}/tasks",
params={
"format": "markdown",
},
)
result = []
for task in tasks or []:
result.append(
{
"id": task.get("id"),
"title": task.get("title"),
"description": task.get("description"),
"done": task.get("done"),
"priority": task.get("priority"),
"due_date": task.get("due_date"),
"start_date": task.get("start_date"),
"end_date": task.get("end_date"),
"project_id": task.get("project_id"),
"labels": task.get("labels", []),
"created": task.get("created"),
"updated": task.get("updated"),
}
)
return {
"project_id": project_id,
"count": len(result),
"tasks": result,
}
@mcp.tool()
async def vikunja_get_task(
task_id: int,
) -> dict:
"""
Retourne le détail d'une tâche Vikunja.
Args:
task_id:
ID de la tâche.
"""
task = await vikunja_get(
f"/tasks/{task_id}",
params={
"format": "markdown",
},
)
return {
"id": task.get("id"),
"title": task.get("title"),
"description": task.get("description"),
"done": task.get("done"),
"priority": task.get("priority"),
"due_date": task.get("due_date"),
"start_date": task.get("start_date"),
"end_date": task.get("end_date"),
"project_id": task.get("project_id"),
"labels": task.get("labels", []),
"assignees": task.get("assignees", []),
"created": task.get("created"),
"updated": task.get("updated"),
}
@mcp.tool()
async def vikunja_create_task(
project_id: int,
title: str,
description: str = "",
due_date: str | None = None,
priority: int = 0,
) -> dict:
"""
Crée une nouvelle tâche dans Vikunja.
Args:
project_id:
ID du projet dans lequel créer la tâche.
title:
Titre de la tâche.
description:
Description facultative.
due_date:
Date d'échéance facultative au format ISO 8601.
Exemple : 2026-09-15T18:00:00+02:00
priority:
Priorité Vikunja.
0 signifie aucune priorité.
"""
data = {
"title": title,
"description": description,
"priority": priority,
}
if due_date:
data["due_date"] = due_date
task = await vikunja_request(
"POST",
f"/projects/{project_id}/tasks",
params={
"format": "markdown",
},
json=data,
)
return {
"success": True,
"task": {
"id": task.get("id"),
"title": task.get("title"),
"description": task.get("description"),
"project_id": task.get("project_id"),
"priority": task.get("priority"),
"due_date": task.get("due_date"),
"done": task.get("done"),
},
}
# =============================================================================
# Main
# =============================================================================
if __name__ == "__main__":
mcp.run()