Add comprehensive CloudFlare error detection and user-friendly error handling

- Add CloudFlareError and PenpotAPIError exception classes to penpot_api.py
- Implement _is_cloudflare_error() method to detect CloudFlare protection blocks
- Add _create_cloudflare_error_message() to provide helpful user instructions
- Update _make_authenticated_request() to catch and handle CloudFlare errors
- Add _handle_api_error() method to MCP server for consistent error formatting
- Update all MCP tool methods to use enhanced error handling
- Provide clear instructions for resolving CloudFlare verification challenges
- Include error_type field for better error categorization in MCP responses

🤖 Generated with [Claude Code](https://claude.ai/code)

Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
chema
2025-06-29 20:04:20 +02:00
parent a5517d6b95
commit d8ed2fac70
2 changed files with 135 additions and 8 deletions

View File

@@ -7,6 +7,28 @@ import requests
from dotenv import load_dotenv 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: class PenpotAPI:
def __init__( def __init__(
self, 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" "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): def set_access_token(self, token: str):
"""Set the auth token for authentication.""" """Set the auth token for authentication."""
self.access_token = token self.access_token = token
@@ -310,6 +396,11 @@ class PenpotAPI:
return response return response
except requests.HTTPError as e: 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 # Handle authentication errors
if e.response.status_code in (401, 403) and self.email and self.password and retry_auth: 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 # Special case: don't retry auth for get-profile to avoid infinite loops
@@ -333,6 +424,15 @@ class PenpotAPI:
else: else:
# Re-raise other errors # Re-raise other errors
raise 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]: def _normalize_transit_response(self, data: Union[Dict, List, Any]) -> Union[Dict, List, Any]:
""" """

View File

@@ -15,7 +15,7 @@ from typing import Dict, List, Optional
from mcp.server.fastmcp import FastMCP, Image 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.tools.penpot_tree import get_object_subtree_with_fields
from penpot_mcp.utils import config from penpot_mcp.utils import config
from penpot_mcp.utils.cache import MemoryCache 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: else:
self._register_resources(resources_only=False) self._register_resources(resources_only=False)
self._register_tools(include_resource_tools=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): def _register_resources(self, resources_only=False):
"""Register all MCP resources. If resources_only is True, only register server://info as a resource.""" """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() projects = self.api.list_projects()
return {"projects": projects} return {"projects": projects}
except Exception as e: except Exception as e:
return {"error": str(e)} return self._handle_api_error(e)
@self.mcp.tool() @self.mcp.tool()
def get_project_files(project_id: str) -> dict: def get_project_files(project_id: str) -> dict:
"""Get all files contained within a specific Penpot project. """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) files = self.api.get_project_files(project_id)
return {"files": files} return {"files": files}
except Exception as e: except Exception as e:
return {"error": str(e)} return self._handle_api_error(e)
def get_cached_file(file_id: str) -> dict: def get_cached_file(file_id: str) -> dict:
"""Internal helper to retrieve a file, using cache if available. """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) self.file_cache.set(file_id, file_data)
return file_data return file_data
except Exception as e: except Exception as e:
return {"error": str(e)} return self._handle_api_error(e)
@self.mcp.tool() @self.mcp.tool()
def get_file(file_id: str) -> dict: 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. """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) self.file_cache.set(file_id, file_data)
return file_data return file_data
except Exception as e: except Exception as e:
return {"error": str(e)} return self._handle_api_error(e)
@self.mcp.tool() @self.mcp.tool()
def export_object( def export_object(
file_id: str, 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 return image
except Exception as e: 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: finally:
if temp_filename and os.path.exists(temp_filename): if temp_filename and os.path.exists(temp_filename):
try: 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 {"format_error": f"Error formatting as YAML: {str(e)}"}
return final_result return final_result
except Exception as e: except Exception as e:
return {"error": str(e)} return self._handle_api_error(e)
@self.mcp.tool() @self.mcp.tool()
def search_object(file_id: str, query: str) -> dict: def search_object(file_id: str, query: str) -> dict:
"""Search for objects within a Penpot file by name. """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} return {'objects': matches}
except Exception as e: except Exception as e:
return {"error": str(e)} return self._handle_api_error(e)
if include_resource_tools: if include_resource_tools:
@self.mcp.tool() @self.mcp.tool()
def penpot_schema() -> dict: def penpot_schema() -> dict: