Orchestrator/.kiro/mcp-servers/orgmylife-api/server.py

"""OrgMyLife MCP Server — provides tools to interact with the OrgMyLife API.

Tools:
- list_projects: Get all project tasks from the kanban board
- update_project: Move a project task to a different status column
- create_project: Create a new project task
- list_tasks: Get open tasks (with optional filters)
- list_calls: Get pending call items
"""

import os
import httpx
from mcp.server.fastmcp import FastMCP

# Configuration from environment
BASE_URL = os.environ.get("ORGMYLIFE_BASE_URL", "https://api.andreknie.de")
API_SECRET = os.environ.get("ORGMYLIFE_API_SECRET", "")

mcp = FastMCP("orgmylife-api", instructions="OrgMyLife project management API. Use these tools to read and update the projects board, task list, and call list.")


def _headers():
    """Build auth headers for the OrgMyLife API."""
    return {
        "Authorization": f"Bearer {API_SECRET}",
        "X-API-Key": API_SECRET,
        "Content-Type": "application/json",
    }


def _get(path: str, params: dict = None) -> dict:
    """Make an authenticated GET request to the OrgMyLife API."""
    r = httpx.get(f"{BASE_URL}{path}", headers=_headers(), params=params, timeout=15)
    r.raise_for_status()
    return r.json()


def _post(path: str, json_body: dict = None) -> dict:
    """Make an authenticated POST request to the OrgMyLife API."""
    r = httpx.post(f"{BASE_URL}{path}", headers=_headers(), json=json_body, timeout=15)
    r.raise_for_status()
    return r.json()


def _put(path: str, json_body: dict = None) -> dict:
    """Make an authenticated PUT request to the OrgMyLife API."""
    r = httpx.put(f"{BASE_URL}{path}", headers=_headers(), json=json_body, timeout=15)
    r.raise_for_status()
    return r.json()


def _delete(path: str) -> dict:
    """Make an authenticated DELETE request to the OrgMyLife API."""
    r = httpx.delete(f"{BASE_URL}{path}", headers=_headers(), timeout=15)
    r.raise_for_status()
    return r.json()


# --- Projects Board ---

@mcp.tool()
def list_projects() -> str:
    """Get all project tasks from the kanban board, grouped by status column."""
    data = _get("/api/projects")
    tasks = data.get("tasks", [])
    if not tasks:
        return "No project tasks found."

    # Group by status
    columns = {}
    for t in tasks:
        status = t.get("status", "backlog")
        columns.setdefault(status, []).append(t)

    lines = []
    status_order = ["backlog", "ready", "in_progress", "blocked", "review", "done"]
    status_labels = {
        "backlog": "📋 Backlog",
        "ready": "🟢 Ready",
        "in_progress": "⚙️ In Progress",
        "blocked": "🚫 Blocked",
        "review": "👁 Review",
        "done": "✅ Done",
    }
    for status in status_order:
        items = columns.get(status, [])
        lines.append(f"\n## {status_labels.get(status, status)} ({len(items)})")
        if items:
            for t in items:
                repo = f" [{t['repo']}]" if t.get("repo") else ""
                lines.append(f"  - #{t['id']}: {t['title']}{repo}")
        else:
            lines.append("  (empty)")

    return "\n".join(lines)


@mcp.tool()
def update_project(task_id: int, status: str) -> str:
    """Move a project task to a different status column.

    Args:
        task_id: The project task ID (from list_projects)
        status: Target column — one of: backlog, ready, in_progress, blocked, review, done
    """
    valid = ["backlog", "ready", "in_progress", "blocked", "review", "done"]
    if status not in valid:
        return f"Invalid status '{status}'. Must be one of: {', '.join(valid)}"
    result = _put(f"/api/projects/{task_id}", {"status": status})
    return f"Moved project task #{task_id} to '{status}'. Result: {result.get('status', 'unknown')}"


@mcp.tool()
def create_project(title: str, repo: str = "") -> str:
    """Create a new project task on the kanban board (starts in Backlog).

    Args:
        title: Task title
        repo: Repository name (e.g., "OrgMyLife", "AI-Orchestrator")
    """
    result = _post("/api/projects/create", {"title": title, "repo": repo})
    if result.get("status") == "success":
        return f"Created project task: '{title}' (ID: {result.get('id')})"
    return f"Failed to create project task: {result}"


# --- Tasks ---

@mcp.tool()
def list_tasks(filter_origin: str = "", filter_label: str = "", sort: str = "priority") -> str:
    """Get open tasks from the task list.

    Args:
        filter_origin: Filter by source — "manual", "nextcloud_todo", "dhive_email", "gmail", or "" for all
        filter_label: Filter by label name, or "" for all
        sort: Sort order — "priority", "due_date", "created", or "" for manual order
    """
    params = {}
    if sort:
        params["sort"] = sort
    if filter_origin:
        params["filter_origin"] = filter_origin
    if filter_label:
        params["filter_label"] = filter_label

    data = _get("/api/tasks", params)
    tasks = data.get("tasks", [])
    total = data.get("total", len(tasks))

    if not tasks:
        return "No open tasks."

    lines = [f"**{total} open tasks:**\n"]
    for t in tasks[:30]:  # Limit to 30 to avoid huge output
        priority = t.get("priority", "")
        due = f" (due: {t['due_date']})" if t.get("due_date") else ""
        labels = f" [{t['labels']}]" if t.get("labels") else ""
        conflict = " ⚠️CONFLICT" if t.get("has_conflict") else ""
        lines.append(f"  #{t['id']}: {t['title']} — {priority}{due}{labels}{conflict}")

    if total > 30:
        lines.append(f"\n  ... and {total - 30} more")

    return "\n".join(lines)


# --- Calls ---

@mcp.tool()
def list_calls() -> str:
    """Get all pending call items from the call list."""
    data = _get("/api/calls")
    calls = data.get("calls", [])

    if not calls:
        return "No pending calls."

    lines = [f"**{len(calls)} pending calls:**\n"]
    for c in calls:
        phone = f" ({c['phone_number']})" if c.get("phone_number") else ""
        lines.append(f"  #{c['id']}: {c['person']}{phone} — {c['reason']}")

    return "\n".join(lines)


# --- Daily Digest ---

@mcp.tool()
def get_digest() -> str:
    """Get the daily digest summary (overdue, due today, top priorities, schedule)."""
    data = _get("/api/digest")

    lines = [f"**{data.get('greeting', 'Hello')}** — {data.get('date', '')}"]
    lines.append(f"Total open tasks: {data.get('total_open', '?')}\n")

    overdue = data.get("overdue", [])
    if overdue:
        lines.append(f"⚠️ **Overdue ({len(overdue)}):**")
        for t in overdue:
            lines.append(f"  - {t['title']} ({t['days_overdue']}d overdue)")

    due_today = data.get("due_today", [])
    if due_today:
        lines.append(f"\n📅 **Due Today ({len(due_today)}):**")
        for t in due_today:
            lines.append(f"  - {t['title']}")

    top = data.get("top_tasks", [])
    if top:
        lines.append(f"\n🎯 **Top Priorities:**")
        for t in top:
            lines.append(f"  - {t['title']} ({t['priority']})")

    events = data.get("events_today", [])
    if events:
        lines.append(f"\n📆 **Today's Schedule:**")
        for e in events:
            lines.append(f"  - {e['time']} {e['title']}")

    calls = data.get("pending_calls", [])
    if calls:
        lines.append(f"\n📞 **Calls to make ({len(calls)}):**")
        for c in calls:
            lines.append(f"  - {c['person']} — {c['reason']}")

    return "\n".join(lines)


if __name__ == "__main__":
    mcp.run()