134 lines
4.4 KiB
Python
134 lines
4.4 KiB
Python
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()
|
|
return response.json()
|
|
|
|
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()
|
|
return response.json()
|
|
|
|
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}")
|
|
if isinstance(data, dict):
|
|
if "boards" in data:
|
|
return data["boards"]
|
|
if "included" in data and isinstance(data["included"], dict) and "boards" in data["included"]:
|
|
return data["included"]["boards"]
|
|
return []
|
|
|
|
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]:
|
|
# Planka comments are in the 'actions' endpoint for a card
|
|
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_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__":
|
|
mcp.run()
|