|
|
@ -1,4 +1,3 @@
|
|
|
|
|
|
|
|
|
|
|
|
import os
|
|
|
|
import os
|
|
|
|
import requests
|
|
|
|
import requests
|
|
|
|
import logging
|
|
|
|
import logging
|
|
|
@ -108,13 +107,13 @@ def search_tasks(query: str):
|
|
|
|
logger.exception("search_tasks: request failed")
|
|
|
|
logger.exception("search_tasks: request failed")
|
|
|
|
return f"Error searching tasks: {e}"
|
|
|
|
return f"Error searching tasks: {e}"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@mcp.tool()
|
|
|
|
@mcp.tool()
|
|
|
|
def active_tasks(project_id: int = None):
|
|
|
|
def active_tasks(project_id: int = None, is_favorite: bool = None):
|
|
|
|
"""
|
|
|
|
"""
|
|
|
|
Returns a list of active tasks (where 'done' is False) from Vikunja.
|
|
|
|
Returns a list of active tasks (where 'done' is False) from Vikunja.
|
|
|
|
|
|
|
|
|
|
|
|
:param project_id: Optional project ID to filter tasks by project.
|
|
|
|
:param project_id: Optional project ID to filter tasks by project.
|
|
|
|
|
|
|
|
:param is_favorite: Optional filter to return only tasks matching favorite status.
|
|
|
|
"""
|
|
|
|
"""
|
|
|
|
if "Authorization" not in session.headers:
|
|
|
|
if "Authorization" not in session.headers:
|
|
|
|
return "Please run the 'login' command first."
|
|
|
|
return "Please run the 'login' command first."
|
|
|
@ -145,12 +144,116 @@ def active_tasks(project_id: int = None):
|
|
|
|
active = [t for t in active if t.get("project_id") == project_id or t.get("projectID") == project_id]
|
|
|
|
active = [t for t in active if t.get("project_id") == project_id or t.get("projectID") == project_id]
|
|
|
|
logger.info("active_tasks: filtered by project_id=%s before=%d after=%d", project_id, before, len(active))
|
|
|
|
logger.info("active_tasks: filtered by project_id=%s before=%d after=%d", project_id, before, len(active))
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# If is_favorite provided, filter by is_favorite / is_favourite field
|
|
|
|
|
|
|
|
if is_favorite is not None:
|
|
|
|
|
|
|
|
before = len(active)
|
|
|
|
|
|
|
|
def fav_val(task):
|
|
|
|
|
|
|
|
return bool(task.get("is_favorite", task.get("is_favourite", False)))
|
|
|
|
|
|
|
|
active = [t for t in active if fav_val(t) == bool(is_favorite)]
|
|
|
|
|
|
|
|
logger.info("active_tasks: filtered by is_favorite=%s before=%d after=%d", is_favorite, before, len(active))
|
|
|
|
|
|
|
|
|
|
|
|
logger.info("active_tasks: fetched=%d active=%d", len(all_tasks), len(active))
|
|
|
|
logger.info("active_tasks: fetched=%d active=%d", len(all_tasks), len(active))
|
|
|
|
return active
|
|
|
|
return active
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
|
|
logger.exception("active_tasks: request failed")
|
|
|
|
logger.exception("active_tasks: request failed")
|
|
|
|
return f"Error retrieving active tasks: {e}"
|
|
|
|
return f"Error retrieving active tasks: {e}"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@mcp.tool()
|
|
|
|
|
|
|
|
def get_task_details(task_id: int):
|
|
|
|
|
|
|
|
"""
|
|
|
|
|
|
|
|
Gets task details in Vikunja.
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
:param task_id: The ID of the task of which to fetch comments.
|
|
|
|
|
|
|
|
"""
|
|
|
|
|
|
|
|
if "Authorization" not in session.headers:
|
|
|
|
|
|
|
|
return "Please run the 'login' command first."
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
try:
|
|
|
|
|
|
|
|
response = session.get(f"{VIKUNJA_URL}/api/v1/tasks/{task_id}")
|
|
|
|
|
|
|
|
response.raise_for_status()
|
|
|
|
|
|
|
|
details = response.json()
|
|
|
|
|
|
|
|
if not details:
|
|
|
|
|
|
|
|
return "No details found for this task."
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# Helpers to produce compact readable strings for nested structures
|
|
|
|
|
|
|
|
def join_names(items):
|
|
|
|
|
|
|
|
if not items:
|
|
|
|
|
|
|
|
return "[]"
|
|
|
|
|
|
|
|
out = []
|
|
|
|
|
|
|
|
for it in items:
|
|
|
|
|
|
|
|
if not isinstance(it, dict):
|
|
|
|
|
|
|
|
out.append(str(it))
|
|
|
|
|
|
|
|
continue
|
|
|
|
|
|
|
|
out.append(str(it.get("name") or it.get("username") or it.get("id") or it))
|
|
|
|
|
|
|
|
return ", ".join(out)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def labels_list(labels):
|
|
|
|
|
|
|
|
if not labels:
|
|
|
|
|
|
|
|
return "[]"
|
|
|
|
|
|
|
|
out = []
|
|
|
|
|
|
|
|
for l in labels:
|
|
|
|
|
|
|
|
if isinstance(l, dict):
|
|
|
|
|
|
|
|
out.append(str(l.get("title") or l.get("name") or l.get("id") or l))
|
|
|
|
|
|
|
|
else:
|
|
|
|
|
|
|
|
out.append(str(l))
|
|
|
|
|
|
|
|
return ", ".join(out)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
fetched_id = details.get("id", "N/A")
|
|
|
|
|
|
|
|
title = details.get("title", "N/A")
|
|
|
|
|
|
|
|
identifier = details.get("identifier", "N/A")
|
|
|
|
|
|
|
|
created = details.get("created", "N/A")
|
|
|
|
|
|
|
|
updated = details.get("updated", "N/A")
|
|
|
|
|
|
|
|
created_by = details.get("created_by") or {}
|
|
|
|
|
|
|
|
created_by_str = f"{created_by.get('name','N/A')} (id={created_by.get('id','N/A')}, email={created_by.get('email','N/A')}, username={created_by.get('username','N/A')})"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
description = details.get("description", "N/A")
|
|
|
|
|
|
|
|
done = details.get("done", "N/A")
|
|
|
|
|
|
|
|
done_at = details.get("done_at", "N/A")
|
|
|
|
|
|
|
|
percent_done = details.get("percent_done", "N/A")
|
|
|
|
|
|
|
|
due_date = details.get("due_date", "N/A")
|
|
|
|
|
|
|
|
start_date = details.get("start_date", "N/A")
|
|
|
|
|
|
|
|
end_date = details.get("end_date", "N/A")
|
|
|
|
|
|
|
|
repeat_mode = details.get("repeat_mode", "N/A")
|
|
|
|
|
|
|
|
repeat_after = details.get("repeat_after", "N/A")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
project_id = details.get("project_id", "N/A")
|
|
|
|
|
|
|
|
bucket_id = details.get("bucket_id", "N/A")
|
|
|
|
|
|
|
|
buckets = details.get("buckets", [])
|
|
|
|
|
|
|
|
index = details.get("index", "N/A")
|
|
|
|
|
|
|
|
position = details.get("position", "N/A")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
priority = details.get("priority", "N/A")
|
|
|
|
|
|
|
|
is_favorite = details.get("is_favorite", details.get("is_favourite", "N/A"))
|
|
|
|
|
|
|
|
hex_color = details.get("hex_color", "N/A")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
assignees = details.get("assignees", [])
|
|
|
|
|
|
|
|
attachments = details.get("attachments", [])
|
|
|
|
|
|
|
|
cover_image_attachment_id = details.get("cover_image_attachment_id", "N/A")
|
|
|
|
|
|
|
|
comments = details.get("comments", [])
|
|
|
|
|
|
|
|
labels = details.get("labels", [])
|
|
|
|
|
|
|
|
reminders = details.get("reminders", [])
|
|
|
|
|
|
|
|
reactions = details.get("reactions", {})
|
|
|
|
|
|
|
|
related_tasks = details.get("related_tasks", {})
|
|
|
|
|
|
|
|
subscription = details.get("subscription", {})
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
result = []
|
|
|
|
|
|
|
|
result.append(f"ID: {fetched_id}, Identifier: {identifier}, Title: {title}")
|
|
|
|
|
|
|
|
result.append(f"Created: {created} by {created_by_str}, Updated: {updated}")
|
|
|
|
|
|
|
|
result.append(f"Done: {done} (done_at={done_at}), Percent Done: {percent_done}, Priority: {priority}, Favorite: {is_favorite}")
|
|
|
|
|
|
|
|
result.append(f"Due Date: {due_date}, Start Date: {start_date}, End Date: {end_date}, Repeat Mode: {repeat_mode}, Repeat After: {repeat_after}")
|
|
|
|
|
|
|
|
result.append(f"Project ID: {project_id}, Bucket ID: {bucket_id}, Buckets: {len(buckets)} items, Index: {index}, Position: {position}")
|
|
|
|
|
|
|
|
result.append(f"Labels: {labels_list(labels)}, Hex Color: {hex_color}")
|
|
|
|
|
|
|
|
result.append(f"Assignees: {join_names(assignees)}, Attachments Count: {len(attachments)}, Cover Image Attachment ID: {cover_image_attachment_id}")
|
|
|
|
|
|
|
|
result.append(f"Comments Count: {len(comments)}, Reminders Count: {len(reminders)}")
|
|
|
|
|
|
|
|
result.append(f"Reactions: {reactions}, Related Tasks: {related_tasks}, Subscription: {subscription}")
|
|
|
|
|
|
|
|
result.append(f"Description: {description}")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
return "\n".join(result)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
|
|
|
|
|
|
logger.exception("get_task_details: request failed for task_id=%s", task_id)
|
|
|
|
|
|
|
|
return f"Error looking up details for task: {e}"
|
|
|
|
|
|
|
|
|
|
|
|
@mcp.tool()
|
|
|
|
@mcp.tool()
|
|
|
|
def add_task(project_id: int, title: str, description: str = ""):
|
|
|
|
def add_task(project_id: int, title: str, description: str = ""):
|
|
|
|
"""
|
|
|
|
"""
|
|
|
@ -423,6 +526,77 @@ def create_project(title: str, description: str = "", is_favorite: bool = False)
|
|
|
|
logger.exception("create_project: request failed for title=%s", title)
|
|
|
|
logger.exception("create_project: request failed for title=%s", title)
|
|
|
|
return f"Error creating project: {e}"
|
|
|
|
return f"Error creating project: {e}"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
@mcp.tool()
|
|
|
|
|
|
|
|
def update_task_details(
|
|
|
|
|
|
|
|
task_id: int,
|
|
|
|
|
|
|
|
title: str = None,
|
|
|
|
|
|
|
|
description: str = None,
|
|
|
|
|
|
|
|
done: bool = None,
|
|
|
|
|
|
|
|
percent_done: int = None,
|
|
|
|
|
|
|
|
due_date: str = None,
|
|
|
|
|
|
|
|
start_date: str = None,
|
|
|
|
|
|
|
|
end_date: str = None,
|
|
|
|
|
|
|
|
repeat_mode: int = None,
|
|
|
|
|
|
|
|
repeat_after: int = None,
|
|
|
|
|
|
|
|
project_id: int = None,
|
|
|
|
|
|
|
|
bucket_id: int = None,
|
|
|
|
|
|
|
|
priority: int = None,
|
|
|
|
|
|
|
|
position: int = None,
|
|
|
|
|
|
|
|
is_favorite: bool = None,
|
|
|
|
|
|
|
|
hex_color: str = None,
|
|
|
|
|
|
|
|
labels: str = None,
|
|
|
|
|
|
|
|
):
|
|
|
|
|
|
|
|
"""
|
|
|
|
|
|
|
|
Updates any primary fields on a task. Only fields provided (not None) will be sent.
|
|
|
|
|
|
|
|
If `labels` is provided as a comma-separated string, it will replace existing labels.
|
|
|
|
|
|
|
|
"""
|
|
|
|
|
|
|
|
if "Authorization" not in session.headers:
|
|
|
|
|
|
|
|
return "Please run the 'login' command first."
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
payload = {}
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# Simple helper to set only when value explicitly passed (not None)
|
|
|
|
|
|
|
|
def set_if(provided, key, transform=lambda x: x):
|
|
|
|
|
|
|
|
if provided is not None:
|
|
|
|
|
|
|
|
payload[key] = transform(provided)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
set_if(title, "title")
|
|
|
|
|
|
|
|
set_if(description, "description")
|
|
|
|
|
|
|
|
set_if(done, "done")
|
|
|
|
|
|
|
|
set_if(percent_done, "percent_done")
|
|
|
|
|
|
|
|
set_if(due_date, "due_date")
|
|
|
|
|
|
|
|
set_if(start_date, "start_date")
|
|
|
|
|
|
|
|
set_if(end_date, "end_date")
|
|
|
|
|
|
|
|
set_if(repeat_mode, "repeat_mode")
|
|
|
|
|
|
|
|
set_if(repeat_after, "repeat_after")
|
|
|
|
|
|
|
|
set_if(project_id, "project_id")
|
|
|
|
|
|
|
|
set_if(bucket_id, "bucket_id")
|
|
|
|
|
|
|
|
set_if(priority, "priority")
|
|
|
|
|
|
|
|
set_if(position, "position")
|
|
|
|
|
|
|
|
set_if(is_favorite, "is_favorite")
|
|
|
|
|
|
|
|
set_if(hex_color, "hex_color")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# Replace labels if provided as comma-separated string
|
|
|
|
|
|
|
|
if labels is not None:
|
|
|
|
|
|
|
|
if isinstance(labels, str):
|
|
|
|
|
|
|
|
parsed = [s.strip() for s in labels.split(",") if s.strip()]
|
|
|
|
|
|
|
|
payload["labels"] = parsed
|
|
|
|
|
|
|
|
else:
|
|
|
|
|
|
|
|
# Allow callers to pass list directly
|
|
|
|
|
|
|
|
payload["labels"] = labels
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
if not payload:
|
|
|
|
|
|
|
|
return "No fields provided to update."
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
try:
|
|
|
|
|
|
|
|
logger.info("update_task_details: updating task_id=%s with payload keys=%s", task_id, list(payload.keys()))
|
|
|
|
|
|
|
|
response = session.post(f"{VIKUNJA_URL}/api/v1/tasks/{task_id}", json=payload)
|
|
|
|
|
|
|
|
response.raise_for_status()
|
|
|
|
|
|
|
|
return response.json()
|
|
|
|
|
|
|
|
except requests.exceptions.RequestException as e:
|
|
|
|
|
|
|
|
logger.exception("update_task_details: request failed for task_id=%s", task_id)
|
|
|
|
|
|
|
|
return f"Error updating task details: {e}"
|
|
|
|
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
if __name__ == "__main__":
|
|
|
|
print("--- Vikunja MCP Client ---")
|
|
|
|
print("--- Vikunja MCP Client ---")
|
|
|
|
print("Available commands: login, search_tasks, add_task, close_task, lookup_project, help, exit")
|
|
|
|
print("Available commands: login, search_tasks, add_task, close_task, lookup_project, help, exit")
|
|
|
|