diff --git a/penpot_mcp/api/penpot_api.py b/penpot_mcp/api/penpot_api.py index fb90067..45a31af 100644 --- a/penpot_mcp/api/penpot_api.py +++ b/penpot_mcp/api/penpot_api.py @@ -7,6 +7,28 @@ import requests from dotenv import load_dotenv +class CloudFlareError(Exception): + """Exception raised when CloudFlare protection blocks the request.""" + + def __init__(self, message: str, status_code: int = None, response_text: str = None): + super().__init__(message) + self.status_code = status_code + self.response_text = response_text + + def __str__(self): + return f"CloudFlare Protection Error: {super().__str__()}" + + +class PenpotAPIError(Exception): + """General exception for Penpot API errors.""" + + def __init__(self, message: str, status_code: int = None, response_text: str = None, is_cloudflare: bool = False): + super().__init__(message) + self.status_code = status_code + self.response_text = response_text + self.is_cloudflare = is_cloudflare + + class PenpotAPI: def __init__( self, @@ -35,6 +57,70 @@ class PenpotAPI: "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" }) + def _is_cloudflare_error(self, response: requests.Response) -> bool: + """Check if the response indicates a CloudFlare error.""" + # Check for CloudFlare-specific indicators + cloudflare_indicators = [ + 'cloudflare', + 'cf-ray', + 'attention required', + 'checking your browser', + 'challenge', + 'ddos protection', + 'security check', + 'cf-browser-verification', + 'cf-challenge-running', + 'please wait while we are checking your browser', + 'enable cookies and reload the page', + 'this process is automatic' + ] + + # Check response headers for CloudFlare + server_header = response.headers.get('server', '').lower() + cf_ray = response.headers.get('cf-ray') + + if 'cloudflare' in server_header or cf_ray: + return True + + # Check response content for CloudFlare indicators + try: + response_text = response.text.lower() + for indicator in cloudflare_indicators: + if indicator in response_text: + return True + except: + # If we can't read the response text, don't assume it's CloudFlare + pass + + # Check for specific status codes that might indicate CloudFlare blocks + if response.status_code in [403, 429, 503]: + # Additional check for CloudFlare-specific error pages + try: + response_text = response.text.lower() + if any(['cloudflare' in response_text, 'cf-ray' in response_text, 'attention required' in response_text]): + return True + except: + pass + + return False + + def _create_cloudflare_error_message(self, response: requests.Response) -> str: + """Create a user-friendly CloudFlare error message.""" + base_message = ( + "CloudFlare protection has blocked this request. This is common on penpot.app. " + "To resolve this issue:\\n\\n" + "1. Open your web browser and navigate to https://design.penpot.app\\n" + "2. Log in to your Penpot account\\n" + "3. Complete any CloudFlare human verification challenges if prompted\\n" + "4. Once verified, try your request again\\n\\n" + "The verification typically lasts for a period of time, after which you may need to repeat the process." + ) + + if response.status_code: + return f"{base_message}\\n\\nHTTP Status: {response.status_code}" + + return base_message + def set_access_token(self, token: str): """Set the auth token for authentication.""" self.access_token = token @@ -310,6 +396,11 @@ class PenpotAPI: return response except requests.HTTPError as e: + # Check for CloudFlare errors first + if self._is_cloudflare_error(e.response): + cloudflare_message = self._create_cloudflare_error_message(e.response) + raise CloudFlareError(cloudflare_message, e.response.status_code, e.response.text) + # Handle authentication errors if e.response.status_code in (401, 403) and self.email and self.password and retry_auth: # Special case: don't retry auth for get-profile to avoid infinite loops @@ -333,6 +424,15 @@ class PenpotAPI: else: # Re-raise other errors raise + except requests.RequestException as e: + # Handle other request exceptions (connection errors, timeouts, etc.) + # Check if we have a response to analyze + if hasattr(e, 'response') and e.response is not None: + if self._is_cloudflare_error(e.response): + cloudflare_message = self._create_cloudflare_error_message(e.response) + raise CloudFlareError(cloudflare_message, e.response.status_code, e.response.text) + # Re-raise if not a CloudFlare error + raise def _normalize_transit_response(self, data: Union[Dict, List, Any]) -> Union[Dict, List, Any]: """ diff --git a/penpot_mcp/server/mcp_server.py b/penpot_mcp/server/mcp_server.py index 94d1410..9c74bee 100644 --- a/penpot_mcp/server/mcp_server.py +++ b/penpot_mcp/server/mcp_server.py @@ -15,7 +15,7 @@ from typing import Dict, List, Optional from mcp.server.fastmcp import FastMCP, Image -from penpot_mcp.api.penpot_api import PenpotAPI +from penpot_mcp.api.penpot_api import PenpotAPI, CloudFlareError, PenpotAPIError from penpot_mcp.tools.penpot_tree import get_object_subtree_with_fields from penpot_mcp.utils import config from penpot_mcp.utils.cache import MemoryCache @@ -91,6 +91,30 @@ Let me know which Penpot design you'd like to convert to code, and I'll guide yo else: self._register_resources(resources_only=False) self._register_tools(include_resource_tools=False) + + def _handle_api_error(self, e: Exception) -> dict: + """Handle API errors and return user-friendly error messages.""" + if isinstance(e, CloudFlareError): + return { + "error": "CloudFlare Protection", + "message": str(e), + "error_type": "cloudflare_protection", + "instructions": [ + "Open your web browser and navigate to https://design.penpot.app", + "Log in to your Penpot account", + "Complete any CloudFlare human verification challenges if prompted", + "Once verified, try your request again" + ] + } + elif isinstance(e, PenpotAPIError): + return { + "error": "Penpot API Error", + "message": str(e), + "error_type": "api_error", + "status_code": getattr(e, 'status_code', None) + } + else: + return {"error": str(e)} def _register_resources(self, resources_only=False): """Register all MCP resources. If resources_only is True, only register server://info as a resource.""" @@ -148,7 +172,7 @@ Let me know which Penpot design you'd like to convert to code, and I'll guide yo projects = self.api.list_projects() return {"projects": projects} except Exception as e: - return {"error": str(e)} + return self._handle_api_error(e) @self.mcp.tool() def get_project_files(project_id: str) -> dict: """Get all files contained within a specific Penpot project. @@ -160,7 +184,7 @@ Let me know which Penpot design you'd like to convert to code, and I'll guide yo files = self.api.get_project_files(project_id) return {"files": files} except Exception as e: - return {"error": str(e)} + return self._handle_api_error(e) def get_cached_file(file_id: str) -> dict: """Internal helper to retrieve a file, using cache if available. @@ -175,7 +199,7 @@ Let me know which Penpot design you'd like to convert to code, and I'll guide yo self.file_cache.set(file_id, file_data) return file_data except Exception as e: - return {"error": str(e)} + return self._handle_api_error(e) @self.mcp.tool() def get_file(file_id: str) -> dict: """Retrieve a Penpot file by its ID and cache it. Don't use this tool for code generation, use 'get_object_tree' instead. @@ -188,7 +212,7 @@ Let me know which Penpot design you'd like to convert to code, and I'll guide yo self.file_cache.set(file_id, file_data) return file_data except Exception as e: - return {"error": str(e)} + return self._handle_api_error(e) @self.mcp.tool() def export_object( file_id: str, @@ -233,7 +257,10 @@ Let me know which Penpot design you'd like to convert to code, and I'll guide yo return image except Exception as e: - raise Exception(f"Export failed: {str(e)}") + if isinstance(e, CloudFlareError): + raise Exception(f"CloudFlare Protection: {str(e)}") + else: + raise Exception(f"Export failed: {str(e)}") finally: if temp_filename and os.path.exists(temp_filename): try: @@ -309,7 +336,7 @@ Let me know which Penpot design you'd like to convert to code, and I'll guide yo return {"format_error": f"Error formatting as YAML: {str(e)}"} return final_result except Exception as e: - return {"error": str(e)} + return self._handle_api_error(e) @self.mcp.tool() def search_object(file_id: str, query: str) -> dict: """Search for objects within a Penpot file by name. @@ -339,7 +366,7 @@ Let me know which Penpot design you'd like to convert to code, and I'll guide yo }) return {'objects': matches} except Exception as e: - return {"error": str(e)} + return self._handle_api_error(e) if include_resource_tools: @self.mcp.tool() def penpot_schema() -> dict: