Flash/planka-mcp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add timeouts, error handling and status tool to server

Browse Source
This commit is contained in:
Flash
2026-04-09 11:13:13 +00:00
parent f0c586c88c
commit eba1b919ed
1 changed files with 84 additions and 31 deletions
Download Patch File Download Diff File Expand all files Collapse all files

115
server.py
View File

@@ -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()