V1
This commit is contained in:
Executable
+6
@@ -0,0 +1,6 @@
|
||||
WAZUH_URL=https://192.168.1.52:55000
|
||||
WAZUH_USER=wazuh-wui
|
||||
WAZUH_PASSWORD=y+tx9RA3cYiBxqpnoE2+138pV08Jhnwa
|
||||
INDEXER_URL=https://192.168.1.52:9200
|
||||
INDEXER_USER=kibanaserver
|
||||
INDEXER_PASSWORD=cIh?n7Y1FeACDZRb063r0+REWoI3dQAA
|
||||
@@ -0,0 +1,3 @@
|
||||
mcp
|
||||
httpx
|
||||
python-dotenv
|
||||
@@ -0,0 +1,327 @@
|
||||
#!/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()
|
||||
Reference in New Issue
Block a user