Spaces:
Running
Running
| #!/usr/bin/env python3 | |
| """ | |
| Web Scraper MCP Server | |
| A Model Context Protocol server that provides web scraping tools. | |
| Exposes functions to scrape websites, convert content to markdown, and generate sitemaps. | |
| """ | |
| import gradio as gr | |
| import requests | |
| from bs4 import BeautifulSoup | |
| from markdownify import markdownify as md | |
| from urllib.parse import urljoin, urlparse | |
| from typing import Tuple, List, Dict | |
| import re | |
| import tempfile | |
| import zipfile | |
| import os | |
| def scrape_website_content(url: str) -> Tuple[str, str]: | |
| """ | |
| Scrape a website and return its main content formatted as markdown and a downloadable file path. | |
| Args: | |
| url (str): The URL to scrape (can include or omit http/https protocol) | |
| Returns: | |
| Tuple[str, str]: The scraped content formatted as markdown, and a file path for download | |
| """ | |
| try: | |
| # Validate URL | |
| if not url.startswith(('http://', 'https://')): | |
| url = 'https://' + url | |
| # Create session with proper headers | |
| session = requests.Session() | |
| session.headers.update({ | |
| 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36' | |
| }) | |
| # Make request | |
| response = session.get(url, timeout=10) | |
| response.raise_for_status() | |
| # Parse HTML | |
| soup = BeautifulSoup(response.content, 'html.parser') | |
| # Remove unwanted elements | |
| for element in soup(['script', 'style', 'nav', 'footer', 'header', 'aside']): | |
| element.decompose() | |
| # Try to find main content area | |
| main_content = ( | |
| soup.find('main') or | |
| soup.find('article') or | |
| soup.find('div', class_=re.compile(r'content|main|post|article')) or | |
| soup.find('body') | |
| ) | |
| if main_content: | |
| # Convert to markdown | |
| markdown_text = md(str(main_content), heading_style="ATX") | |
| # Clean up the markdown | |
| # Remove excessive newlines | |
| markdown_text = re.sub(r'\n{3,}', '\n\n', markdown_text) | |
| # Remove empty links | |
| markdown_text = re.sub(r'\[\s*\]\([^)]*\)', '', markdown_text) | |
| # Clean up whitespace | |
| markdown_text = re.sub(r'[ \t]+', ' ', markdown_text) | |
| # Add title if available | |
| title = soup.find('title') | |
| if title: | |
| markdown_text = f"# {title.get_text().strip()}\n\n{markdown_text}" | |
| markdown_text = markdown_text.strip() | |
| # Write to temp file for download | |
| with tempfile.NamedTemporaryFile(delete=False, suffix=".md", mode="w", encoding="utf-8") as f: | |
| f.write(markdown_text) | |
| temp_path = f.name | |
| return markdown_text, temp_path | |
| return "No main content found on the webpage.", None | |
| except requests.exceptions.RequestException as e: | |
| return f"Error fetching URL: {str(e)}", None | |
| except Exception as e: | |
| return f"Error processing content: {str(e)}", None | |
| def generate_sitemap(url: str, max_links_per_domain: int = None) -> Tuple[str, str]: | |
| """ | |
| Generate a sitemap from all links found on a webpage and provide a downloadable file path. | |
| Args: | |
| url (str): The URL to analyze for links (can include or omit http/https protocol) | |
| max_links_per_domain (int, optional): Maximum number of links to display per domain. | |
| If None, shows all links. Defaults to None. | |
| Returns: | |
| Tuple[str, str]: A markdown-formatted sitemap of all links found on the page, and a file path for download | |
| """ | |
| try: | |
| # Validate URL | |
| if not url.startswith(('http://', 'https://')): | |
| url = 'https://' + url | |
| # Create session with proper headers | |
| session = requests.Session() | |
| session.headers.update({ | |
| 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36' | |
| }) | |
| # Make request | |
| response = session.get(url, timeout=10) | |
| response.raise_for_status() | |
| # Parse HTML | |
| soup = BeautifulSoup(response.content, 'html.parser') | |
| # Find all links | |
| links = soup.find_all('a', href=True) | |
| # Process links | |
| sitemap_data = [] | |
| seen_urls = set() | |
| for link in links: | |
| href = link.get('href') | |
| text = link.get_text().strip() | |
| if not href: | |
| continue | |
| # Convert relative URLs to absolute | |
| full_url = urljoin(url, href) | |
| # Filter out unwanted links | |
| if (full_url in seen_urls or | |
| href.startswith(('#', 'javascript:', 'mailto:', 'tel:')) or | |
| full_url == url): | |
| continue | |
| seen_urls.add(full_url) | |
| # Create link entry | |
| if not text: | |
| text = href | |
| sitemap_data.append({ | |
| 'text': text[:100] + '...' if len(text) > 100 else text, | |
| 'url': full_url | |
| }) | |
| # Generate sitemap markdown | |
| if not sitemap_data: | |
| return "No links found on this page.", None | |
| sitemap_md = "# Sitemap\n\n" | |
| sitemap_md += f"Found {len(sitemap_data)} links:\n\n" | |
| # Group by domain for better organization | |
| domain_groups = {} | |
| parsed_base = urlparse(url) | |
| for item in sitemap_data: | |
| parsed_url = urlparse(item['url']) | |
| if parsed_url.netloc == parsed_base.netloc: | |
| domain_key = "Internal Links" | |
| else: | |
| domain_key = f"External Links ({parsed_url.netloc})" | |
| if domain_key not in domain_groups: | |
| domain_groups[domain_key] = [] | |
| domain_groups[domain_key].append(item) | |
| # Format sitemap | |
| for domain, links in domain_groups.items(): | |
| sitemap_md += f"## {domain}\n\n" | |
| # Use the limit parameter or show all links if None | |
| if max_links_per_domain is None: | |
| links_to_show = links | |
| remaining_links = 0 | |
| else: | |
| links_to_show = links[:max_links_per_domain] | |
| remaining_links = max(0, len(links) - max_links_per_domain) | |
| for link in links_to_show: | |
| sitemap_md += f"- [{link['text']}]({link['url']})\n" | |
| if remaining_links > 0: | |
| sitemap_md += f"- ... and {remaining_links} more links\n" | |
| sitemap_md += "\n" | |
| # Write to temp file for download | |
| with tempfile.NamedTemporaryFile(delete=False, suffix=".md", mode="w", encoding="utf-8") as f: | |
| f.write(sitemap_md) | |
| temp_path = f.name | |
| return sitemap_md, temp_path | |
| except requests.exceptions.RequestException as e: | |
| return f"Error fetching URL: {str(e)}", None | |
| except Exception as e: | |
| return f"Error processing content: {str(e)}", None | |
| def extract_all_content_as_zip(url: str, max_links: int = None) -> Tuple[str, str]: | |
| """ | |
| Extract text content from all links found on a webpage and create a downloadable zip file. | |
| Args: | |
| url (str): The URL to analyze for links (can include or omit http/https protocol) | |
| max_links (int, optional): Maximum number of links to process. If None, processes all links. Defaults to None. | |
| Returns: | |
| Tuple[str, str]: Status message and zip file path for download | |
| """ | |
| try: | |
| # Validate URL | |
| if not url.startswith(('http://', 'https://')): | |
| url = 'https://' + url | |
| # Create session with proper headers | |
| session = requests.Session() | |
| session.headers.update({ | |
| 'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36' | |
| }) | |
| # First get the sitemap to find all links | |
| response = session.get(url, timeout=10) | |
| response.raise_for_status() | |
| soup = BeautifulSoup(response.content, 'html.parser') | |
| links = soup.find_all('a', href=True) | |
| # Process links to get unique URLs | |
| unique_urls = set() | |
| parsed_base = urlparse(url) | |
| for link in links: | |
| href = link.get('href') | |
| if not href: | |
| continue | |
| full_url = urljoin(url, href) | |
| # Filter out unwanted links and focus on same domain for safety | |
| if (href.startswith(('#', 'javascript:', 'mailto:', 'tel:')) or | |
| full_url == url): | |
| continue | |
| # Only include internal links to avoid scraping too many external sites | |
| parsed_url = urlparse(full_url) | |
| if parsed_url.netloc == parsed_base.netloc: | |
| unique_urls.add(full_url) | |
| if not unique_urls: | |
| return "No internal links found to extract content from.", None | |
| # Use all URLs or limit if specified | |
| urls_to_process = list(unique_urls) | |
| total_links_found = len(urls_to_process) | |
| # Apply limit if specified | |
| if max_links is not None: | |
| urls_to_process = urls_to_process[:max_links] | |
| limited_message = f" (limited to {max_links} out of {total_links_found})" | |
| else: | |
| limited_message = "" | |
| # Create temporary zip file | |
| with tempfile.NamedTemporaryFile(delete=False, suffix=".zip") as temp_zip: | |
| zip_path = temp_zip.name | |
| successful_extractions = 0 | |
| failed_extractions = 0 | |
| with zipfile.ZipFile(zip_path, 'w', zipfile.ZIP_DEFLATED) as zip_file: | |
| for i, link_url in enumerate(urls_to_process, 1): | |
| try: | |
| # Get content from each link | |
| link_response = session.get(link_url, timeout=10) | |
| link_response.raise_for_status() | |
| # Parse and extract content | |
| link_soup = BeautifulSoup(link_response.content, 'html.parser') | |
| # Remove unwanted elements | |
| for element in link_soup(['script', 'style', 'nav', 'footer', 'header', 'aside']): | |
| element.decompose() | |
| # Find main content | |
| main_content = ( | |
| link_soup.find('main') or | |
| link_soup.find('article') or | |
| link_soup.find('div', class_=re.compile(r'content|main|post|article')) or | |
| link_soup.find('body') | |
| ) | |
| if main_content: | |
| # Convert to markdown | |
| markdown_text = md(str(main_content), heading_style="ATX") | |
| # Clean up the markdown | |
| markdown_text = re.sub(r'\n{3,}', '\n\n', markdown_text) | |
| markdown_text = re.sub(r'\[\s*\]\([^)]*\)', '', markdown_text) | |
| markdown_text = re.sub(r'[ \t]+', ' ', markdown_text) | |
| # Add title if available | |
| title = link_soup.find('title') | |
| if title: | |
| markdown_text = f"# {title.get_text().strip()}\n\n{markdown_text}" | |
| markdown_text = markdown_text.strip() | |
| # Create safe filename | |
| parsed_link = urlparse(link_url) | |
| safe_filename = re.sub(r'[^\w\-_.]', '_', parsed_link.path or 'index') | |
| if not safe_filename.endswith('.md'): | |
| safe_filename += '.md' | |
| # Ensure unique filename | |
| if safe_filename == '.md' or safe_filename == 'index.md': | |
| safe_filename = f"page_{i}.md" | |
| # Add source URL as header | |
| final_content = f"<!-- Source: {link_url} -->\n\n{markdown_text}" | |
| # Add to zip | |
| zip_file.writestr(safe_filename, final_content) | |
| successful_extractions += 1 | |
| else: failed_extractions += 1 | |
| except Exception as e: | |
| failed_extractions += 1 | |
| continue | |
| status_message = f"Successfully extracted content from {successful_extractions} pages{limited_message}" | |
| if failed_extractions > 0: | |
| status_message += f", failed to extract from {failed_extractions} pages" | |
| status_message += f". Created zip file with {successful_extractions} markdown files." | |
| return status_message, zip_path | |
| except requests.exceptions.RequestException as e: | |
| return f"Error fetching URL: {str(e)}", None | |
| except Exception as e: | |
| return f"Error processing content: {str(e)}", None | |
| def generate_sitemap_for_ui(url: str) -> Tuple[str, str]: | |
| """ | |
| Wrapper function for the Gradio UI that shows all links without limitation. | |
| Args: | |
| url (str): The URL to analyze for links | |
| Returns: | |
| Tuple[str, str]: A markdown-formatted sitemap of all links found on the page, and a file path for download | |
| """ | |
| return generate_sitemap(url, max_links_per_domain=None) | |
| def generate_sitemap_with_limit(url: str, max_links_per_domain: int) -> Tuple[str, str]: | |
| """ | |
| Wrapper function for Gradio UI that allows configurable link limits per domain. | |
| Args: | |
| url (str): The URL to analyze for links | |
| max_links_per_domain (int): Maximum number of links to display per domain (0 = show all) | |
| Returns: | |
| Tuple[str, str]: A markdown-formatted sitemap of all links found on the page, and a file path for download | |
| """ | |
| limit = None if max_links_per_domain == 0 else max_links_per_domain | |
| return generate_sitemap(url, max_links_per_domain=limit) | |
| def extract_all_content_for_ui(url: str) -> Tuple[str, str]: | |
| """ | |
| Wrapper function for the Gradio UI that extracts content from all internal links without limitation. | |
| Args: | |
| url (str): The URL to analyze for links | |
| Returns: | |
| Tuple[str, str]: Status message and zip file path for download | |
| """ | |
| return extract_all_content_as_zip(url, max_links=None) | |
| def extract_limited_content_as_zip(url: str, max_links: int) -> Tuple[str, str]: | |
| """ | |
| Wrapper function for Gradio UI that allows configurable link limits for bulk extraction. | |
| Args: | |
| url (str): The URL to analyze for links | |
| max_links (int): Maximum number of links to process (0 = process all) | |
| Returns: | |
| Tuple[str, str]: Status message and zip file path for download | |
| """ | |
| limit = None if max_links == 0 else max_links | |
| return extract_all_content_as_zip(url, max_links=limit) | |
| # Create Gradio interfaces for each function | |
| def create_mcp_interface(): | |
| """Create Gradio interface that exposes web scraping tools as MCP functions.""" | |
| # Create individual interfaces for each tool | |
| scrape_interface = gr.Interface( | |
| fn=scrape_website_content, | |
| inputs=gr.Textbox( | |
| label="Website URL", | |
| placeholder="https://example.com or example.com" | |
| ), | |
| outputs=[ | |
| gr.Textbox( | |
| label="Scraped Content", | |
| lines=20, | |
| max_lines=50, | |
| show_copy_button=True, | |
| container=True | |
| ), | |
| gr.File(label="Download Markdown") | |
| ], | |
| title="Website Content Scraper", | |
| description="Extract and format website content as markdown", | |
| api_name="scrape_content" ) | |
| sitemap_interface = gr.Interface( | |
| fn=generate_sitemap_for_ui, | |
| inputs=gr.Textbox( | |
| label="Website URL", | |
| placeholder="https://example.com or example.com" | |
| ), | |
| outputs=[ | |
| gr.Textbox( | |
| label="Sitemap", | |
| lines=20, | |
| max_lines=50, | |
| show_copy_button=True, | |
| container=True | |
| ), | |
| gr.File(label="Download Sitemap") | |
| ], | |
| title="Website Sitemap Generator", | |
| description="Generate a sitemap of all links found on a webpage", | |
| api_name="generate_sitemap" | |
| ) | |
| bulk_extract_interface = gr.Interface( | |
| fn=extract_all_content_for_ui, | |
| inputs=gr.Textbox( | |
| label="Website URL", | |
| placeholder="https://example.com or example.com" | |
| ), | |
| outputs=[ | |
| gr.Textbox( | |
| label="Extraction Status", | |
| lines=10, | |
| max_lines=20, | |
| show_copy_button=True, | |
| container=True | |
| ), | |
| gr.File(label="Download ZIP Archive") | |
| ], | |
| title="Bulk Content Extractor", | |
| description="Extract text content from all internal links and download as ZIP", | |
| api_name="extract_all_content" ) | |
| # Enhanced sitemap interface with configurable limits | |
| sitemap_limited_interface = gr.Interface( | |
| fn=generate_sitemap_with_limit, | |
| inputs=[ | |
| gr.Textbox( | |
| label="Website URL", | |
| placeholder="https://example.com or example.com" | |
| ), | |
| gr.Number( | |
| label="Max Links Per Domain", | |
| value=0, | |
| info="Enter 0 to show all links, or a positive number to limit display per domain", | |
| minimum=0, | |
| maximum=1000 | |
| ) | |
| ], | |
| outputs=[ | |
| gr.Textbox( | |
| label="Sitemap", | |
| lines=20, | |
| max_lines=50, | |
| show_copy_button=True, | |
| container=True | |
| ), | |
| gr.File(label="Download Sitemap") | |
| ], | |
| title="Configurable Sitemap Generator", | |
| description="Generate a sitemap with optional display limits (0 = show all links)", | |
| api_name="generate_sitemap_limited" | |
| ) | |
| # Enhanced bulk extract interface with configurable limits | |
| bulk_limited_interface = gr.Interface( | |
| fn=extract_limited_content_as_zip, | |
| inputs=[ | |
| gr.Textbox( | |
| label="Website URL", | |
| placeholder="https://example.com or example.com" | |
| ), | |
| gr.Number( | |
| label="Max Pages to Extract", | |
| value=0, | |
| info="Enter 0 to process all pages, or a positive number to limit extraction", | |
| minimum=0, | |
| maximum=1000 | |
| ) | |
| ], | |
| outputs=[ | |
| gr.Textbox( | |
| label="Extraction Status", | |
| lines=10, | |
| max_lines=20, | |
| show_copy_button=True, | |
| container=True | |
| ), | |
| gr.File(label="Download ZIP Archive") | |
| ], | |
| title="Limited Bulk Content Extractor", | |
| description="Extract text content from internal links with optional processing limits (0 = extract all)", | |
| api_name="extract_limited_content" | |
| ) | |
| # Combine into tabbed interface | |
| demo = gr.TabbedInterface( | |
| [scrape_interface, sitemap_interface, sitemap_limited_interface, bulk_extract_interface, bulk_limited_interface], | |
| ["Content Scraper", "All Links Sitemap", "Limited Sitemap", "Bulk Extractor", "Limited Bulk Extractor"], | |
| title="🕷️ Web Scraper MCP Server" | |
| ) | |
| return demo | |
| if __name__ == "__main__": | |
| # Create and launch the MCP server | |
| app = create_mcp_interface() | |
| app.launch( | |
| mcp_server=True | |
| ) | |