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") # Set a reasonable timeout to prevent OpenClaw from hanging DEFAULT_TIMEOUT = 5.0 # 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, timeout=DEFAULT_TIMEOUT ) 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, timeout=DEFAULT_TIMEOUT ) 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}") if isinstance(data, dict) and "included" in data and "cards" in data["included"]: return data["included"]["cards"] return [] 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 def safe_tool_call(func, *args, **kwargs): """Wrapper to handle network errors gracefully.""" try: return func(*args, **kwargs) except requests.exceptions.Timeout: return "Error: Planka server timed out. It might be down or slow." except requests.exceptions.ConnectionError: return "Error: Could not connect to Planka server. Please check the URL and connectivity." except requests.exceptions.HTTPError as e: return f"Error: Planka API returned an error: {e}" except Exception as e: return f"Error: An unexpected error occurred: {e}" @mcp.tool() def list_projects() -> str: """List all Planka projects available to the user.""" def _logic(): 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]) return safe_tool_call(_logic) @mcp.tool() def list_boards(project_id: str) -> str: """List all boards for a given project ID.""" def _logic(): 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]) return safe_tool_call(_logic) @mcp.tool() def list_board_columns(board_id: str) -> str: """List all columns (lists) for a given board ID.""" def _logic(): 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]) return safe_tool_call(_logic) @mcp.tool() def list_cards(board_id: str) -> str: """List all cards for a given board ID.""" def _logic(): 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]) return safe_tool_call(_logic) @mcp.tool() def create_card(board_id: str, list_id: str, name: str, description: str = "") -> str: """Create a new card on a Planka board.""" def _logic(): card = get_client().create_card(board_id, list_id, name, description) return f"Created card: {card['name']} (ID: {card['id']})" return safe_tool_call(_logic) @mcp.tool() def list_comments(card_id: str) -> str: """List all comments for a given card ID.""" def _logic(): 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]) return safe_tool_call(_logic) @mcp.tool() def add_comment(card_id: str, text: str) -> str: """Add a comment to a card.""" def _logic(): get_client().create_comment(card_id, text) return f"Comment added to card {card_id}." return safe_tool_call(_logic) @mcp.tool() def check_planka_status() -> str: """Check if the Planka server is responsive.""" try: response = requests.get(f"{PLANKA_URL}/api/projects", headers={"Authorization": f"Bearer {PLANKA_TOKEN}"}, timeout=DEFAULT_TIMEOUT) if response.status_code == 200: return f"Planka is UP and reachable at {PLANKA_URL}" else: return f"Planka is reachable but returned status {response.status_code}" except Exception as e: return f"Planka is UNREACHABLE: {e}" 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])) elif cmd == "status": print(check_planka_status()) else: mcp.run()