90 lines
2.7 KiB
Python
90 lines
2.7 KiB
Python
|
|
import os
|
|
import requests
|
|
from fastmcp import FastMCP
|
|
|
|
# --- Configuration ---
|
|
VIKUNJA_URL = os.getenv("VIKUNJA_URL")
|
|
VIKUNJA_USERNAME = os.getenv("VIKUNJA_USERNAME")
|
|
VIKUNJA_PASSWORD = os.getenv("VIKUNJA_PASSWORD")
|
|
|
|
# --- MCP Application Setup ---
|
|
mcp = FastMCP()
|
|
session = requests.Session()
|
|
|
|
# --- 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:
|
|
response = session.get(f"{VIKUNJA_URL}/api/v1/tasks/search?query={query}")
|
|
response.raise_for_status()
|
|
return response.json()
|
|
except requests.exceptions.RequestException as e:
|
|
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.post(
|
|
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}"
|
|
|
|
if __name__ == "__main__":
|
|
print("--- Vikunja MCP Client ---")
|
|
print("Available commands: login, search_tasks, add_task, help, exit")
|
|
mcp.run()
|