# #crew_tool.py import json import os import requests from crewai.tools import tool from dotenv import load_dotenv # Load environment variables load_dotenv() def get_landing_ai_api_key(): """ Get Landing AI API key from environment variables. Tries multiple sources for compatibility with different deployment environments. Returns: str: The API key if found, None otherwise """ # Try different environment variable names for flexibility api_key = ( os.getenv("LANDING_AI_API_KEY") or # HuggingFace secrets or production os.getenv("LANDING_AI_API_KEY_LOCAL") or # Local development alternative os.getenv("LANDINGAI_API_KEY") or # Alternative naming os.getenv("LANDING_API_KEY") # Another alternative ) return api_key def safe_format_with_indicators(text): """Replace emojis with text indicators""" if not isinstance(text, str): text = str(text) emoji_replacements = { '🔋': '[BATTERY]', '✅': '[SUCCESS]', '❌': '[ERROR]', '📄': '[DOCUMENT]', '📊': '[CHART]', '⚠️': '[WARNING]', '🔍': '[SEARCH]', '💡': '[INSIGHT]', '🎯': '[TARGET]', '✨': '[HIGHLIGHT]', '🚀': '[LAUNCH]', '📈': '[GROWTH]', '📉': '[DECLINE]', '🔧': '[TOOL]', '💻': '[CODE]', '🌟': '[STAR]', '🎨': '[DESIGN]', '📱': '[MOBILE]', '🖥️': '[DESKTOP]', '🔗': '[LINK]', '📝': '[NOTE]', '🏆': '[ACHIEVEMENT]', '💰': '[MONEY]', '🎉': '[CELEBRATION]', '🔒': '[SECURE]', '🔓': '[UNLOCKED]', '⭐': '[RATING]', '❤️': '[FAVORITE]', '👍': '[THUMBS_UP]', '👎': '[THUMBS_DOWN]', '🔥': '[HOT]', '❄️': '[COLD]', '🌡️': '[TEMPERATURE]' } for emoji, replacement in emoji_replacements.items(): text = text.replace(emoji, replacement) return text def format_api_response(response_data): """ Format API response and replace emojis with safe indicators. Args: response_data: Raw API response (dict, str, or other) Returns: str: Formatted response with safe emoji replacements """ try: if isinstance(response_data, dict): # Convert dict to readable format formatted_parts = [] for key, value in response_data.items(): if isinstance(value, (str, int, float)): safe_value = safe_format_with_indicators(str(value)) formatted_parts.append(f"{key}: {safe_value}") elif isinstance(value, list): safe_items = [safe_format_with_indicators(str(item)) for item in value] formatted_parts.append(f"{key}: {', '.join(safe_items)}") elif isinstance(value, dict): # Handle nested dictionaries nested_items = [] for nested_key, nested_value in value.items(): safe_nested = safe_format_with_indicators(str(nested_value)) nested_items.append(f"{nested_key}: {safe_nested}") formatted_parts.append(f"{key}: {'; '.join(nested_items)}") result = "\n".join(formatted_parts) return safe_format_with_indicators(result) else: return safe_format_with_indicators(str(response_data)) except Exception as e: error_msg = f"Error formatting response: {str(e)}" return safe_format_with_indicators(error_msg) @tool("LandingAI Document Analysis") def landing_ai_document_analysis(file_path: str, file_type: str = "image") -> str: """ Analyze images or PDFs using LandingAI's document analysis API. Args: file_path (str): Path to the image or PDF file to analyze file_type (str): Type of file, either "image" or "pdf" Returns: str: Analysis results from the API """ try: # Get API key with improved error handling api_key = get_landing_ai_api_key() if not api_key: error_msg = "Landing AI API key not found. Please ensure LANDING_AI_API_KEY is set in environment variables or HuggingFace secrets." return safe_format_with_indicators(f"[ERROR] {error_msg}") # Validate file exists if not os.path.exists(file_path): error_msg = f"File not found: {file_path}" return safe_format_with_indicators(f"[ERROR] {error_msg}") # API endpoint url = "https://api.va.landing.ai/v1/tools/agentic-document-analysis" # Prepare the file for upload based on file_type try: with open(file_path, "rb") as file_obj: if file_type.lower() == "pdf": files = {"pdf": file_obj} else: files = {"image": file_obj} # Prepare headers with authentication headers = {"Authorization": f"Basic {api_key}"} # Make the API request with timeout response = requests.post( url, files=files, headers=headers, timeout=60 # 60 second timeout ) # Handle different response status codes if response.status_code == 200: try: response_json = response.json() formatted_result = format_api_response(response_json) return safe_format_with_indicators(formatted_result) except json.JSONDecodeError: # If response isn't JSON, format the text response safe_text = safe_format_with_indicators(response.text) return safe_format_with_indicators(safe_text) elif response.status_code == 401: error_msg = "Authentication failed. Please check your Landing AI API key." return safe_format_with_indicators(f"[ERROR] {error_msg}") elif response.status_code == 403: error_msg = "Access forbidden. Your API key may not have the required permissions." return safe_format_with_indicators(f"[ERROR] {error_msg}") elif response.status_code == 429: error_msg = "Rate limit exceeded. Please try again later." return safe_format_with_indicators(f"[ERROR] {error_msg}") elif response.status_code == 500: error_msg = "Server error from Landing AI. Please try again later." return safe_format_with_indicators(f"[ERROR] {error_msg}") else: error_msg = f"API request failed with status code {response.status_code}: {response.text}" return safe_format_with_indicators(f"[ERROR] {error_msg}") except FileNotFoundError: error_msg = f"File not found: {file_path}" return safe_format_with_indicators(f"[ERROR] {error_msg}") except requests.exceptions.Timeout: error_msg = "Request timed out. The file may be too large or the service is slow." return safe_format_with_indicators(f"[ERROR] {error_msg}") except requests.exceptions.ConnectionError: error_msg = "Connection error. Please check your internet connection." return safe_format_with_indicators(f"[ERROR] {error_msg}") except requests.exceptions.RequestException as e: error_msg = f"Request failed: {str(e)}" return safe_format_with_indicators(f"[ERROR] {error_msg}") except Exception as e: error_msg = f"Unexpected error in document analysis: {str(e)}" return safe_format_with_indicators(f"[ERROR] {error_msg}") def test_landing_ai_connection(): """ Test function to verify Landing AI API connection. This can be called to verify the setup before processing documents. Returns: bool: True if API key is available and valid format, False otherwise """ api_key = get_landing_ai_api_key() if not api_key: print("❌ Landing AI API key not found in environment variables") print("Available environment variables:") for key in os.environ.keys(): if 'LANDING' in key.upper() or 'API' in key.upper(): print(f" - {key}: {'SET' if os.environ[key] else 'EMPTY'}") return False # Basic format validation (adjust based on Landing AI key format) if len(api_key) < 10: print("❌ Landing AI API key appears to be too short") return False print("✅ Landing AI API key found and appears valid") return True # Test the connection when module is imported (optional) if __name__ == "__main__": test_landing_ai_connection()