网络爬虫与MCP
项目介绍
技术文档摘要
使用Anthropic的MCP进行网络爬取
本指南解释了如何设置一个MCP服务器以实现按需数据提取，连接开发工具，并利用Bright Data获取即时AI兼容的网络信息。
- 理解限制：为什么LLMs需要帮助与现实世界交互
- MCP的重要性
- 理解模型上下文协议
- MCP架构解析
- 开发自己的MCP服务器
- 连接您的MCP服务器
- 使用Bright Data的MCP进行专业网络数据提取
- 进一步阅读
理解限制：为什么LLMs需要帮助与现实世界交互
大型语言模型（LLMs）在处理和生成来自广泛训练数据集的文本方面表现出色。然而，它们面临一个关键约束——它们无法自然地与外部世界互动。这意味着它们缺乏访问本地文件、执行自定义脚本或从网站检索当前信息的能力。
考虑一个基本示例：要求Claude从活跃的亚马逊产品页面中提取详细信息是不可能的，除非有额外的工具。为什么？因为Claude没有浏览互联网或触发外部操作的内在能力。
没有补充工具，LLMs无法执行依赖于实时数据或与外部系统集成的实际任务。
这就是Anthropic的模型上下文协议（MCP）变得有价值的地方。它使LLMs能够通过安全且标准化的方式与外部工具（如数据提取器、API或脚本）通信。
这是实际操作中的区别。在整合了一个自定义MCP服务器后，我们成功地通过Claude直接提取了结构化的亚马逊产品信息：
MCP的重要性
- 标准化：MCP为基于LLM的系统提供了一种统一的接口，用于连接到外部工具和数据——类似于API标准化了网络集成。这显著减少了定制集成的需求，加速了开发。
- 灵活性和可扩展性：开发者可以替换LLMs或托管平台，而无需重写工具集成。MCP支持多种通信方法（如
stdio），使其适应各种配置。 - 增强LLM功能：通过将LLMs连接到当前数据和外部工具，MCP允许它们超越静态响应。现在，它们可以提供最新的相关信息并根据上下文触发实际操作。
类比：
将MCP视为LLMs的USB接口。就像USB允许不同的设备（键盘、打印机、外部驱动器）插到任何兼容的机器上而不需要特殊驱动程序一样，MCP让LLMs使用标准化协议连接到广泛的工具——每次都不需要定制集成。
理解模型上下文协议
模型上下文协议（MCP）是由Anthropic开发的一个开放标准，它使大型语言模型（LLMs）能够以一致且安全的方式与外部工具、API和数据源交互。它充当通用连接器，使LLMs能够执行诸如提取网站数据、查询数据库或执行脚本等实际任务。
虽然Anthropic引入了它，但MCP是开放和可扩展的，意味着任何人都可以实施或贡献到该标准。如果您已经了解过检索增强生成（RAG），您会欣赏这个概念。MCP在此基础上发展，通过轻量级JSON-RPC接口标准化交互，以便模型可以访问实时数据并采取行动。
MCP架构解析
在基础层面，MCP标准化了AI模型与外部能力之间的通信。
核心思想：一个标准化的接口（通常是在stdio传输上的JSON-RPC 2.0）允许LLM（通过客户端）发现并调用由外部服务器暴露的工具。
MCP通过客户端-服务器架构运行，包含三个关键组件：
- MCP主机：启动和管理LLM与外部工具之间交互的环境或应用程序。例如，AI助手如_Claude Desktop_或IDE如_Cursor_。
- MCP客户端：主机内的一个组件，建立并维护与MCP服务器的连接，处理通信协议并管理数据交换。
- MCP服务器：我们开发者创建的程序，实现了MCP协议并暴露特定的一组能力。MCP服务器可能与数据库、网络服务或在我们的案例中，与网站（亚马逊）接口。服务器以标准化方式暴露其功能：
- 工具：可调用函数（例如_scrape_amazon_product_，get_weather_data）
- 资源：只读端点，用于检索静态数据（例如获取文件，返回JSON记录）
- 提示：预定义模板，指导LLM与工具和资源的交互
以下是MCP架构图：
图片来源：模型上下文协议
在这种设置下，主机（Claude Desktop或Cursor IDE）启动一个MCP客户端，然后连接到外部MCP服务器。该服务器暴露工具、资源和提示，允许AI根据需要与其交互。
简而言之，工作流程如下：
- 用户发送消息，如_"从这个亚马逊链接获取产品信息"._
- MCP客户端检查是否有注册的工具可以处理该任务
- 客户端向MCP服务器发送结构化请求
- MCP服务器执行适当的操作（例如，启动无头浏览器）
- 服务器将结构化结果返回给MCP客户端
- 客户端将结果转发给LLM，LLM将其呈现给用户
开发自己的MCP服务器
让我们构建一个Python MCP服务器来从亚马逊产品页面提取数据。
此服务器将提供两个工具：一个下载HTML，另一个提取组织的信息。您将通过Cursor或Claude Desktop中的LLM客户端与服务器交互。
第一步：准备环境
首先，验证已安装Python 3。然后，创建并激活虚拟环境：
python -m venv mcp-amazon-scraper
# 在macOS/Linux上：
source mcp-amazon-scraper/bin/activate
# 在Windows上：
.\mcp-amazon-scraper\Scripts\activate
安装必要的库：MCP Python SDK，Playwright 和 LXML。
pip install mcp playwright lxml
# 安装Playwright的浏览器二进制文件
python -m playwright install
这将安装：
- mcp：处理所有JSON-RPC通信细节的Model Context Protocol服务器和客户端的Python SDK
- playwright：提供无头浏览器能力以渲染和抓取JavaScript密集型网站的浏览器自动化库
- lxml：快速的XML/HTML解析库，使用XPath查询轻松从网页中提取特定数据元素
简而言之，MCP Python SDK（mcp）处理所有协议细节，让您能够暴露Claude或Cursor可以通过自然语言提示调用的工具。Playwright允许我们完全渲染网页（包括JavaScript内容），而lxml提供了强大的HTML解析能力。
第二步：启动MCP服务器
创建一个名为amazon_scraper_mcp.py的Python文件。开始导入所需的模块并初始化FastMCP服务器：
import os
import asyncio
from lxml import html as lxml_html
from mcp.server.fastmcp import FastMCP
from playwright.async_api import async_playwright
# 定义HTML内容的临时文件路径
HTML_FILE = os.path.join(os.getenv("TMPDIR", "/tmp"), "amazon_product_page.html")
# 使用描述性名称初始化MCP服务器
mcp = FastMCP("Amazon Product Scraper")
print("MCP Server Initialized: Amazon Product Scraper")
这创建了一个MCP服务器实例。我们现在将添加工具到其中。
第三步：实现fetch_page工具
此工具将接受一个URL作为输入，使用Playwright导航到该页面，等待内容加载，下载HTML，并将其保存到我们的临时文件中。
@mcp.tool()
async def fetch_page(url: str) -> str:
"""
使用Playwright获取给定Amazon产品URL的HTML内容，并将其保存到临时文件中。返回状态消息。
"""
print(f"Executing fetch_page for URL: {url}")
try:
async with async_playwright() as p:
# 启动无头Chromium浏览器
browser = await p.chromium.launch(headless=True)
page = await browser.new_page()
# 导航到URL，带有慷慨的超时时间
await page.goto(url, timeout=90000, wait_until="domcontentloaded")
# 等待一个关键元素（例如，body）确保基本加载
await page.wait_for_selector("body", timeout=30000)
# 添加一个小延迟，以处理JavaScript渲染的动态内容
await asyncio.sleep(5)
html_content = await page.content()
with open(HTML_FILE, "w", encoding="utf-8") as f:
f.write(html_content)
await browser.close()
print(f"Successfully fetched and saved HTML to {HTML_FILE}")
return f"HTML content for {url} downloaded and saved successfully to {HTML_FILE}."
except Exception as e:
error_message = f"Error fetching page {url}: {str(e)}"
print(error_message)
return error_message
这个异步函数使用Playwright处理Amazon页面上的潜在JavaScript渲染。@mcp.tool()装饰器将此函数注册为服务器内的可调用工具。
第四步：实现extract_info工具
此工具读取由fetch_page保存的HTML文件，使用LXML和XPath选择器解析它，并返回一个包含提取的产品详细信息的字典。
def _extract_xpath(tree, xpath, default="N/A"):
"""辅助函数，使用XPath提取文本，如果未找到则返回默认值。"""
try:
# 使用text_content()从节点及其子节点获取文本，去除空白
result = tree.xpath(xpath)
if result:
return result[0].text_content().strip()
return default
except Exception:
return default
def _extract_price(price_str):
"""辅助函数，将价格字符串解析为浮点数。"""
if price_str == "N/A":
return None
try:
# 移除货币符号和逗号，处理潜在的空白
cleaned_price = "".join(filter(str.isdigit or str.__eq__("."), price_str))
return float(cleaned_price)
except (ValueError, TypeError):
return None
@mcp.tool()
def extract_info() -> dict:
"""
解析保存的HTML文件（由fetch_page下载）以提取
亚马逊产品详情，如标题、价格、评分、特性等。
返回一个包含提取数据的字典。
"""
print(f"Executing extract_info from file: {HTML_FILE}")
if not os.path.exists(HTML_FILE):
return {
"error": f"HTML file not found at {HTML_FILE}. Please run fetch_page first."
}
try:
with open(HTML_FILE, "r", encoding="utf-8") as f:
page_html = f.read()
tree = lxml_html.fromstring(page_html)
# --- XPath选择器用于Amazon产品详情 ---
title = _extract_xpath(tree, '//span[@id="productTitle"]')
# 处理不同的价格结构（主要价格、促销价格）
price_whole = _extract_xpath(tree, '//span[contains(@class, "a-price-whole")]')
price_fraction = _extract_xpath(
tree, '//span[contains(@class, "a-price-fraction")]'
)
price_str = (
f"{price_whole}.{price_fraction}"
if price_whole != "N/A"
else _extract_xpath(tree, '//span[contains(@class,"a-offscreen")]')
) # 如果需要，回退到屏幕外
price = _extract_price(price_str)
# 原价（划线）
original_price_str = _extract_xpath(
tree, '//span[@class="a-price a-text-price"]//span[@class="a-offscreen"]'
)
original_price = _extract_price(original_price_str)
# 评分
rating_text = _extract_xpath(tree, '//span[@id="acrPopover"]/@title')
rating = None
if rating_text != "N/A":
try:
rating = float(rating_text.split()[0])
except (ValueError, IndexError):
rating = None
# 评论数量
reviews_text = _extract_xpath(tree, '//span[@id="acrCustomerReviewText"]')
review_count = None
if reviews_text != "N/A":
try:
review_count = int(reviews_text.split()[0].replace(",", ""))
except (ValueError, IndexError):
review_count = None
# 可用性
availability = _extract_xpath(
tree,
'//div[@id="availability"]//span/text()',
)
# 特性（项目符号）
feature_elements = tree.xpath(
'//div[@id="feature-bullets"]//li//span[@class="a-list-item"]'
)
features = [
elem.text_content().strip()
for elem in feature_elements
if elem.text_content().strip()
]
# 计算折扣
discount = None
if price and original_price and original_price > price:
discount = round(((original_price - price) / original_price) * 100)
extracted_data = {
"title": title,
"price": price,
"original_price": original_price,
"discount_percent": discount,
"rating_stars": rating,
"review_count": review_count,
"features": features,
"availability": availability.strip(),
}
print(f"Successfully extracted data: {extracted_data}")
return extracted_data
except Exception as e:
error_message = f"Error parsing HTML: {str(e)}"
print(error_message) # 用于日志记录
return {"error": error_message}
此函数使用LXML的fromstring解析HTML，并使用健壮的XPath选择器查找所需元素
第五步：运行服务器
最后，在您的amazon_scraper_mcp.py脚本末尾添加以下行，使用stdio传输机制启动服务器，这是本地MCP服务器与像Claude Desktop或Cursor这样的客户端通信的标准方式。
if __name__ == "__main__":
print("Starting MCP Server with stdio transport...")
# 运行服务器，通过标准输入/输出监听
mcp.run(transport="stdio")
完整源代码
import os
import asyncio
from lxml import html as lxml_html
from mcp.server.fastmcp import FastMCP
from playwright.async_api import async_playwright
# 定义HTML内容的临时文件路径
HTML_FILE = os.path.join(os.getenv("TMPDIR", "/tmp"), "amazon_product_page.html")
# 使用描述性名称初始化MCP服务器
mcp = FastMCP("Amazon Product Scraper")
print("MCP Server Initialized: Amazon Product Scraper")
@mcp.tool()
async def fetch_page(url: str) -> str:
"""
使用Playwright获取给定Amazon产品URL的HTML内容，并将其保存到临时文件中。返回状态消息。
"""
print(f"Executing fetch_page for URL: {url}")
try:
async with async_playwright() as p:
# 启动无头Chromium浏览器
browser = await p.chromium.launch(headless=True)
page = await browser.new_page()
# 导航到URL，带有慷慨的超时时间
await page.goto(url, timeout=90000, wait_until="domcontentloaded")
# 等待一个关键元素（例如，body）确保基本加载
await page.wait_for_selector("body", timeout=30000)
# 添加一个小延迟，以处理JavaScript渲染的动态内容
await asyncio.sleep(5)
html_content = await page.content()
with open(HTML_FILE, "w", encoding="utf-8") as f:
f.write(html_content)
await browser.close()
print(f"Successfully fetched and saved HTML to {HTML_FILE}")
return f"HTML content for {url} downloaded and saved successfully to {HTML_FILE}."
except Exception as e:
error_message = f"Error fetching page {url}: {str(e)}"
print(error_message)
return error_message
def _extract_xpath(tree, xpath, default="N/A"):
"""辅助函数，使用XPath提取文本，如果未找到则返回默认值。"""
try:
# 使用text_content()从节点及其子节点获取文本，去除空白
result = tree.xpath(xpath)
if result:
return result[0].text_content().strip()
return default
except Exception:
return default
def _extract_price(price_str):
"""辅助函数，将价格字符串解析为浮点数。"""
if price_str == "N/A":
return None
try:
# 移除货币符号和逗号，处理潜在的空白
cleaned_price = "".join(filter(str.isdigit or str.__eq__("."), price_str))
return float(cleaned_price)
except (ValueError, TypeError):
return None
@mcp.tool()
def extract_info() -> dict:
"""
解析保存的HTML文件（由fetch_page下载）以提取
亚马逊产品详情，如标题、价格、评分、特性等。
返回一个包含提取数据的字典