oc
2a73f2ef5e
Standard-library Python only, so any device can run it without a virtualenv: vk CLI -- for harnesses that can shell out vikunja_mcp.py MCP server (stdio) -- for harnesses that speak MCP vkclient.py shared client vk-watch@.service systemd template for the per-device task watcher AGENT-ONBOARDING.md is a prompt you can hand to an agent so it configures its own access and verifies it.
262 lines
10 KiB
Python
262 lines
10 KiB
Python
"""
|
|
Thin Vikunja API client shared by the `vk` CLI and the MCP server.
|
|
|
|
Standard library only, on purpose: these tools get invoked from a half-dozen
|
|
different agent harnesses, and every one of them would otherwise need its own
|
|
virtualenv with the same two dependencies in it.
|
|
|
|
Auth is an API token (`Authorization: Bearer tk_...`). Tokens cannot be minted
|
|
by a token -- Vikunja excludes the `tokens` route group from the set an API
|
|
token can be granted (pkg/models/api_routes.go) -- so `bootstrap()` logs in
|
|
with a password to get a JWT and mints the token with that. This is why bots
|
|
must be local accounts: an OIDC-only account has no password to log in with,
|
|
and would need a browser round-trip to ever get its first token.
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
DEFAULT_URL = "https://todo.dominat.us"
|
|
CONFIG_PATH = os.path.expanduser(
|
|
os.environ.get("VIKUNJA_CONFIG", "~/.config/vikunja-agents/config.json")
|
|
)
|
|
|
|
# Vikunja caps page size at `maxitemsperpage` (50 on this instance), so every
|
|
# read-all has to page. Callers get the whole set and never see the cursor.
|
|
PAGE_SIZE = 50
|
|
|
|
|
|
class VikunjaError(Exception):
|
|
"""An API call failed. `status` is None for transport-level failures."""
|
|
|
|
def __init__(self, message, status=None, payload=None):
|
|
super().__init__(message)
|
|
self.status = status
|
|
self.payload = payload
|
|
|
|
|
|
def load_config():
|
|
try:
|
|
with open(CONFIG_PATH) as f:
|
|
return json.load(f)
|
|
except FileNotFoundError:
|
|
return {}
|
|
except ValueError as e:
|
|
raise VikunjaError(f"{CONFIG_PATH} is not valid JSON: {e}")
|
|
|
|
|
|
def save_config(config):
|
|
os.makedirs(os.path.dirname(CONFIG_PATH), exist_ok=True)
|
|
# The file holds bot API tokens; don't create it world-readable.
|
|
fd = os.open(CONFIG_PATH, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
|
|
with os.fdopen(fd, "w") as f:
|
|
json.dump(config, f, indent=2, sort_keys=True)
|
|
f.write("\n")
|
|
|
|
|
|
def resolve(url=None, token=None, profile=None):
|
|
"""Settle on (url, token, username) from args, then env, then the config file."""
|
|
config = load_config()
|
|
profile = profile or os.environ.get("VIKUNJA_PROFILE") or config.get("default_profile")
|
|
entry = (config.get("profiles") or {}).get(profile, {}) if profile else {}
|
|
|
|
url = url or os.environ.get("VIKUNJA_URL") or entry.get("url") or config.get("url") or DEFAULT_URL
|
|
token = token or os.environ.get("VIKUNJA_TOKEN") or entry.get("token")
|
|
username = os.environ.get("VIKUNJA_USERNAME") or entry.get("username")
|
|
return url.rstrip("/"), token, username
|
|
|
|
|
|
class Vikunja:
|
|
def __init__(self, url=None, token=None, profile=None, api_version="v1"):
|
|
self.url, self.token, self.username = resolve(url, token, profile)
|
|
self.api = f"{self.url}/api/{api_version}"
|
|
|
|
# -- plumbing ---------------------------------------------------------
|
|
|
|
def request(self, method, path, body=None, params=None, auth=None):
|
|
target = f"{self.api}{path}"
|
|
if params:
|
|
clean = {k: v for k, v in params.items() if v is not None}
|
|
if clean:
|
|
target += "?" + urllib.parse.urlencode(clean)
|
|
|
|
data = json.dumps(body).encode() if body is not None else None
|
|
req = urllib.request.Request(target, data=data, method=method)
|
|
req.add_header("Content-Type", "application/json")
|
|
req.add_header("Accept", "application/json")
|
|
|
|
bearer = auth if auth is not None else self.token
|
|
if bearer:
|
|
req.add_header("Authorization", f"Bearer {bearer}")
|
|
elif auth is None:
|
|
raise VikunjaError(
|
|
"No API token. Run `vk bootstrap --username <bot>`, or set VIKUNJA_TOKEN."
|
|
)
|
|
|
|
try:
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
raw = resp.read()
|
|
except urllib.error.HTTPError as e:
|
|
raw = e.read()
|
|
try:
|
|
payload = json.loads(raw)
|
|
detail = payload.get("message") or payload.get("error") or raw.decode()
|
|
except ValueError:
|
|
payload, detail = None, raw.decode(errors="replace")[:400]
|
|
raise VikunjaError(f"{method} {path} -> {e.code}: {detail}", e.code, payload)
|
|
except Exception as e:
|
|
raise VikunjaError(f"{method} {path} failed: {e}")
|
|
|
|
if not raw:
|
|
return None
|
|
try:
|
|
return json.loads(raw)
|
|
except ValueError:
|
|
return raw.decode(errors="replace")
|
|
|
|
def paged(self, path, params=None, limit=None):
|
|
"""Follow pagination to the end (or until `limit` items)."""
|
|
out, page = [], 1
|
|
while True:
|
|
batch = self.request(
|
|
"GET", path, params={**(params or {}), "page": page, "per_page": PAGE_SIZE}
|
|
)
|
|
if not batch:
|
|
break
|
|
out.extend(batch)
|
|
if len(batch) < PAGE_SIZE or (limit and len(out) >= limit):
|
|
break
|
|
page += 1
|
|
return out[:limit] if limit else out
|
|
|
|
# -- bootstrap --------------------------------------------------------
|
|
|
|
def login(self, username, password, totp=None):
|
|
body = {"username": username, "password": password}
|
|
if totp:
|
|
body["totp_passcode"] = totp
|
|
result = self.request("POST", "/login", body=body, auth="")
|
|
jwt = (result or {}).get("token")
|
|
if not jwt:
|
|
raise VikunjaError(f"Login succeeded but returned no token: {result}")
|
|
return jwt
|
|
|
|
def token_permissions(self, jwt, wanted):
|
|
"""Intersect the permissions we want with what this server actually offers.
|
|
|
|
Route-group names are derived from the registered routes at startup, so
|
|
they can drift between versions. Asking /routes beats hardcoding names
|
|
and getting a 400 at token-creation time.
|
|
"""
|
|
available = self.request("GET", "/routes", auth=jwt) or {}
|
|
granted, missing = {}, []
|
|
for group, verbs in wanted.items():
|
|
if group not in available:
|
|
missing.append(group)
|
|
continue
|
|
ok = [v for v in verbs if v in available[group]]
|
|
missing += [f"{group}.{v}" for v in verbs if v not in available[group]]
|
|
if ok:
|
|
granted[group] = ok
|
|
return granted, missing
|
|
|
|
def create_token(self, jwt, title, permissions, expires_at):
|
|
return self.request(
|
|
"PUT",
|
|
"/tokens",
|
|
body={"title": title, "permissions": permissions, "expires_at": expires_at},
|
|
auth=jwt,
|
|
)
|
|
|
|
# -- the operations that carry the workflow ---------------------------
|
|
|
|
def whoami(self):
|
|
"""The account this token belongs to.
|
|
|
|
API tokens can never reach /user -- Vikunja excludes every `user_*`
|
|
route group from the token-grantable set -- so the username is recorded
|
|
in the config at bootstrap time and read back from there. Falling back
|
|
to the API keeps this working under a JWT.
|
|
"""
|
|
if self.username:
|
|
return self.username
|
|
user = self.request("GET", "/user")
|
|
return (user or {}).get("username")
|
|
|
|
def projects(self):
|
|
return self.paged("/projects")
|
|
|
|
def tasks(self, project_id=None, filter=None, sort_by=None, limit=None):
|
|
path = f"/projects/{project_id}/tasks" if project_id else "/tasks"
|
|
return self.paged(
|
|
path, params={"filter": filter, "sort_by": sort_by, "filter_include_nulls": "false"}, limit=limit
|
|
)
|
|
|
|
def task(self, task_id):
|
|
return self.request("GET", f"/tasks/{task_id}")
|
|
|
|
def create_task(self, project_id, title, description=None, priority=None, due_date=None):
|
|
body = {"title": title}
|
|
if description:
|
|
body["description"] = description
|
|
if priority is not None:
|
|
body["priority"] = priority
|
|
if due_date:
|
|
body["due_date"] = due_date
|
|
return self.request("PUT", f"/projects/{project_id}/tasks", body=body)
|
|
|
|
def update_task(self, task_id, **fields):
|
|
"""Vikunja replaces the whole task on update, so merge onto current state."""
|
|
current = self.task(task_id)
|
|
current.update({k: v for k, v in fields.items() if v is not None})
|
|
return self.request("POST", f"/tasks/{task_id}", body=current)
|
|
|
|
def comments(self, task_id):
|
|
return self.paged(f"/tasks/{task_id}/comments")
|
|
|
|
def comment(self, task_id, text):
|
|
return self.request("PUT", f"/tasks/{task_id}/comments", body={"comment": text})
|
|
|
|
def assignees(self, task_id):
|
|
return self.request("GET", f"/tasks/{task_id}/assignees")
|
|
|
|
def assign(self, task_id, user_id):
|
|
"""Assign, treating an existing assignment as success.
|
|
|
|
Agents use assignment to claim work, and a claim that fails because the
|
|
claim already happened is not a failure worth propagating.
|
|
"""
|
|
try:
|
|
return self.request("PUT", f"/tasks/{task_id}/assignees", body={"user_id": user_id})
|
|
except VikunjaError as e:
|
|
if e.status == 400 and "already assigned" in str(e):
|
|
return {"already_assigned": True, "user_id": user_id}
|
|
raise
|
|
|
|
def unassign(self, task_id, user_id):
|
|
return self.request("DELETE", f"/tasks/{task_id}/assignees/{user_id}")
|
|
|
|
def find_user(self, username, project_id):
|
|
"""Resolve a username to a user object, scoped to one project.
|
|
|
|
The instance-wide /users search is not in the token-grantable route set,
|
|
so this uses /projects/{id}/projectusers instead -- which is arguably the
|
|
better check anyway: it only finds users who actually have access to the
|
|
project you are about to assign work in.
|
|
"""
|
|
found = self.request("GET", f"/projects/{project_id}/projectusers",
|
|
params={"s": username}) or []
|
|
for u in found:
|
|
if u.get("username") == username:
|
|
return u
|
|
return None
|
|
|
|
def labels(self):
|
|
return self.paged("/labels")
|
|
|
|
def add_label(self, task_id, label_id):
|
|
return self.request("PUT", f"/tasks/{task_id}/labels", body={"label_id": label_id})
|