#!/usr/bin/env python3 """ Hugging Face Gradio App for RDF Validation with MCP Server and Anthropic AI This app serves both as a web interface and can expose MCP server functionality. Deploy this on Hugging Face Spaces with your Anthropic API key. """ import gradio as gr import os import json import sys import asyncio import logging import requests import re from typing import Any, Dict, List, Optional import threading import time # Add current directory to path sys.path.append(os.path.dirname(os.path.abspath(__file__))) # Import our validation logic try: from validator import validate_rdf VALIDATOR_AVAILABLE = True except ImportError: VALIDATOR_AVAILABLE = False print("āš ļø Warning: validator.py not found. Some features may be limited.") # Optional: Check if OpenAI and requests are available try: from openai import OpenAI OPENAI_AVAILABLE = True except ImportError: OPENAI_AVAILABLE = False print("šŸ’” Install 'openai' package for AI-powered corrections: pip install openai") try: import requests HF_INFERENCE_AVAILABLE = True except ImportError: HF_INFERENCE_AVAILABLE = False print("šŸ’” Install 'requests' package for AI-powered corrections: pip install requests") # Set up logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Configuration - Your specific Hugging Face Inference Endpoint (hardcoded) HF_API_KEY = os.getenv('HF_API_KEY', '') # Hugging Face API key from Secret HF_ENDPOINT_URL = "https://evxgv66ksxjlfrts.us-east-1.aws.endpoints.huggingface.cloud/v1/" HF_MODEL = "lmstudio-community/Llama-3.3-70B-Instruct-GGUF" # Correct model name for your endpoint # AI Correction Configuration MAX_CORRECTION_ATTEMPTS = 2 # Reduced from 3 to speed up processing ENABLE_VALIDATION_LOOP = False # Disable validation loop to prevent hanging # OpenAI client configuration for the endpoint def get_openai_client(): """Get configured OpenAI client for HF Inference Endpoint""" if not HF_API_KEY: print("āŒ No HF_API_KEY available for OpenAI client") return None print(f"šŸ”— Creating OpenAI client with:") print(f" base_url: {HF_ENDPOINT_URL}") print(f" api_key: {'***' + HF_API_KEY[-4:] if len(HF_API_KEY) > 4 else 'HIDDEN'}") return OpenAI( base_url=HF_ENDPOINT_URL, api_key=HF_API_KEY, timeout=120.0 # Increase timeout for cold starts ) # Sample RDF data for examples SAMPLE_VALID_RDF = ''' Sample Monograph Title Sample Author Author Example Library ''' SAMPLE_INVALID_RDF = ''' Incomplete Title ''' # MCP Server Tools (can be used independently) def validate_rdf_tool(rdf_content: str, template: str = "monograph") -> dict: """ Validate RDF/XML content against SHACL templates. This tool validates RDF/XML data against predefined SHACL shapes to ensure compliance with metadata standards like BIBFRAME. Returns detailed validation results with conformance status and specific violation information. Args: rdf_content (str): The RDF/XML content to validate template (str): Validation template to use ('monograph' or 'custom') Returns: dict: Validation results with conformance status and detailed feedback """ if not rdf_content: return {"error": "No RDF/XML content provided", "conforms": False} if not VALIDATOR_AVAILABLE: return { "error": "Validator not available - ensure validator.py is present", "conforms": False } try: conforms, results_text = validate_rdf(rdf_content.encode('utf-8'), template) return { "conforms": conforms, "results": results_text, "template": template, "status": "āœ… Valid RDF" if conforms else "āŒ Invalid RDF" } except Exception as e: logger.error(f"Validation error: {str(e)}") return { "error": f"Validation failed: {str(e)}", "conforms": False } def filter_validation_results_by_class(validation_results: str, rdf_content: str) -> dict: """ Filter validation results by RDF class (Work, Instance, etc.) Args: validation_results (str): Full validation results rdf_content (str): Original RDF content Returns: dict: Validation results organized by class """ import re # Parse validation results to extract class information class_results = { 'Work': [], 'Instance': [], 'Title': [], 'Contribution': [], 'Other': [] } lines = validation_results.split('\n') current_section = [] current_class = 'Other' for line in lines: # Detect which class this error relates to if 'bf:Work' in line or '/work/' in line: current_class = 'Work' elif 'bf:Instance' in line or '/instance/' in line: current_class = 'Instance' elif 'bf:Title' in line: current_class = 'Title' elif 'bf:Contribution' in line: current_class = 'Contribution' # Collect lines for current violation if 'Constraint Violation' in line: if current_section: class_results[current_class].extend(current_section) current_section = [line] elif line.strip(): current_section.append(line) # Add last section if current_section: class_results[current_class].extend(current_section) # Remove empty classes return {k: '\n'.join(v) for k, v in class_results.items() if v} def get_ai_suggestions(validation_results: str, rdf_content: str, include_warnings: bool = False) -> str: """ Generate AI-powered fix suggestions for invalid RDF/XML. This tool analyzes validation results and provides actionable suggestions for fixing RDF/XML validation errors using AI or rule-based analysis. Args: validation_results (str): The validation error messages rdf_content (str): The original RDF/XML content that failed validation include_warnings (bool): Whether to include warnings in suggestions Returns: str: Detailed suggestions for fixing the RDF validation issues """ if not OPENAI_AVAILABLE: return generate_manual_suggestions(validation_results) # Get API key dynamically at runtime current_api_key = os.getenv('HF_API_KEY', '') if not current_api_key: return f""" šŸ”‘ **AI suggestions disabled**: Please set your Hugging Face API key as a Secret in your Space settings. {generate_manual_suggestions(validation_results)} """ try: # Use OpenAI client with your Hugging Face Inference Endpoint client = get_openai_client() if not client: return f""" šŸ”‘ **AI suggestions disabled**: HF_API_KEY not configured. {generate_manual_suggestions(validation_results)} """ severity_instruction = "Focus only on violations (errors) and ignore any warnings." if not include_warnings else "Address both violations and warnings." # Filter validation results by class to reduce token usage class_results = filter_validation_results_by_class(validation_results, rdf_content) # Determine primary class with most errors primary_class = max(class_results.keys(), key=lambda k: len(class_results[k])) focused_results = class_results[primary_class] # Extract only relevant RDF section for the primary class relevant_rdf = extract_relevant_rdf_section(rdf_content, primary_class) prompt = f"""You are an expert in RDF/XML and SHACL validation. Analyze the validation errors for the {primary_class} class and provide CONCISE, ACTIONABLE fixes. {severity_instruction} Validation Errors for {primary_class}: {focused_results[:1500]} Relevant RDF Section: {relevant_rdf[:800]} Instructions: 1. ONE sentence: What's wrong with this {primary_class}? 2. List errors (max 3 words each) 3. Show exact XML fixes Format: **Issue:** [One sentence about the {primary_class} problem] **Errors:** ā€¢ Error 1 ā€¢ Error 2 **Fix:** ```xml [Complete corrected {primary_class} section] ``` Be ultra-concise. Show the fix, not explanations.""" # Make API call using OpenAI client print(f"šŸ”„ Making focused API call for {primary_class} class") print(f"šŸ”„ Sending {len(focused_results)} chars instead of {len(validation_results)} chars") chat_completion = client.chat.completions.create( model=HF_MODEL, messages=[ { "role": "user", "content": prompt } ], max_tokens=800, # Reduced since we're focused on one class temperature=0.5, # Lower temperature for more focused responses top_p=0.9 ) print("āœ… API call successful") generated_text = chat_completion.choices[0].message.content # Add note about other classes if present other_classes = [k for k in class_results.keys() if k != primary_class] class_note = f"\n\nšŸ“Œ **Note:** Focused on {primary_class} errors. " + \ (f"Also found issues in: {', '.join(other_classes)}" if other_classes else "") return f"šŸ¤– **AI-Powered Suggestions ({('Violations + Warnings' if include_warnings else 'Violations Only')}):**\n\n{generated_text}{class_note}" except Exception as e: logger.error(f"OpenAI/HF Inference Endpoint error: {str(e)}") return f""" āŒ **AI suggestions error**: {str(e)} {generate_manual_suggestions(validation_results)} """ def extract_relevant_rdf_section(rdf_content: str, class_name: str) -> str: """ Extract only the relevant RDF section for a specific class Args: rdf_content (str): Full RDF content class_name (str): Class name to extract (Work, Instance, etc.) Returns: str: Relevant RDF section """ import re # Map class names to RDF patterns patterns = { 'Work': r'', 'Instance': r'', 'Title': r'', 'Contribution': r'' } pattern = patterns.get(class_name) if not pattern: return rdf_content[:1000] # Fallback to first 1000 chars # Extract matching section match = re.search(pattern, rdf_content, re.DOTALL) if match: section = match.group(0) # Also include namespace declarations namespaces = re.findall(r'xmlns:\w+="[^"]*"', rdf_content[:500]) if namespaces: return f"\n{section}" return section return rdf_content[:1000] # Fallback def get_ai_correction(validation_results: str, rdf_content: str, template: str = 'monograph', max_attempts: int = None, include_warnings: bool = False) -> str: """ Generate AI-powered corrected RDF/XML based on validation errors. This tool takes invalid RDF/XML and validation results, then generates a corrected version that addresses all identified validation issues. The generated correction is validated before being returned to the user. Args: validation_results (str): The validation error messages rdf_content (str): The original invalid RDF/XML content template (str): The validation template to use max_attempts (int): Maximum number of attempts to generate valid RDF (uses MAX_CORRECTION_ATTEMPTS if None) include_warnings (bool): Whether to fix warnings in addition to violations Returns: str: Corrected RDF/XML that should pass validation """ # Use configuration default if not specified if max_attempts is None: max_attempts = MAX_CORRECTION_ATTEMPTS # Check if validation loop is enabled if not ENABLE_VALIDATION_LOOP: max_attempts = 1 # Fall back to single attempt if validation loop disabled if not OPENAI_AVAILABLE: return generate_manual_correction_hints(validation_results, rdf_content) # Get API key dynamically at runtime current_api_key = os.getenv('HF_API_KEY', '') if not current_api_key: return f""" {generate_manual_correction_hints(validation_results, rdf_content)}""" try: client = get_openai_client() if not client: return f""" {generate_manual_correction_hints(validation_results, rdf_content)}""" # Add timeout protection import time start_time = time.time() timeout = 60 # 60 second timeout severity_instruction = "Fix only the violations (errors) and ignore any warnings." if not include_warnings else "Fix both violations and warnings." # Filter validation results by class class_results = filter_validation_results_by_class(validation_results, rdf_content) # Process each class separately to avoid overwhelming the LLM corrected_sections = {} for class_name, class_errors in class_results.items(): if not class_errors: continue # Check timeout if time.time() - start_time > timeout - 10: print(f"ā° Approaching timeout, skipping {class_name}") break print(f"šŸ”„ Correcting {class_name} section") # Extract relevant section relevant_section = extract_relevant_rdf_section(rdf_content, class_name) prompt = f"""Fix this {class_name} RDF section based on these specific errors. {severity_instruction} Errors for {class_name}: {class_errors[:800]} Current {class_name} RDF: {relevant_section[:800]} Return ONLY the corrected {class_name} XML section. No explanations.""" try: chat_completion = client.chat.completions.create( model=HF_MODEL, messages=[ { "role": "user", "content": prompt } ], max_tokens=1000, temperature=0.3, timeout=20 # Shorter timeout per section ) corrected_section = chat_completion.choices[0].message.content.strip() corrected_sections[class_name] = extract_rdf_from_response(corrected_section) except Exception as e: print(f"āŒ Error correcting {class_name}: {str(e)}") continue # Merge corrections back into original RDF if corrected_sections: corrected_rdf = merge_corrected_sections(rdf_content, corrected_sections) return f""" {corrected_rdf}""" else: return f""" {generate_manual_correction_hints(validation_results, rdf_content)}""" except Exception as e: logger.error(f"LLM API error: {str(e)}") return f""" {generate_manual_correction_hints(validation_results, rdf_content)}""" def merge_corrected_sections(original_rdf: str, corrected_sections: dict) -> str: """ Merge corrected class sections back into the original RDF Args: original_rdf (str): Original RDF content corrected_sections (dict): Corrected sections by class Returns: str: Merged RDF with corrections """ import re result = original_rdf # Replace each corrected section for class_name, corrected_section in corrected_sections.items(): patterns = { 'Work': r'', 'Instance': r'', 'Title': r'', 'Contribution': r'' } pattern = patterns.get(class_name) if pattern: result = re.sub(pattern, corrected_section, result, count=1, flags=re.DOTALL) return result # Sample RDF data for examples # MCP Server Tools (can be used independently) # Note: This section exists earlier in the file, we're removing the duplicates """ Generate AI-powered fix suggestions for invalid RDF/XML. This tool analyzes validation results and provides actionable suggestions for fixing RDF/XML validation errors using AI or rule-based analysis. Args: validation_results (str): The validation error messages rdf_content (str): The original RDF/XML content that failed validation include_warnings (bool): Whether to include warnings in suggestions Returns: str: Detailed suggestions for fixing the RDF validation issues """ if not OPENAI_AVAILABLE: return generate_manual_suggestions(validation_results) # Get API key dynamically at runtime current_api_key = os.getenv('HF_API_KEY', '') if not current_api_key: return f""" šŸ”‘ **AI suggestions disabled**: Please set your Hugging Face API key as a Secret in your Space settings. {generate_manual_suggestions(validation_results)} """ try: # Use OpenAI client with your Hugging Face Inference Endpoint client = get_openai_client() if not client: return f""" šŸ”‘ **AI suggestions disabled**: HF_API_KEY not configured. {generate_manual_suggestions(validation_results)} """ severity_instruction = "Focus only on violations (errors) and ignore any warnings." if not include_warnings else "Address both violations and warnings." prompt = f"""You are an expert in RDF/XML and SHACL validation. Analyze the validation errors and provide CONCISE, ACTIONABLE fix suggestions. {severity_instruction} Validation Results: {validation_results} Original RDF (first 1000 chars): {rdf_content[:1000]}... Instructions: 1. Start with a ONE-SENTENCE summary of the main issue 2. List the specific errors in bullet points (max 5 words per error) 3. Provide the exact fix for each error with code snippets 4. Keep explanations minimal - focus on solutions Format: **Main Issue:** [One sentence] **Errors Found:** ā€¢ Error 1 name ā€¢ Error 2 name **Fixes:** 1. **Error 1**: ```xml [exact code to add/fix] ``` 2. **Error 2**: ```xml [exact code to add/fix] ``` Be direct and solution-focused. No lengthy explanations.""" # Make API call using OpenAI client print(f"šŸ”„ Making API call to: {HF_ENDPOINT_URL}") print(f"šŸ”„ Using model: {HF_MODEL}") print(f"šŸ”„ Include warnings: {include_warnings}") chat_completion = client.chat.completions.create( model=HF_MODEL, messages=[ { "role": "user", "content": prompt } ], max_tokens=1500, temperature=0.7, top_p=0.9 ) print("āœ… API call successful") generated_text = chat_completion.choices[0].message.content return f"šŸ¤– **AI-Powered Suggestions ({('Violations + Warnings' if include_warnings else 'Violations Only')}):**\n\n{generated_text}" except Exception as e: logger.error(f"OpenAI/HF Inference Endpoint error: {str(e)}") return f""" āŒ **AI suggestions error**: {str(e)} {generate_manual_suggestions(validation_results)} """ def extract_rdf_from_response(response: str) -> str: """ Extract RDF/XML content from AI response, handling code blocks. Args: response (str): AI response that may contain RDF wrapped in code blocks Returns: str: Extracted RDF/XML content """ response = response.strip() # Handle ```xml code blocks if "```xml" in response: try: return response.split("```xml")[1].split("```")[0].strip() except IndexError: pass # Handle generic ``` code blocks if "```" in response and response.count("```") >= 2: try: return response.split("```")[1].split("```")[0].strip() except IndexError: pass # If no code blocks found, return the response as-is return response def get_ai_correction(validation_results: str, rdf_content: str, template: str = 'monograph', max_attempts: int = None, include_warnings: bool = False) -> str: """ Generate AI-powered corrected RDF/XML based on validation errors. This tool takes invalid RDF/XML and validation results, then generates a corrected version that addresses all identified validation issues. The generated correction is validated before being returned to the user. Args: validation_results (str): The validation error messages rdf_content (str): The original invalid RDF/XML content template (str): The validation template to use max_attempts (int): Maximum number of attempts to generate valid RDF (uses MAX_CORRECTION_ATTEMPTS if None) include_warnings (bool): Whether to fix warnings in addition to violations Returns: str: Corrected RDF/XML that should pass validation """ # Use configuration default if not specified if max_attempts is None: max_attempts = MAX_CORRECTION_ATTEMPTS # Check if validation loop is enabled if not ENABLE_VALIDATION_LOOP: max_attempts = 1 # Fall back to single attempt if validation loop disabled if not OPENAI_AVAILABLE: return generate_manual_correction_hints(validation_results, rdf_content) # Get API key dynamically at runtime current_api_key = os.getenv('HF_API_KEY', '') if not current_api_key: return f""" {generate_manual_correction_hints(validation_results, rdf_content)}""" try: client = get_openai_client() if not client: return f""" {generate_manual_correction_hints(validation_results, rdf_content)}""" # Add timeout protection import time start_time = time.time() timeout = 60 # 60 second timeout severity_instruction = "Fix only the violations (errors) and ignore any warnings." if not include_warnings else "Fix both violations and warnings." # Try multiple attempts to generate valid RDF for attempt in range(max_attempts): # Check timeout if time.time() - start_time > timeout: print(f"ā° Timeout reached after {timeout} seconds") break print(f"šŸ”„ Correction attempt {attempt + 1}/{max_attempts}") prompt = f"""You are an expert in RDF/XML. Fix the following RDF/XML based on the validation errors provided. {severity_instruction} Validation Errors: {validation_results} Original RDF/XML: {rdf_content} {f"Previous attempt {attempt} still had validation errors. Please fix ALL issues this time." if attempt > 0 else ""} Please provide the corrected RDF/XML that addresses all validation issues. - Return only the corrected XML without additional explanation - Maintain the original structure as much as possible while fixing errors - Ensure all namespace declarations are present - Add any missing required properties - Fix any syntax or structural issues""" try: chat_completion = client.chat.completions.create( model=HF_MODEL, messages=[ { "role": "user", "content": prompt } ], max_tokens=2000, temperature=0.3, timeout=30 # 30 second timeout per API call ) corrected_rdf = chat_completion.choices[0].message.content.strip() # Extract RDF content if it's wrapped in code blocks corrected_rdf = extract_rdf_from_response(corrected_rdf) # Only validate if we have the validator and haven't hit timeout if VALIDATOR_AVAILABLE and (time.time() - start_time < timeout - 10): try: # Quick validation check conforms, new_results = validate_rdf(corrected_rdf.encode('utf-8'), template) if conforms: print(f"āœ… Correction validated successfully on attempt {attempt + 1}") return f""" {corrected_rdf}""" else: print(f"āŒ Correction attempt {attempt + 1} still has validation errors") # Update validation_results for next attempt validation_results = new_results except Exception as e: print(f"āš ļø Error validating correction attempt {attempt + 1}: {str(e)}") # If validation fails, return the correction anyway return f""" {corrected_rdf}""" else: # If validator not available or timeout approaching, return the correction print("āš ļø Returning correction without validation") return f""" {corrected_rdf}""" except Exception as api_error: print(f"āŒ API error on attempt {attempt + 1}: {str(api_error)}") if attempt == max_attempts - 1: # Last attempt raise api_error continue # All attempts failed or timed out return f""" {generate_manual_correction_hints(validation_results, rdf_content)}""" except Exception as e: logger.error(f"LLM API error: {str(e)}") return f""" {generate_manual_correction_hints(validation_results, rdf_content)}""" def generate_manual_suggestions(validation_results: str) -> str: """Generate rule-based suggestions when AI is not available""" suggestions = [] if "Constraint Violation" in validation_results: suggestions.append("ā€¢ Fix SHACL constraint violations by ensuring required properties are present") if "Missing property" in validation_results or "missing" in validation_results.lower(): suggestions.append("ā€¢ Add missing required properties (check template requirements)") if "datatype" in validation_results.lower(): suggestions.append("ā€¢ Correct data type mismatches (ensure proper literal types)") if "namespace" in validation_results.lower() or "prefix" in validation_results.lower(): suggestions.append("ā€¢ Add missing namespace declarations at the top of your RDF") if "XML" in validation_results or "syntax" in validation_results.lower(): suggestions.append("ā€¢ Fix XML syntax errors (check for unclosed tags, invalid characters)") if not suggestions: suggestions.append("ā€¢ Review detailed validation results for specific issues") suggestions.append("ā€¢ Ensure your RDF follows the selected template requirements") suggestions_text = "\n".join(suggestions) return f""" šŸ“‹ **Manual Analysis:** {suggestions_text} šŸ’” **General Tips:** ā€¢ Check namespace declarations at the top of your RDF ā€¢ Ensure all required properties are present ā€¢ Verify data types match expected formats ā€¢ Make sure XML structure is well-formed šŸ”§ **Common Fixes:** ā€¢ Add missing namespace prefixes ā€¢ Include required properties like rdf:type ā€¢ Fix malformed URIs or literals ā€¢ Ensure proper XML syntax """ def generate_manual_correction_hints(validation_results: str, rdf_content: str) -> str: """Generate manual correction hints when AI is not available""" return f""" {rdf_content} """ def validate_rdf_interface(rdf_content: str, template: str, use_ai: bool = True, include_warnings: bool = False): """ Main validation function for Gradio interface and MCP server. This function provides comprehensive RDF/XML validation with AI-powered suggestions and corrections. It serves as the primary interface for both the Gradio web UI and MCP client tools. Args: rdf_content (str): The RDF/XML content to validate template (str): Validation template to use ('monograph' or 'custom') use_ai (bool): Whether to enable AI-powered suggestions and corrections include_warnings (bool): Whether to include warnings in AI corrections (violations only by default) Returns: tuple: (status, results_text, suggestions, corrected_rdf) containing: - status: Validation status message - results_text: Detailed validation results - suggestions: AI or manual fix suggestions - corrected_rdf: AI-generated corrections or success message """ if not rdf_content.strip(): return "āŒ Error", "No RDF/XML data provided", "", "" # Validate RDF result = validate_rdf_tool(rdf_content, template) if "error" in result: return f"āŒ Error: {result['error']}", "", "", "" status = result["status"] results_text = result["results"] # Filter results if warnings should be excluded filtered_results = results_text if not include_warnings and "Warning" in results_text: # Split results into lines and filter out warnings lines = results_text.split('\n') filtered_lines = [] skip_until_next_section = False for line in lines: if "Warning" in line and ("Constraint Violation" in line or "sh:Warning" in line): skip_until_next_section = True elif "Constraint Violation" in line and "Warning" not in line: skip_until_next_section = False filtered_lines.append(line) elif not skip_until_next_section: filtered_lines.append(line) filtered_results = '\n'.join(filtered_lines) if result["conforms"]: suggestions = "āœ… No issues found! Your RDF/XML is valid according to the selected template." corrected_rdf = "āœ… Your RDF/XML is already valid - no corrections needed!" else: if use_ai: # Pass filtered results to AI functions suggestions = get_ai_suggestions(filtered_results, rdf_content, include_warnings) corrected_rdf = get_ai_correction(filtered_results, rdf_content, template, include_warnings=include_warnings) else: suggestions = generate_manual_suggestions(filtered_results) corrected_rdf = generate_manual_correction_hints(filtered_results, rdf_content) return status, results_text, suggestions, corrected_rdf def get_rdf_examples(example_type: str = "valid") -> str: """ Retrieve example RDF/XML snippets for testing and learning. This tool provides sample RDF/XML content that can be used to test the validation system or learn proper RDF structure. Examples include valid BibFrame Work records, invalid records for testing corrections, and BibFrame Instance records. Args: example_type (str): Type of example to retrieve. Options: - 'valid': A complete, valid BibFrame Work record - 'invalid': An incomplete BibFrame Work with validation errors - 'bibframe': A BibFrame Instance record example Returns: str: Complete RDF/XML example content ready for validation testing """ examples = { "valid": SAMPLE_VALID_RDF, "invalid": SAMPLE_INVALID_RDF, "bibframe": ''' Example Book Title 2024 New York ''' } return examples.get(example_type, examples["valid"]) # Create Gradio Interface def create_interface(): """Create the main Gradio interface""" # Check API key status dynamically current_api_key = os.getenv('HF_API_KEY', '') api_status = "šŸ”‘ AI features enabled" if (OPENAI_AVAILABLE and current_api_key) else "āš ļø AI features disabled (set HF_API_KEY)" with gr.Blocks( title="RDF Validation Server with AI", theme=gr.themes.Soft(), css=""" .status-box { font-weight: bold; padding: 10px; border-radius: 5px; } .header-text { text-align: center; padding: 20px; } """ ) as demo: # Header debug_info = f""" Debug Info: - OPENAI_AVAILABLE: {OPENAI_AVAILABLE} - HF_INFERENCE_AVAILABLE: {HF_INFERENCE_AVAILABLE} - HF_API_KEY set: {'Yes' if current_api_key else 'No'} - HF_API_KEY length: {len(current_api_key) if current_api_key else 0} - HF_ENDPOINT_URL: {HF_ENDPOINT_URL} - HF_MODEL: {HF_MODEL} """ gr.HTML(f"""

šŸ” RDF Validation Server with AI

Validate RDF/XML against SHACL schemas with AI-powered suggestions and corrections

Status: {api_status}

Debug Info
{debug_info}
""") # Main interface with gr.Row(): with gr.Column(scale=1): gr.Markdown("### šŸ“ Input") rdf_input = gr.Textbox( label="RDF/XML Content", placeholder="Paste your RDF/XML content here...", lines=15, show_copy_button=True ) with gr.Row(): template_dropdown = gr.Dropdown( label="Validation Template", choices=["monograph", "custom"], value="monograph", info="Select the SHACL template to validate against" ) use_ai_checkbox = gr.Checkbox( label="Use AI Features", value=True, info="Enable AI-powered suggestions and corrections" ) include_warnings_checkbox = gr.Checkbox( label="Include Warnings", value=False, info="Include warnings in AI corrections (violations only by default)" ) validate_btn = gr.Button("šŸ” Validate RDF", variant="primary", size="lg") # Examples and controls gr.Markdown("### šŸ“š Examples & Tools") with gr.Row(): example1_btn = gr.Button("āœ… Valid RDF Example", variant="secondary") example2_btn = gr.Button("āŒ Invalid RDF Example", variant="secondary") clear_btn = gr.Button("šŸ—‘ļø Clear All", variant="stop") # Results section with gr.Row(): with gr.Column(): gr.Markdown("### šŸ“Š Results") status_output = gr.Textbox( label="Validation Status", interactive=False, lines=1, elem_classes=["status-box"] ) results_output = gr.Textbox( label="Detailed Validation Results", interactive=False, lines=8, show_copy_button=True ) suggestions_output = gr.Textbox( label="šŸ’” Fix Suggestions", interactive=False, lines=8, show_copy_button=True ) # Corrected RDF section with gr.Row(): with gr.Column(): gr.Markdown("### šŸ› ļø AI-Generated Corrections") corrected_output = gr.Textbox( label="Corrected RDF/XML", interactive=False, lines=15, show_copy_button=True, placeholder="Corrected RDF will appear here after validation..." ) # Event handlers validate_btn.click( fn=validate_rdf_interface, inputs=[rdf_input, template_dropdown, use_ai_checkbox, include_warnings_checkbox], outputs=[status_output, results_output, suggestions_output, corrected_output] ) # Remove auto-validation to prevent processing loops # rdf_input.change( # fn=validate_rdf_interface, # inputs=[rdf_input, template_dropdown, use_ai_checkbox], # outputs=[status_output, results_output, suggestions_output, corrected_output] # ) # Example buttons example1_btn.click( lambda: get_rdf_examples("valid"), outputs=[rdf_input] ) example2_btn.click( lambda: get_rdf_examples("invalid"), outputs=[rdf_input] ) clear_btn.click( lambda: ("", "", "", "", ""), outputs=[rdf_input, status_output, results_output, suggestions_output, corrected_output] ) # Footer with instructions gr.Markdown(""" --- ### šŸš€ **Deployment Instructions for Hugging Face Spaces:** 1. **Create a new Space** on [Hugging Face](https://huggingface.co/spaces) 2. **Set up your Hugging Face Inference Endpoint** and get the endpoint URL 3. **Set your tokens** in Space settings (use Secrets for security): - Go to Settings ā†’ Repository secrets - Add: `HF_API_KEY` = `your_huggingface_api_key_here` - Endpoint is now hardcoded to your specific Inference Endpoint 4. **Upload these files** to your Space repository 5. **Install requirements**: The Space will auto-install from `requirements.txt` ### šŸ”§ **MCP Server Mode:** This app functions as both a web interface AND an MCP server for Claude Desktop and other MCP clients. **Available MCP Tools:** - `validate_rdf_tool`: Validate RDF/XML against SHACL shapes - `get_ai_suggestions`: Get AI-powered fix suggestions - `get_ai_correction`: Generate corrected RDF/XML - `get_rdf_examples`: Retrieve example RDF snippets - `validate_rdf_interface`: Complete validation with AI suggestions and corrections (primary tool) **MCP Configuration (Streamable HTTP):** Add this configuration to your MCP client (Claude Desktop, etc.): ```json { "mcpServers": { "rdf-validator": { "url": "https://jimfhahn-mcp4rdf.hf.space/gradio_api/mcp/" } } } ``` **Alternative SSE Configuration:** ```json { "mcpServers": { "rdf-validator": { "url": "https://jimfhahn-mcp4rdf.hf.space/gradio_api/mcp/sse" } } } ``` ### šŸ’” **Features:** - āœ… Real-time RDF/XML validation against SHACL schemas - šŸ¤– AI-powered error suggestions and corrections (with HF Inference Endpoint) - šŸ“š Built-in examples and templates - ļæ½ Manual validation on-demand (click to validate) - šŸ“‹ Copy results with one click **Note:** AI features require a valid Hugging Face API key (HF_API_KEY) set as a Secret. Manual suggestions are provided as fallback. """) return demo # Launch configuration if __name__ == "__main__": demo = create_interface() # Configuration for different environments port = int(os.getenv('PORT', 7860)) # Hugging Face uses PORT env variable demo.launch( server_name="0.0.0.0", # Important for external hosting server_port=port, # Use environment PORT or default to 7860 share=False, # Don't create gradio.live links in production show_error=True, # Show errors in the interface show_api=True, # Enable API endpoints allowed_paths=["."], # Allow serving files from current directory mcp_server=True # Enable MCP server functionality (Gradio 5.28+) )