vikunjamcp/main.py

import os
import requests
import logging
from fastmcp import FastMCP

# --- Configuration ---
VIKUNJA_URL = os.getenv("VIKUNJA_URL")
VIKUNJA_USERNAME = os.getenv("VIKUNJA_USERNAME")
VIKUNJA_PASSWORD = os.getenv("VIKUNJA_PASSWORD")
LOG_LEVEL = os.getenv("LOG_LEVEL", "INFO")
DEBUG_TASK_MATCHES = os.getenv("DEBUG_TASK_MATCHES", "false").lower() in ("1", "true", "yes")

# --- MCP Application Setup ---
mcp = FastMCP()
session = requests.Session()

# Configure basic logging. Level can be controlled with LOG_LEVEL env var.
level = getattr(logging, LOG_LEVEL.upper(), logging.INFO)
logging.basicConfig(level=level, format='%(asctime)s %(levelname)s %(message)s')
logger = logging.getLogger(__name__)
if DEBUG_TASK_MATCHES:
    logger.info("DEBUG_TASK_MATCHES is enabled: per-task match attempts will be logged at INFO level")

# --- Input Validation ---
if not all([VIKUNJA_URL, VIKUNJA_USERNAME, VIKUNJA_PASSWORD]):
    print("Error: Please set the VIKUNJA_URL, VIKUNJA_USERNAME, and VIKUNJA_PASSWORD environment variables.")
    exit(1)

@mcp.tool()
def login():
    """
    Authenticates with the Vikunja API to get a session token.
    """
    global session
    try:
        response = session.post(
            f"{VIKUNJA_URL}/api/v1/login",
            json={"username": VIKUNJA_USERNAME, "password": VIKUNJA_PASSWORD}
        )
        response.raise_for_status()
        token = response.json().get("token")
        if not token:
            return "Login failed: Token not found in response."
        
        session.headers.update({"Authorization": f"Bearer {token}"})
        return "Login successful. Token stored for session."
    except requests.exceptions.RequestException as e:
        return f"Login failed: {e}"

@mcp.tool()
def search_tasks(query: str):
    """
    Searches for tasks in Vikunja.
    
    :param query: The search string to use for finding tasks.
    """
    if "Authorization" not in session.headers:
        return "Please run the 'login' command first."
    
    try:
        # Vikunja does not expose a /tasks/search endpoint in the public API.
        # Fetch all tasks and filter client-side by title/description.
        logger.info("search_tasks: fetching all tasks from %s", f"{VIKUNJA_URL}/api/v1/tasks/all")
        
        all_tasks = []
        page = 1
        per_page = 50 
        
        while True:
            response = session.get(f"{VIKUNJA_URL}/api/v1/tasks/all?page={page}&limit={per_page}")
            response.raise_for_status()
            data = response.json()
            tasks = data if isinstance(data, list) else data.get("tasks", [])
            if not tasks:
                break
            all_tasks.extend(tasks)
            page += 1

        tasks = all_tasks

        q = (query or "").strip()
        logger.info("search_tasks: raw query='%s'", q)
        if not q:
            logger.info("search_tasks: empty query, returning %d tasks", len(tasks))
            return tasks

        terms = [t.lower() for t in q.split() if t]
        logger.info("search_tasks: parsed terms=%s", terms)

        def matches(task: dict) -> bool:
            title = (task.get("title") or "")
            description = (task.get("description") or "")
            title_l = title.lower()
            desc_l = description.lower()
            for term in terms:
                if term in title_l or term in desc_l:
                    logger.debug("search_tasks: task id=%s matched term='%s' (title=%r, description=%r)", task.get("id"), term, title, description)
                    return True
                else:
                    logger.debug("search_tasks: task id=%s did not match term='%s' (title=%r)", task.get("id"), term, title)
            return False

        filtered = [t for t in tasks if matches(t)]
        logger.info("search_tasks: fetched=%d filtered=%d", len(tasks), len(filtered))
        return filtered
    except requests.exceptions.RequestException as e:
        logger.exception("search_tasks: request failed")
        return f"Error searching tasks: {e}"

@mcp.tool()
def add_task(project_id: int, title: str, description: str = ""):
    """
    Adds a new task to a Vikunja project.
    
    :param project_id: The ID of the project to add the task to.
    :param title: The title of the new task.
    :param description: An optional description for the task.
    """
    if "Authorization" not in session.headers:
        return "Please run the 'login' command first."
        
    task_payload = {
        "project_id": project_id,
        "title": title,
        "description": description
    }
    
    try:
        response = session.put(
            f"{VIKUNJA_URL}/api/v1/projects/{project_id}/tasks",
            json=task_payload
        )
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        return f"Error adding task: {e}"

@mcp.tool()
def close_task(task_id: int):
    """
    Closes (marks as done) a task in Vikunja.
    
    :param task_id: The ID of the task to close.
    """
    if "Authorization" not in session.headers:
        return "Please run the 'login' command first."
    
    task_payload = {
        "done": True
    }
    
    try:
        response = session.post(
            f"{VIKUNJA_URL}/api/v1/tasks/{task_id}",
            json=task_payload
        )
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        return f"Error closing task: {e}"

@mcp.tool()
def comment_task(task_id: int, description: str):
    """
    Adds a comment to a task in Vikunja.

    :param task_id: The ID of the task to comment on.
    :param description: The text of the comment to add.
    """
    if "Authorization" not in session.headers:
        return "Please run the 'login' command first."

    if not (description or "").strip():
        return "Comment description cannot be empty."

    payload = {"comment": description}

    try:
        response = session.put(
            f"{VIKUNJA_URL}/api/v1/tasks/{task_id}/comments",
            json=payload
        )
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        logger.exception("comment_task: request failed for task_id=%s", task_id)
        return f"Error adding comment to task: {e}"

    
    
@mcp.tool()
def update_comment(task_id: int, comment_id: int, comment: str):
    """
    Updates an existing comment on a task.

    :param task_id: The ID of the task.
    :param comment_id: The ID of the comment to update.
    :param comment: The updated comment text.
    """
    if "Authorization" not in session.headers:
        return "Please run the 'login' command first."

    if not (comment or "").strip():
        return "Comment cannot be empty."

    payload = {"comment": comment}

    try:
        response = session.post(
            f"{VIKUNJA_URL}/api/v1/tasks/{task_id}/comments/{comment_id}",
            json=payload
        )
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        logger.exception("update_comment: request failed for task_id=%s, comment_id=%s", task_id, comment_id)
        return f"Error updating comment for task: {e}"
    
@mcp.tool()
def lookup_project():
    """
    Retrieves all projects the user has access to and lists them by name and ID.
    """
    if "Authorization" not in session.headers:
        return "Please run the 'login' command first."
    
    try:
        response = session.get(f"{VIKUNJA_URL}/api/v1/projects")
        response.raise_for_status()
        projects = response.json()
        
        if not projects:
            return "No projects found."
        
        result = []
        for project in projects:
            project_id = project.get("id", "N/A")
            project_title = project.get("title", "Untitled")
            result.append(f"ID: {project_id}, Name: {project_title}")
        
        return "\n".join(result)
    except requests.exceptions.RequestException as e:
        return f"Error retrieving projects: {e}"


@mcp.tool()
def create_project(title: str, description: str = "", is_favorite: bool = False):
    """
    Creates a new project in Vikunja.

    :param title: The title of the project.
    :param description: Optional description for the project.
    :param is_favorite: Whether the project should be marked as favorite.
    """
    if "Authorization" not in session.headers:
        return "Please run the 'login' command first."

    if not (title or "").strip():
        return "Project title cannot be empty."

    payload = {
        "title": title,
        "description": description,
        "is_favourite": bool(is_favorite)
    }

    try:
        response = session.put(
            f"{VIKUNJA_URL}/api/v1/projects",
            json=payload
        )
        response.raise_for_status()
        return response.json()
    except requests.exceptions.RequestException as e:
        logger.exception("create_project: request failed for title=%s", title)
        return f"Error creating project: {e}"

if __name__ == "__main__":
    print("--- Vikunja MCP Client ---")
    print("Available commands: login, search_tasks, add_task, close_task, lookup_project, help, exit")
    mcp.run()