#!/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()