使用Anthropic的MCP进行网络抓取
本指南解释了如何设置MCP服务器进行按需数据提取，如何连接开发工具，以及如何利用Bright data获取即时AI兼容的网络信息。
- 理解局限性：为什么法学硕士在现实世界的互动中需要帮助
- MCP的重要性
- 理解模型上下文协议
- MCP架构说明
- 开发自己的MCP服务器
- 连接您的MCP服务器
- 利用Bright Data的MCP进行专业Web数据提取
- 进一步阅读
理解局限性：为什么法学硕士在现实世界的互动中需要帮助
大型语言模型（LLM）擅长从广泛的训练数据集中处理和生成文本。然而，他们面临着一个关键的限制——他们不能自然地与外部世界互动。这意味着它们缺乏访问本地文件、执行自定义脚本或从网站检索当前信息的内置功能。
考虑一个基本的例子：如果没有额外的工具，让克劳德从活动的亚马逊产品页面中提取细节是不可能的。为什么？因为它没有浏览互联网或触发外部操作的固有能力。
如果没有辅助工具，LLM就无法执行依赖于实时数据或与外部系统集成的实际任务。
就在这里 Anthropic模型上下文协议（MCP） 变得有价值。它使LLM能够以安全和标准化的方式与外部工具（如数据提取器、API或脚本）进行通信。
这就是行动上的区别。集成自定义MCP服务器后，我们直接通过Claude成功提取了结构化的亚马逊产品信息：
MCP的重要性
- 标准化： MCP为基于LLM的系统提供了一个统一的接口，用于连接外部工具和数据，类似于API标准化web集成的方式。这大大减少了对自定义集成的需求，加快了开发速度。
- 灵活性和可扩展性： 开发人员可以替换LLM或托管平台，而无需重写工具集成。MCP支持多种通信方式（例如
stdio)，使其能够适应各种配置。 - 增强的LLM功能： 通过将LLM连接到当前数据和外部工具，MCP允许它们超越静态响应。他们现在可以提供最新的相关信息，并根据上下文触发现实世界的行动。
类比: 将MCP视为LLM的USB接口。就像USB允许不同的设备（键盘、打印机、外部驱动器）插入任何兼容的机器而不需要特殊的驱动程序一样，MCP允许LLM使用标准化协议连接到各种工具，而无需每次进行自定义集成。
理解模型上下文协议
模型上下文协议（MCP）是Anthropic开发的一个开放标准，它使大型语言模型（LLM）能够以一致、安全的方式与外部工具、API和数据源进行交互。它作为一个通用连接器，允许LLM执行真实世界的任务，如提取网站数据、查询数据库或执行脚本。
虽然Anthropic引入了它，但MCP是开放和可扩展的，这意味着任何人都可以实现或贡献该标准。如果你曾与 检索增强生成（RAG），你会欣赏这个概念的。MCP基于这一理念，通过轻量级JSON-RPC接口对交互进行标准化，以便模型可以访问实时数据并采取行动。
MCP架构说明
在其基础上，MCP规范了人工智能模型和外部能力之间的通信。
核心思想： 一个标准化的接口（通常是JSON-RPC 2.0，通过以下传输方式 stdio)允许LLM（通过客户端）发现和调用外部服务器公开的工具。
MCP通过具有三个关键组件的客户端-服务器架构运行：
- MCP主机：启动和管理LLM和外部工具之间交互的环境或应用程序。例子包括AI助手，如 _克劳德桌面版_ 或IDE，如 _光标_.
- MCP客户端：主机内的一个组件，用于建立和维护与MCP服务器的连接，处理通信协议和管理数据交换。
- MCP服务器： 一个实现MCP协议并公开一组特定功能的程序（由我们开发人员创建）。MCP服务器可能与数据库、web服务，或者在我们的例子中，与网站（亚马逊）连接。服务器以标准化的方式公开其功能：
- 工具： 可调用函数（例如。 _scrape_amazon_product_, _获取天气数据_) - 资源： 用于检索静态数据的只读端点（例如，获取文件，返回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
# On macOS/Linux:
source mcp-amazon-scraper/bin/activate
# On Windows:
.\mcp-amazon-scraper\Scripts\activatepip install mcp playwright lxml
# Install browser binaries for Playwright
python -m playwright install这将安装：
- 主控程序：用于处理所有JSON-RPC通信细节的模型上下文协议服务器和客户端的Python SDK
- 剧作家：浏览器自动化库，提供无头浏览器功能，用于渲染和抓取JavaScript繁重的网站
- lxml 文件：快速XML/HTML解析库，使用XPath查询可以轻松从网页中提取特定数据元素
简而言之，MCP Python SDK(mcp)处理所有协议细节，让您公开Claude或Cursor可以通过自然语言提示调用的工具。Playwright允许我们完全渲染网页（包括JavaScript内容），lxml为我们提供了强大的HTML解析功能。
步骤2：启动MCP服务器
创建一个名为的Python文件 amazon_scraper_mcp.py。首先导入所需的模块并初始化 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
# Define a temporary file path for the HTML content
HTML_FILE = os.path.join(os.getenv("TMPDIR", "/tmp"), "amazon_product_page.html")
# Initialize the MCP server with a descriptive name
mcp = FastMCP("Amazon Product Scraper")
print("MCP Server Initialized: Amazon Product Scraper")这将创建MCP服务器的实例。我们现在将为其添加工具。
步骤3：实施 fetch_page 工具
此工具将以URL作为输入，使用Playwright导航到页面，等待内容加载，下载HTML，并将其保存到我们的临时文件中。
@mcp.tool()
async def fetch_page(url: str) -> str:
"""
Fetches the HTML content of the given Amazon product URL using Playwright
and saves it to a temporary file. Returns a status message.
"""
print(f"Executing fetch_page for URL: {url}")
try:
async with async_playwright() as p:
# Launch headless Chromium browser
browser = await p.chromium.launch(headless=True)
page = await browser.new_page()
# Navigate to the URL with a generous timeout
await page.goto(url, timeout=90000, wait_until="domcontentloaded")
# Wait for a key element (e.g., body) to ensure basic loading
await page.wait_for_selector("body", timeout=30000)
# Add a small delay for any dynamic content rendering via 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来处理亚马逊页面上潜在的JavaScript渲染。这 @mcp.tool() 装饰器将此函数注册为服务器中的可调用工具。
步骤4：实施 extract_info 工具
此工具读取由保存的HTML文件 fetch_page，使用LXML和XPath选择器解析它，并返回一个包含提取的产品详细信息的字典。
def _extract_xpath(tree, xpath, default="N/A"):
"""Helper function to extract text using XPath, returning default if not found."""
try:
# Use text_content() to get text from node and children, strip whitespace
result = tree.xpath(xpath)
if result:
return result[0].text_content().strip()
return default
except Exception:
return default
def _extract_price(price_str):
"""Helper function to parse price string into a float."""
if price_str == "N/A":
return None
try:
# Remove currency symbols and commas, handle potential whitespace
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:
"""
Parses the saved HTML file (downloaded by fetch_page) to extract
Amazon product details like title, price, rating, features, etc.
Returns a dictionary of the extracted data.
"""
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 Selectors for Amazon Product Details ---
title = _extract_xpath(tree, '//span[@id="productTitle"]')
# Handle different price structures (main price, sale price)
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")]')
) # Fallback to offscreen if needed
price = _extract_price(price_str)
# Original price (strike-through)
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
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
# Review Count
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
availability = _extract_xpath(
tree,
'//div[@id="availability"]//span/text()',
)
# Features (bullet points)
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()
]
# Calculate Discount
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) # Added for logging
return {"error": error_message}此函数使用LXML fromstring 解析HTML和健壮的XPath选择器以找到所需的元素
步骤5：运行服务器
最后，将以下行添加到您的 amazon_scraper_mcp.py 使用脚本启动服务器 stdio 传输机制，这是本地MCP服务器与Claude Desktop或Cursor等客户端通信的标准。
if __name__ == "__main__":
print("Starting MCP Server with stdio transport...")
# Run the server, listening via standard input/output
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
# Define a temporary file path for the HTML content
HTML_FILE = os.path.join(os.getenv("TMPDIR", "/tmp"), "amazon_product_page.html")
# Initialize the MCP server with a descriptive name
mcp = FastMCP("Amazon Product Scraper")
print("MCP Server Initialized: Amazon Product Scraper")
@mcp.tool()
async def fetch_page(url: str) -> str:
"""
Fetches the HTML content of the given Amazon product URL using Playwright
and saves it to a temporary file. Returns a status message.
"""
print(f"Executing fetch_page for URL: {url}")
try:
async with async_playwright() as p:
# Launch headless Chromium browser
browser = await p.chromium.launch(headless=True)
page = await browser.new_page()
# Navigate to the URL with a generous timeout
await page.goto(url, timeout=90000, wait_until="domcontentloaded")
# Wait for a key element (e.g., body) to ensure basic loading
await page.wait_for_selector("body", timeout=30000)
# Add a small delay for any dynamic content rendering via 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"):
"""Helper function to extract text using XPath, returning default if not found."""
try:
# Use text_content() to get text from node and children, strip whitespace
result = tree.xpath(xpath)
if result:
return result[0].text_content().strip()
return default
except Exception:
return default
def _extract_price(price_str):
"""Helper function to parse price string into a float."""
if price_str == "N/A":
return None
try:
# Remove currency symbols and commas, handle potential whitespace
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:
"""
Parses the saved HTML file (downloaded by fetch_page) to extract
Amazon product details like title, price, rating, features, etc.
Returns a dictionary of the extracted data.
"""
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 Selectors for Amazon Product Details ---
title = _extract_xpath(tree, '//span[@id="productTitle"]')
# Handle different price structures (main price, sale price)
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")]')
) # Fallback to offscreen if needed
price = _extract_price(price_str)
# Original price (strike-through)
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
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
# Review Count
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
availability = _extract_xpath(
tree,
'//div[@id="availability"]//span/text()',
)
# Features (bullet points)
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()
]
# Calculate Discount
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) # Added for logging
return {"error": error_message}
if __name__ == "__main__":
print("Starting MCP Server with stdio transport...")
# Run the server, listening via standard input/output
mcp.run(transport="stdio")连接您的MCP服务器
现在服务器脚本已经准备就绪，让我们将其连接到MCP客户端，如Claude Desktop和Cursor。
使用Claude Desktop进行设置
第一步： 打开克劳德桌面。
第二步： 导航至 Settings -> Developer -> Edit Config。这将打开 claude_desktop_config.json 默认文本编辑器中的文件。
步骤3： 在下面为您的服务器添加条目 mcpServers 钥匙。确保替换中的路径 args 与你的绝对路径 amazon_scraper_mcp.py 文件。
{
"mcpServers": {
"amazon_product_scraper": {
"command": "python", // Or python3 if needed
"args": ["/full/path/to/your/amazon_scraper_mcp.py"], // `Settings` -> `Developer` -> `Edit Config` (`claude_desktop_config.json`).
**步骤3：** 在下面插入Bright Data服务器配置 `mcpServers`。用您的实际凭据替换占位符。
{ "mcpServers": { "Bright Data": { // Choose a name for the server "command": "npx", "args": ["@brightdata/mcp"], "env": { "API_TOKEN": "YOUR_BRIGHTDATA_API_TOKEN", // Paste your API token here "WEB_UNLOCKER_ZONE": "mcp_unlocker", // Your Web Unlocker zone name // Optional: Add if using Scraping Browser tools "BROWSER_AUTH": "brd-customer-ACCOUNTID-zone-YOURZONE:PASSWORD" } } } }
**步骤4：** 保存配置文件并重新启动Claude Desktop。
**步骤5：** 将鼠标悬停在锤子图标上(🔨) 在克劳德桌面。现在，您应该看到多个可用的MCP工具。
让我们尝试从Zillow提取数据，Zillow是一个以潜在限制抓取器而闻名的网站。提示克劳德“_从Zillow URL中提取JSON格式的关键属性数据： [https://www.zillow.com/apartments/arverne-ny/the-tides-at-arverne-by-the-sea/ChWHPZ/](https://www.zillow.com/apartments/arverne-ny/the-tides-at-arverne-by-the-sea/ChWHPZ/)_"
允许Claude使用必要的Bright Data MCP工具。Bright Data的MCP服务器将管理底层的复杂性（代理轮换，如果需要，通过Scraping Browser进行JavaScript渲染）。
Bright Data的服务器进行提取并提供结构化数据，Claude将呈现这些数据。
以下是潜在输出的示例：
{ "propertyInfo": { "name": "The Tides At Arverne By The Sea", "address": "190 Beach 69th St, Arverne, NY 11692", "propertyType": "Apartment building", // ... more info }, "rentPrices": { "studio": { "startingPrice": "$2,750", /* ... */ }, "oneBed": { "startingPrice": "$2,900", /* ... */ }, "twoBed": { "startingPrice": "$3,350", /* ... */ } }, // ... amenities, policies, etc. }
**另一个例子：黑客新闻头条**
一个更直接的问题：“_给我Hacker news最近5篇新闻文章的标题_".
这展示了Bright Data的MCP服务器如何简化直接在AI工作流程中访问动态或高度安全的web内容。
## 进一步阅读
以下是我们早期关于人工智能和大型语言模型（LLM）的指南，以获取更深入的知识：
- [查找法学硕士培训数据的主要来源](https://brightdata.com/blog/web-data/llm-training-data)
- [使用LLaMA 3进行Web抓取：将任何网站转换为结构化JSON](https://brightdata.com/blog/web-data/web-scraping-with-llama-3)
- [基于LangChain和明亮数据的Web抓取](https://brightdata.com/blog/web-data/web-scraping-with-langchain-and-bright-data)
- [如何使用SERP数据使用GPT-4o创建RAG聊天机器人](https://brightdata.com/blog/web-data/build-a-rag-chatbot)
## 结论
Anthropic的模型上下文协议代表了人工智能系统与外部世界交互方式的根本转变。您可以为特定任务构建自定义MCP服务器。Bright Data的MCP集成通过提供企业级网络抓取功能进一步增强了这一点，这些功能可以规避反机器人保护和供应 [AI就绪的结构化数据](https://brightdata.com/use-cases/data-for-ai).
注册并试用 [人工智能解决方案](https://brightdata.com/ai) 今天免费！