vikunja-agents/vikunja_mcp.py

#!/usr/bin/env python3
"""
Vikunja MCP server -- stdio transport, JSON-RPC 2.0, standard library only.

Deliberately dependency-free: this gets registered in several different agent
harnesses, and requiring each of them to provision a virtualenv for one SDK is
how a "works everywhere" integration stops working anywhere.

Register it (Claude Code):
    claude mcp add vikunja -- /srv/pods/vikunja-agents/vikunja_mcp.py

Auth comes from the same place the CLI gets it: VIKUNJA_TOKEN / VIKUNJA_PROFILE
in the server's environment, or ~/.config/vikunja-agents/config.json.

Tool descriptions below state *when* to call each tool, not just what it does --
that trigger condition is what actually drives correct tool selection.
"""

import json
import os
import sys

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from vkclient import Vikunja, VikunjaError  # noqa: E402

PROTOCOL_VERSIONS = {"2024-11-05", "2025-03-26", "2025-06-18"}
FALLBACK_PROTOCOL = "2024-11-05"
SERVER_INFO = {"name": "vikunja", "version": "1.0.0"}

TOOLS = [
    {
        "name": "vikunja_list_tasks",
        "description": (
            "List tasks from Vikunja, newest-relevant first. Call this at the start of a "
            "work session to find out what has been assigned to you, and whenever you need "
            "to check whether something is already tracked before creating a duplicate. "
            "Defaults to undone tasks assigned to the calling bot account."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "assigned_to_me": {
                    "type": "boolean",
                    "description": "Only tasks assigned to this bot account. Defaults to true.",
                },
                "project_id": {"type": "integer", "description": "Restrict to one project."},
                "include_done": {
                    "type": "boolean",
                    "description": "Include completed tasks. Defaults to false.",
                },
                "filter": {
                    "type": "string",
                    "description": "Raw Vikunja filter expression, e.g. \"priority >= 3\". ANDed with the other options.",
                },
                "limit": {"type": "integer", "description": "Maximum tasks to return."},
            },
        },
    },
    {
        "name": "vikunja_get_task",
        "description": (
            "Fetch one task with its full description and comment thread. Call this before "
            "starting work on a task -- the comment thread carries the conversation and "
            "prior agents' findings, which the task list does not show."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {"task_id": {"type": "integer"}},
            "required": ["task_id"],
        },
    },
    {
        "name": "vikunja_comment_task",
        "description": (
            "Post a comment on a task. This is the primary way to report progress, findings, "
            "and blockers so other agents and the human can follow along. Call it when you "
            "finish a unit of work, when you discover something that changes the task's "
            "scope, and when you are blocked -- not for routine narration."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "task_id": {"type": "integer"},
                "comment": {"type": "string", "description": "Markdown is rendered."},
            },
            "required": ["task_id", "comment"],
        },
    },
    {
        "name": "vikunja_create_task",
        "description": (
            "Create a task. Call this when work is identified that will not be done in the "
            "current session -- follow-up items, discovered bugs, work for another agent. "
            "Check vikunja_list_tasks first so you do not file a duplicate."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "project_id": {"type": "integer"},
                "title": {"type": "string"},
                "description": {"type": "string", "description": "Markdown is rendered."},
                "priority": {
                    "type": "integer",
                    "description": "0 unset, 1 low, 2 medium, 3 high, 4 urgent, 5 DO NOW.",
                },
                "due_date": {"type": "string", "description": "RFC3339 timestamp."},
                "assign_to": {"type": "string", "description": "Username to assign it to."},
            },
            "required": ["project_id", "title"],
        },
    },
    {
        "name": "vikunja_update_task",
        "description": (
            "Update a task's fields, including marking it done. Call this when the work is "
            "actually finished and verified -- marking a task done is the signal other "
            "agents and the human rely on. Fields you omit are left unchanged."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "task_id": {"type": "integer"},
                "title": {"type": "string"},
                "description": {"type": "string"},
                "priority": {"type": "integer"},
                "done": {"type": "boolean"},
            },
            "required": ["task_id"],
        },
    },
    {
        "name": "vikunja_assign_task",
        "description": (
            "Assign a task to a user or bot account. Call this to hand work to a specific "
            "agent, or to claim a task before starting on it so two agents do not "
            "duplicate each other."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "task_id": {"type": "integer"},
                "username": {"type": "string"},
            },
            "required": ["task_id", "username"],
        },
    },
    {
        "name": "vikunja_list_projects",
        "description": (
            "List projects with their IDs. Call this when you need a project_id to file a "
            "task and do not already know it."
        ),
        "inputSchema": {"type": "object", "properties": {}},
    },
]


def _slim(task):
    """Trim a task to what an agent needs. Full payloads are mostly nulls."""
    return {
        "id": task.get("id"),
        "title": task.get("title"),
        "done": task.get("done"),
        "priority": task.get("priority"),
        "due_date": task.get("due_date"),
        "project_id": task.get("project_id"),
        "assignees": [a.get("username") for a in (task.get("assignees") or [])],
        "labels": [l.get("title") for l in (task.get("labels") or [])],
    }


def call_tool(vk, name, args):
    if name == "vikunja_list_tasks":
        filters = []
        if args.get("assigned_to_me", True):
            filters.append(f"assignees in '{vk.whoami()}'")
        if not args.get("include_done", False):
            filters.append("done = false")
        if args.get("filter"):
            filters.append(args["filter"])
        tasks = vk.tasks(
            project_id=args.get("project_id"),
            filter=" && ".join(filters) or None,
            limit=args.get("limit"),
        )
        return [_slim(t) for t in tasks]

    if name == "vikunja_get_task":
        task = vk.task(args["task_id"])
        detail = _slim(task)
        detail["description"] = task.get("description")
        detail["comments"] = [
            {
                "author": (c.get("author") or {}).get("username"),
                "created": c.get("created"),
                "comment": c.get("comment"),
            }
            for c in vk.comments(args["task_id"])
        ]
        return detail

    if name == "vikunja_comment_task":
        vk.comment(args["task_id"], args["comment"])
        return {"ok": True, "task_id": args["task_id"]}

    if name == "vikunja_create_task":
        task = vk.create_task(
            args["project_id"], args["title"], args.get("description"),
            args.get("priority"), args.get("due_date"),
        )
        if args.get("assign_to"):
            user = vk.find_user(args["assign_to"], args["project_id"])
            if not user:
                raise VikunjaError(
                    f"No user '{args['assign_to']}' with access to project {args['project_id']}")
            vk.assign(task["id"], user["id"])
            task = vk.task(task["id"])
        return _slim(task)

    if name == "vikunja_update_task":
        fields = {k: args[k] for k in ("title", "description", "priority", "done") if k in args}
        if not fields:
            raise VikunjaError("Nothing to update -- supply at least one field besides task_id.")
        return _slim(vk.update_task(args["task_id"], **fields))

    if name == "vikunja_assign_task":
        task = vk.task(args["task_id"])
        user = vk.find_user(args["username"], task["project_id"])
        if not user:
            raise VikunjaError(
                f"No user '{args['username']}' with access to project {task['project_id']}")
        vk.assign(args["task_id"], user["id"])
        return _slim(vk.task(args["task_id"]))

    if name == "vikunja_list_projects":
        return [{"id": p["id"], "title": p.get("title")} for p in vk.projects()]

    raise VikunjaError(f"Unknown tool: {name}")


def handle(message, state):
    """Return a JSON-RPC response dict, or None for notifications."""
    method = message.get("method")
    msg_id = message.get("id")
    params = message.get("params") or {}
    is_notification = msg_id is None

    def ok(result):
        return None if is_notification else {"jsonrpc": "2.0", "id": msg_id, "result": result}

    def err(code, msg):
        return None if is_notification else {
            "jsonrpc": "2.0", "id": msg_id, "error": {"code": code, "message": msg}
        }

    if method == "initialize":
        asked = params.get("protocolVersion")
        return ok({
            # Echo the client's version when we recognise it; otherwise name the
            # one we know it can speak.
            "protocolVersion": asked if asked in PROTOCOL_VERSIONS else FALLBACK_PROTOCOL,
            "capabilities": {"tools": {}},
            "serverInfo": SERVER_INFO,
        })

    if method in ("notifications/initialized", "initialized", "notifications/cancelled"):
        return None

    if method == "ping":
        return ok({})

    if method == "tools/list":
        return ok({"tools": TOOLS})

    if method == "tools/call":
        name = params.get("name")
        args = params.get("arguments") or {}
        try:
            result = call_tool(state["vk"], name, args)
        except VikunjaError as e:
            # Tool failures are results with isError, not protocol errors -- the
            # model needs to see the message so it can adapt.
            return ok({"content": [{"type": "text", "text": f"Error: {e}"}], "isError": True})
        except KeyError as e:
            return ok({
                "content": [{"type": "text", "text": f"Error: missing required argument {e}"}],
                "isError": True,
            })
        return ok({"content": [{"type": "text", "text": json.dumps(result, indent=2, default=str)}]})

    if method in ("shutdown", "exit"):
        state["running"] = False
        return ok({})

    return err(-32601, f"Method not found: {method}")


def main():
    state = {"vk": Vikunja(), "running": True}
    stdout = sys.stdout

    for line in sys.stdin:
        if not state["running"]:
            break
        line = line.strip()
        if not line:
            continue
        try:
            message = json.loads(line)
        except ValueError:
            stdout.write(json.dumps({
                "jsonrpc": "2.0", "id": None,
                "error": {"code": -32700, "message": "Parse error"},
            }) + "\n")
            stdout.flush()
            continue

        try:
            response = handle(message, state)
        except Exception as e:  # never let one bad call kill the server
            response = {
                "jsonrpc": "2.0", "id": message.get("id"),
                "error": {"code": -32603, "message": f"Internal error: {e}"},
            }
            if message.get("id") is None:
                response = None

        if response is not None:
            stdout.write(json.dumps(response) + "\n")
            stdout.flush()


if __name__ == "__main__":
    main()