Add a full get details, add an option 'is_favorite' to active lookup, add update details
CICD / Explore-Gitea-Actions (push) Successful in 38s
Details
CICD / Explore-Gitea-Actions (push) Successful in 38s
Details
This commit is contained in:
parent
e125661161
commit
fe9d20ca2d
|
@ -20,4 +20,4 @@ RUN pip install --no-cache-dir -r requirements.txt
|
|||
COPY . .
|
||||
|
||||
CMD ["python", "main.py"]
|
||||
#harbor.freshbrewed.science/library/vikunjamcp:0.18
|
||||
#harbor.freshbrewed.science/library/vikunjamcp:0.19
|
||||
|
|
|
@ -1,6 +1,6 @@
|
|||
{
|
||||
"name": "vikunja",
|
||||
"version": "1.0.18",
|
||||
"version": "1.0.19",
|
||||
"mcpServers": {
|
||||
"nodeServer": {
|
||||
"command": "docker",
|
||||
|
@ -14,7 +14,7 @@
|
|||
"VIKUNJA_USERNAME",
|
||||
"-e",
|
||||
"VIKUNJA_PASSWORD",
|
||||
"harbor.freshbrewed.science/library/vikunjamcp:0.18"
|
||||
"harbor.freshbrewed.science/library/vikunjamcp:0.19"
|
||||
],
|
||||
"env": {
|
||||
"VIKUNJA_URL": "$VIKUNJA_URL",
|
||||
|
|
173
main.py
173
main.py
|
@ -1,4 +1,3 @@
|
|||
|
||||
import os
|
||||
import requests
|
||||
import logging
|
||||
|
@ -108,13 +107,13 @@ def search_tasks(query: str):
|
|||
logger.exception("search_tasks: request failed")
|
||||
return f"Error searching tasks: {e}"
|
||||
|
||||
|
||||
@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.
|
||||
|
||||
: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:
|
||||
return "Please run the 'login' command first."
|
||||
|
@ -145,6 +144,14 @@ 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]
|
||||
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))
|
||||
return active
|
||||
except requests.exceptions.RequestException as e:
|
||||
|
@ -167,23 +174,82 @@ def get_task_details(task_id: int):
|
|||
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 = []
|
||||
task_id = details.get("id", "N/A")
|
||||
task_created = details.get("created", "N/A")
|
||||
task_done = details.get("done", "N/A")
|
||||
task_percent_done = details.get("percent_done", "N/A")
|
||||
task_due_date = details.get("due_date", "N/A")
|
||||
task_favorite = details.get("is_favorite", "N/A")
|
||||
task_position = details.get("position", "N/A")
|
||||
task_priority = details.get("priority", "N/A")
|
||||
task_title = details.get("title", "N/A")
|
||||
task_description = details.get("description", "N/A")
|
||||
task_project_id = details.get("project_id", "N/A")
|
||||
|
||||
result.append(f"ID: {task_id}, Created: {task_created}, Done: {task_done}, Percent Done: {task_percent_done}, Due Date: {task_due_date}, Favorite: {task_favorite}, Position: {task_position}, Priority: {task_priority}, Title: {task_title}, Description: {task_description}, Project ID: {task_project_id}")
|
||||
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}"
|
||||
|
@ -460,6 +526,77 @@ def create_project(title: str, description: str = "", is_favorite: bool = False)
|
|||
logger.exception("create_project: request failed for title=%s", title)
|
||||
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__":
|
||||
print("--- Vikunja MCP Client ---")
|
||||
print("Available commands: login, search_tasks, add_task, close_task, lookup_project, help, exit")
|
||||
|
|
Loading…
Reference in New Issue