vincent
2b43592151
-Clean and deduplicate URLs -Clean extracted text -Semantic selection to keep only text relevant to the query -Limit number of extracted characters per URL and also limit the total characters extracted to avoid returning too much text Arguments: urls: List of URLs to process. query: User question or information wanted Semantic ranking is performed against this query. Optional Arguments: max_chars_per_url: Maximum characters returned for each URL. Default set to 6000 max_total_chars: Maximum characters returned across all URLs. Default set to 30000 top_k_chunks: Maximum number of relevant chunks per URL. Default set to 6 min_relevance_score: Minimum semantic similarity score. Default set to 0.25 Returns: List of relevant web content results. Below an example of info return for one URL: ======================================== === SOURCE URL: https://www.marketsandmarkets.com/Market-Reports/3d-scanner-market-119952472.html === STATUS: success CONTENT: [Semantic Retrieval] Query: Extract information about companies, products, pricing, key features, target customers, market trends, growth, and competitive information related to 3D scanners. Chunks considered: 109 Chunks selected: 6 Relevance scores: [0.759, 0.757, 0.749, 0.71, 0.7, 0.696] Content characters: 3046 Truncated: False --- RELEVANT CONTENT --- Chunk 1 Chunk 2 Chunk 3 Chunk 4 Chunk 5 Chunk 6 ========================================
52 lines
1.8 KiB
Python
52 lines
1.8 KiB
Python
# mcp_server.py
|
|
import requests
|
|
import json
|
|
from bs4 import BeautifulSoup
|
|
from fastmcp import FastMCP
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
|
|
mcp = FastMCP("BatchWebContentExtraction")
|
|
|
|
def scrape_single_url(url: str) -> dict:
|
|
"""Helper function to scrape and clean a single URL."""
|
|
try:
|
|
headers = {"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"}
|
|
response = requests.get(url, headers=headers, timeout=10)
|
|
response.raise_for_status()
|
|
|
|
soup = BeautifulSoup(response.text, 'html.parser')
|
|
for script in soup(["script", "style", "nav", "footer", "header", "aside"]):
|
|
script.decompose()
|
|
|
|
raw_text = soup.get_text(separator=' ')
|
|
lines = (line.strip() for line in raw_text.splitlines())
|
|
chunks = (phrase.strip() for line in lines for phrase in line.split(" "))
|
|
clean_text = '\n'.join(chunk for chunk in chunks if chunk)
|
|
|
|
return {"url": url, "status": "success", "content": clean_text[:5000]}
|
|
|
|
except Exception as e:
|
|
return {"url": url, "status": "failed", "content": str(e)}
|
|
|
|
@mcp.tool()
|
|
def batch_web_content_extraction(urls: list[str]) -> str:
|
|
"""
|
|
Fetches text content from a list of multiple URLs simultaneously in parallel.
|
|
Returns a unified JSON string of results.
|
|
"""
|
|
if not urls:
|
|
return json.dumps({"error": "No URLs provided"})
|
|
|
|
results = []
|
|
with ThreadPoolExecutor(max_workers=min(len(urls), 5)) as executor:
|
|
futures_map = {executor.submit(scrape_single_url, url): url for url in urls}
|
|
for future in futures_map:
|
|
results.append(future.result())
|
|
|
|
return json.dumps(results)
|
|
|
|
if __name__ == "__main__":
|
|
# Use standard streamable-http protocol on port 8000
|
|
mcp.run(transport="streamable-http", host="0.0.0.0", port=8000)
|
|
|