planka-mcp/server.py

import os
import requests
import sys
from typing import Any, List, Optional
from mcp.server.fastmcp import FastMCP
from dotenv import load_dotenv

load_dotenv()

PLANKA_URL = os.getenv("PLANKA_URL", "http://localhost:3000")
PLANKA_TOKEN = os.getenv("PLANKA_TOKEN")

# Initialize FastMCP server
mcp = FastMCP("Planka")

class PlankaClient:
    def __init__(self, url: str, token: str):
        self.url = url.rstrip("/")
        self.token = token
        self.headers = {"Authorization": f"Bearer {token}"}

    def _get(self, endpoint: str, params: Optional[dict] = None) -> Any:
        response = requests.get(f"{self.url}/api/{endpoint}", headers=self.headers, params=params)
        response.raise_for_status()
        if not response.text:
            return None
        try:
            return response.json()
        except requests.exceptions.JSONDecodeError:
            return None

    def _post(self, endpoint: str, data: dict) -> Any:
        response = requests.post(f"{self.url}/api/{endpoint}", headers=self.headers, json=data)
        response.raise_for_status()
        if not response.text:
            return None
        try:
            return response.json()
        except requests.exceptions.JSONDecodeError:
            return None

    def get_projects(self) -> List[dict]:
        data = self._get("projects")
        if isinstance(data, dict) and "items" in data:
            return data["items"]
        return data

    def get_boards(self, project_id: str) -> List[dict]:
        data = self._get(f"projects/{project_id}")
        boards = []
        if isinstance(data, dict):
            if "boards" in data:
                boards.extend(data["boards"])
            if "item" in data and isinstance(data["item"], dict) and "boards" in data["item"]:
                boards.extend(data["item"]["boards"])
            if "included" in data and isinstance(data["included"], dict) and "boards" in data["included"]:
                boards.extend(data["included"]["boards"])
        return boards

    def get_board_lists(self, board_id: str) -> List[dict]:
        data = self._get(f"boards/{board_id}")
        lists = []
        if isinstance(data, dict):
            if "lists" in data:
                lists.extend(data["lists"])
            if "item" in data and isinstance(data["item"], dict) and "lists" in data["item"]:
                lists.extend(data["item"]["lists"])
            if "included" in data and isinstance(data["included"], dict) and "lists" in data["included"]:
                lists.extend(data["included"]["lists"])
        return lists

    def get_cards(self, board_id: str) -> List[dict]:
        data = self._get(f"boards/{board_id}/cards")
        if isinstance(data, dict) and "items" in data:
            return data["items"]
        return data

    def create_card(self, board_id: str, list_id: str, name: str, description: str = "") -> dict:
        return self._post("cards", {
            "boardId": board_id,
            "listId": list_id,
            "name": name,
            "description": description
        })

    def get_actions(self, card_id: str) -> List[dict]:
        data = self._get(f"cards/{card_id}/actions")
        if isinstance(data, dict) and "items" in data:
            return data["items"]
        return data

    def create_comment(self, card_id: str, text: str) -> dict:
        return self._post(f"cards/{card_id}/actions", {
            "type": "commentCard",
            "data": {
                "text": text
            }
        })

# Initialize client lazily
client = None

def get_client():
    global client
    if client is None:
        if not PLANKA_TOKEN:
            raise ValueError("PLANKA_TOKEN is required")
        client = PlankaClient(PLANKA_URL, PLANKA_TOKEN)
    return client

@mcp.tool()
def list_projects() -> str:
    """List all Planka projects available to the user."""
    projects = get_client().get_projects()
    if not projects:
        return "No projects found."
    return "\n".join([f"- {p['name']} (ID: {p['id']})" for p in projects])

@mcp.tool()
def list_boards(project_id: str) -> str:
    """List all boards for a given project ID."""
    boards = get_client().get_boards(project_id)
    if not boards:
        return "No boards found."
    return "\n".join([f"- {b['name']} (ID: {b['id']})" for b in boards])

@mcp.tool()
def list_board_columns(board_id: str) -> str:
    """List all columns (lists) for a given board ID."""
    lists = get_client().get_board_lists(board_id)
    if not lists:
        return "No columns found."
    return "\n".join([f"- {l['name']} (ID: {l['id']})" for l in lists])

@mcp.tool()
def list_cards(board_id: str) -> str:
    """List all cards for a given board ID."""
    cards = get_client().get_cards(board_id)
    if not cards:
        return "No cards found."
    return "\n".join([f"- {c['name']} (ID: {c['id']})" for c in cards])

@mcp.tool()
def create_card(board_id: str, list_id: str, name: str, description: str = "") -> str:
    """Create a new card on a Planka board."""
    card = get_client().create_card(board_id, list_id, name, description)
    return f"Created card: {card['name']} (ID: {card['id']})"

@mcp.tool()
def list_comments(card_id: str) -> str:
    """List all comments for a given card ID."""
    actions = get_client().get_actions(card_id)
    comments = [a for a in actions if a.get('type') == 'commentCard']
    if not comments:
        return "No comments found."
    return "\n".join([f"- {c['data']['text']} (By: {c['userId']}, at: {c['createdAt']})" for c in comments])

@mcp.tool()
def add_comment(card_id: str, text: str) -> str:
    """Add a comment to a card."""
    action = get_client().create_comment(card_id, text)
    return f"Comment added to card {card_id}."

if __name__ == "__main__":
    if len(sys.argv) > 1:
        # CLI debug mode
        cmd = sys.argv[1]
        c = get_client()
        if cmd == "list_projects":
            print(list_projects())
        elif cmd == "list_boards":
            print(list_boards(sys.argv[2]))
        elif cmd == "list_cards":
            print(list_cards(sys.argv[2]))
        elif cmd == "list_columns":
            print(list_board_columns(sys.argv[2]))
    else:
        mcp.run()