fixing-search #1
|
@ -8,6 +8,10 @@ ENV VIKUNJA_URL=$VIKUNJA_URL
|
|||
ENV VIKUNJA_USERNAME=$VIKUNJA_USERNAME
|
||||
ENV VIKUNJA_PASSWORD=$VIKUNJA_PASSWORD
|
||||
|
||||
# Comment out for less verbose debug
|
||||
ENV LOG_LEVEL=DEBUG
|
||||
ENV DEBUG_TASK_MATCHES=true
|
||||
|
||||
WORKDIR /app
|
||||
|
||||
COPY requirements.txt requirements.txt
|
||||
|
|
46
main.py
46
main.py
|
@ -1,17 +1,27 @@
|
|||
|
||||
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.")
|
||||
|
@ -49,10 +59,42 @@ def search_tasks(query: str):
|
|||
return "Please run the 'login' command first."
|
||||
|
||||
try:
|
||||
response = session.get(f"{VIKUNJA_URL}/api/v1/tasks/search?query={query}")
|
||||
# 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")
|
||||
response = session.get(f"{VIKUNJA_URL}/api/v1/tasks/all")
|
||||
response.raise_for_status()
|
||||
return response.json()
|
||||
data = response.json()
|
||||
# The API might return a list or an object with a 'tasks' key
|
||||
tasks = data if isinstance(data, list) else data.get("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()
|
||||
|
|
Loading…
Reference in New Issue