diff --git a/server.py b/server.py index 6d33030..239d499 100644 --- a/server.py +++ b/server.py @@ -9,6 +9,8 @@ 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") @@ -20,7 +22,12 @@ class PlankaClient: 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 = 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 @@ -30,7 +37,12 @@ class PlankaClient: return None def _post(self, endpoint: str, data: dict) -> Any: - response = requests.post(f"{self.url}/api/{endpoint}", headers=self.headers, json=data) + 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 @@ -70,10 +82,10 @@ class PlankaClient: 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 + 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", { @@ -108,58 +120,97 @@ def get_client(): 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.""" - 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]) + 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.""" - 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]) + 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.""" - 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]) + 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.""" - 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]) + 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.""" - card = get_client().create_card(board_id, list_id, name, description) - return f"Created card: {card['name']} (ID: {card['id']})" + 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.""" - 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]) + 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.""" - action = get_client().create_comment(card_id, text) - return f"Comment added to card {card_id}." + 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: @@ -174,5 +225,7 @@ if __name__ == "__main__": 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()