Vikunja tooling for agent harnesses
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.
This commit is contained in:
oc
+261
@@ -0,0 +1,261 @@
|
||||
"""
|
||||
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})
|
||||
Reference in New Issue
Block a user