Research-Stack/4-Infrastructure/infra/vikunja_mcp.py

#!/usr/bin/env python3
"""
MCP server for Vikunja kanban board on neon.
Exposes tools for agents/LLMs to manage projects, tasks, and lists.

Environment:
  VIKUNJA_URL  — default http://100.92.88.64:3456
  VIKUNJA_TOKEN — API token from /api/v1/login
"""

from __future__ import annotations

import json
import os
import sys
from typing import Any
from urllib.request import Request, urlopen
from urllib.error import HTTPError

SERVER_NAME = "vikunja-mcp"
SERVER_VERSION = "0.1.0"
VIKUNJA_URL = os.environ.get("VIKUNJA_URL", "http://100.92.88.64:3456")
_token_file = os.environ.get("VIKUNJA_TOKEN_FILE", "")
VIKUNJA_TOKEN = os.environ.get("VIKUNJA_TOKEN", "")
if not VIKUNJA_TOKEN and _token_file:
    try:
        with open(_token_file) as f:
            VIKUNJA_TOKEN = f.read().strip()
    except FileNotFoundError:
        pass


_VIEW_ID = 13  # Kanban view ID for project 3 (Research Stack)


def _api(method: str, path: str, data: dict | None = None) -> Any:
    url = f"{VIKUNJA_URL}/api/v1{path}"
    headers = {"Content-Type": "application/json"}
    if VIKUNJA_TOKEN:
        headers["Authorization"] = f"Bearer {VIKUNJA_TOKEN}"
    body = json.dumps(data).encode() if data else None
    req = Request(url, data=body, headers=headers, method=method)
    try:
        with urlopen(req) as resp:
            return json.loads(resp.read())
    except HTTPError as e:
        return {"error": f"{e.code}: {e.reason}", "body": e.read().decode(errors="replace")[:500]}


def json_text(data: Any) -> list[dict[str, str]]:
    return [{"type": "text", "text": json.dumps(data, indent=2, sort_keys=True)}]


def _view_id(pid: int) -> int:
    """Return the kanban view ID for a project."""
    if pid == 3:
        return 13
    views = _api("GET", f"/projects/{pid}/views")
    if isinstance(views, list):
        for v in views:
            if v.get("view_kind") == "kanban":
                return v["id"]
    return 0


def tool_list_projects(_: dict[str, Any]) -> dict[str, Any]:
    projects = _api("GET", "/projects")
    return {"ok": True, "projects": projects}


def tool_list_lists(args: dict[str, Any]) -> dict[str, Any]:
    pid = args.get("project_id", 3)
    vid = _view_id(pid)
    if not vid:
        return {"ok": False, "error": "no kanban view found"}
    buckets = _api("GET", f"/projects/{pid}/views/{vid}/buckets")
    return {"ok": True, "lists": buckets}


def tool_list_tasks(args: dict[str, Any]) -> dict[str, Any]:
    pid = args.get("project_id", 3)
    lid = args.get("list_id", 0)
    vid = _view_id(pid)
    if not vid:
        tasks = _api("GET", f"/projects/{pid}/tasks")
        return {"ok": True, "tasks": tasks}
    buckets = _api("GET", f"/projects/{pid}/views/{vid}/tasks")
    if isinstance(buckets, list):
        if lid:
            for b in buckets:
                if b.get("id") == lid:
                    return {"ok": True, "tasks": b.get("tasks", [])}
            return {"ok": True, "tasks": []}
        all_tasks = []
        for b in buckets:
            all_tasks.extend(b.get("tasks", []))
        return {"ok": True, "tasks": all_tasks}
    return {"ok": True, "tasks": []}


def tool_create_task(args: dict[str, Any]) -> dict[str, Any]:
    pid = args.get("project_id", 3)
    title = args.get("title", "")
    desc = args.get("description", "")
    body = {"title": title, "description": desc}
    result = _api("PUT", f"/projects/{pid}/tasks", body)
    task_id = result.get("id")
    if task_id and "list_id" in args:
        lid = args["list_id"]
        _api("POST", f"/tasks/{task_id}", {"bucket_id": lid})
    return {"ok": bool(task_id), "task": result,
            "note": "Use web UI at {VIKUNJA_URL} to assign tasks to kanban buckets"}


def tool_move_task(args: dict[str, Any]) -> dict[str, Any]:
    task_id = args.get("task_id", 0)
    lid = args.get("list_id", 0)
    _api("POST", f"/tasks/{task_id}", {"bucket_id": lid})
    return {"ok": True,
            "note": "Open the web UI to confirm and adjust bucket assignment: "
                    f"{VIKUNJA_URL}/projects/3?view=13"}


def tool_update_task(args: dict[str, Any]) -> dict[str, Any]:
    task_id = args.get("task_id", 0)
    body = {k: v for k, v in args.items() if k in ("title", "description", "done", "priority") and v is not None}
    result = _api("POST", f"/tasks/{task_id}", body)
    return {"ok": True, "task": result}


def tool_status(_: dict[str, Any]) -> dict[str, Any]:
    try:
        info = _api("GET", "/info")
        ok = "version" in info
    except Exception as exc:
        info = {"error": str(exc)}
        ok = False
    return {"ok": ok, "server": SERVER_NAME, "version": SERVER_VERSION,
            "web_ui": f"{VIKUNJA_URL}/projects/3?view=13",
            **info}


TOOLS = {
    "kanban_list_projects": {
        "description": "List all kanban projects/boards.",
        "inputSchema": {"type": "object", "properties": {}},
        "handler": tool_list_projects,
    },
    "kanban_list_lists": {
        "description": "List buckets/columns in a project's kanban view.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "project_id": {"type": "integer", "default": 3},
            },
        },
        "handler": tool_list_lists,
    },
    "kanban_list_tasks": {
        "description": "List all tasks, or tasks in a specific bucket. "
                       "Shows bucket membership from the kanban view.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "project_id": {"type": "integer", "default": 3},
                "list_id": {"type": "integer", "default": 0},
            },
        },
        "handler": tool_list_tasks,
    },
    "kanban_create_task": {
        "description": "Create a new task. Tag with list_id to assign to a bucket. "
                       "NOTE: Vikunja v2 does not persist bucket_id on the task model; "
                       "use the web UI to move tasks between kanban columns.",
        "inputSchema": {
            "type": "object",
            "required": ["title"],
            "properties": {
                "project_id": {"type": "integer", "default": 3},
                "title": {"type": "string"},
                "description": {"type": "string", "default": ""},
                "list_id": {"type": "integer", "description": "Bucket ID from kanban_list_lists"},
            },
        },
        "handler": tool_create_task,
    },
    "kanban_move_task": {
        "description": "Move a task to a different bucket. "
                       "NOTE: Staged via task update; open the web UI to confirm "
                       "drag-and-drop placement in the kanban view.",
        "inputSchema": {
            "type": "object",
            "required": ["task_id", "list_id"],
            "properties": {
                "task_id": {"type": "integer"},
                "list_id": {"type": "integer"},
            },
        },
        "handler": tool_move_task,
    },
    "kanban_update_task": {
        "description": "Update a task's title, description, done status, or priority. "
                       "Works with standard Vikunja task API (POST /tasks/{id}).",
        "inputSchema": {
            "type": "object",
            "required": ["task_id"],
            "properties": {
                "task_id": {"type": "integer"},
                "title": {"type": "string"},
                "description": {"type": "string"},
                "done": {"type": "boolean"},
                "priority": {"type": "integer"},
            },
        },
        "handler": tool_update_task,
    },
    "kanban_status": {
        "description": "Return kanban server health, version, and web UI URL.",
        "inputSchema": {"type": "object", "properties": {}},
        "handler": tool_status,
    },
}


def handle(message: dict[str, Any]) -> dict[str, Any] | None:
    method = message.get("method")
    msg_id = message.get("id")
    if method == "initialize":
        return {
            "jsonrpc": "2.0", "id": msg_id,
            "result": {
                "protocolVersion": "2024-11-05",
                "capabilities": {"tools": {}},
                "serverInfo": {"name": SERVER_NAME, "version": SERVER_VERSION},
            },
        }
    if method == "tools/list":
        return {
            "jsonrpc": "2.0", "id": msg_id,
            "result": {
                "tools": [
                    {"name": name, "description": data["description"], "inputSchema": data["inputSchema"]}
                    for name, data in TOOLS.items()
                ]
            },
        }
    if method == "tools/call":
        params = message.get("params") or {}
        name = params.get("name")
        args = params.get("arguments") or {}
        if name not in TOOLS:
            result = {"ok": False, "error": f"unknown tool: {name}"}
        else:
            result = TOOLS[name]["handler"](args)
        return {"jsonrpc": "2.0", "id": msg_id, "result": {"content": json_text(result)}}
    if method and method.startswith("notifications/"):
        return None
    return {"jsonrpc": "2.0", "id": msg_id, "error": {"code": -32601, "message": f"method not found: {method}"}}


def main() -> None:
    if not VIKUNJA_TOKEN:
        err = {"jsonrpc": "2.0", "id": None, "error": {"code": -32000, "message": "VIKUNJA_TOKEN not set"}}
        print(json.dumps(err, separators=(",", ":")), flush=True)
        sys.exit(1)
    for line in sys.stdin:
        if not line.strip():
            continue
        try:
            response = handle(json.loads(line))
        except Exception as exc:
            response = {"jsonrpc": "2.0", "id": None, "error": {"code": -32000, "message": f"{type(exc).__name__}: {exc}"}}
        if response is not None:
            print(json.dumps(response, separators=(",", ":")), flush=True)


if __name__ == "__main__":
    main()