ScrapingBee

ScrapingBee MCP 服务

一款适用于生产环境的模型上下文协议（MCP）服务，让 AI Agent 能够使用 ScrapingBee 这一工具访问结构化网页数据。

本项目演示了如何将 ScrapingBee 转化为 AI 原生数据层。AI Agent 无需将抓取逻辑直接嵌入提示词，而是将 ScrapingBee 注册为工具，并按需请求结构化网页数据。

如果你正在构建：

• 需要实时网页访问的 AI Agent
• 自主研究系统
• 电商智能机器人
• 市场监控 Agent
• 使用工具执行功能的大模型工作流
• AI 驱动的网页抓取基础设施

本仓库提供了完整的 MCP 集成蓝图。

快速开始

该服务由 ScrapingBee 托管，地址为 https://mcp.scrapingbee.com/mcp（Streamable HTTP）—— 无需安装任何内容。你需要一个 ScrapingBee API 密钥 （在此获取，包含试用额度）， 并通过标头 Authorization: Bearer <key> 发送。注册表名称：com.scrapingbee/mcp。

Claude Code

claude mcp add --transport http scrapingbee https://mcp.scrapingbee.com/mcp \
  --header "Authorization: Bearer YOUR_API_KEY"

Codex CLI

codex mcp add scrapingbee --url https://mcp.scrapingbee.com/mcp \
  --bearer-token-env-var SCRAPINGBEE_API_KEY

Cursor、VS Code 及其他使用 JSON 配置的客户端

{
  "mcpServers": {
    "scrapingbee": {
      "url": "https://mcp.scrapingbee.com/mcp",
      "headers": { "Authorization": "Bearer YOUR_API_KEY" }
    }
  }
}

也可通过 Glama、 Smithery 和 MCP Registry 使用。

工具

工具	功能
get_page_text	获取 URL 并将其内容作为文本或 Markdown 返回（渲染 JavaScript，处理区块）
get_page_html	获取 URL 并返回渲染后的 HTML
get_screenshot	获取 URL 并返回截图
get_file	从 URL 下载文件（PDF、图片等）
extract_page_data	获取 URL，并使用 CSS 或 XPath 选择器提取特定数据
fast_search	快速网页搜索，返回精简的自然搜索结果和头条新闻——适用于常规搜索的首选工具
get_google_search_results	以 JSON 格式返回 Google SERP，适用于新闻、地图、Lens、购物、图片和 AI 模式等专业搜索
get_amazon_search_results	以 JSON 格式返回 Amazon 搜索结果
get_amazon_product_details	根据 ASIN 获取 Amazon 商品详情
get_amazon_pricing	根据 ASIN 获取 Amazon 当前价格和卖家报价
get_walmart_search_results	以 JSON 格式返回 Walmart 搜索结果
get_walmart_product_details	根据 ID 获取 Walmart 商品详情
get_youtube_search_results	获取 YouTube 搜索结果
get_youtube_video_metadata	获取视频标题、描述、统计信息等
get_youtube_video_subtitles	获取视频字幕 / 转录文本
ask_chatgpt	向 ChatGPT 发送提示词，可选择启用网页搜索
ask_gemini	向 Gemini 发送提示词，并在可用时提供引用
get_scrapingbee_usage	获取账户剩余额度和并发数

每次调用消耗的 API 额度与对应的 ScrapingBee API 请求完全相同 （参见定价和 API 文档）。服务每次调用处理一个页面或查询；如果需要处理数百个 URL、爬取网站、将结果写入文件或安排周期性任务，请使用 ScrapingBee CLI。

相关关键词： scrapingbee mcp
model context protocol server
mcp scraping
ai agent web scraping
mcp web scraping tool

什么是模型上下文协议（MCP）？

模型上下文协议（MCP）定义了 AI 系统如何以结构化方式与外部工具交互。

与其让语言模型猜测如何抓取或浏览网页，MCP 允许它：

1. 发现可用工具
2. 使用结构化参数调用工具
3. 接收规范化的 JSON 响应
4. 将结果注入推理循环

ScrapingBee MCP 服务将网页抓取功能作为正式注册的工具提供，AI 系统可以安全、可预测地使用它。

为什么将 ScrapingBee 与 MCP 配合使用？

AI 模型无法可靠地：

• 绕过反机器人保护
• 管理代理轮换
• 渲染 JavaScript 密集型页面
• 处理 CAPTCHA 验证
• 维护抓取基础设施

ScrapingBee 会处理所有这些工作。

MCP 服务充当 AI Agent 与 ScrapingBee 托管抓取基础设施之间的桥梁。

这样，无需脆弱的基于提示词的技巧，就能真正实现 AI Agent 网页抓取。

架构概览

AI Agent
→ MCP 客户端
→ ScrapingBee MCP 服务
→ ScrapingBee API
→ 目标网站
→ 结构化 JSON 响应
→ 注入回 Agent 上下文

AI Agent 不会直接抓取网页。
它会调用工具。
MCP 服务负责执行。

安装

无需安装或运行任何内容：该服务由 ScrapingBee 运营，地址为 https://mcp.scrapingbee.com/mcp。唯一需要配置的是 API 密钥，下面的每个客户端都会将其作为 Authorization: Bearer 标头发送：

export SCRAPINGBEE_API_KEY=your_api_key

连接到 ScrapingBee MCP 服务

你可以通过两种方式连接：

1. 通过 Claude Desktop（远程 MCP）
2. 通过自定义 Python MCP 客户端

选项 1：连接到 Claude Desktop

将 ScrapingBee MCP 服务添加到 claude_desktop_config.json 文件中：

{
  "mcpServers": {
    "scrapingbee": {
      "command": "npx",
      "args": [
        "mcp-remote",
        "https://mcp.scrapingbee.com/mcp",
        "--header",
        "Authorization:${AUTH_HEADER}"
      ],
      "env": {
        "AUTH_HEADER": "Bearer YOUR_API_KEY"
      }
    }
  }
}

（AUTH_HEADER 变量用于规避 mcp-remote 对参数中空格的处理问题。）

这样便会在 Claude Desktop 中将 ScrapingBee 注册为 MCP 工具提供方。

重启 Claude 后，这些工具便会自动可用。

选项 2：自定义 Python MCP 客户端

对于高级 AI 工作流，你可以使用 Python 构建一个精简的异步 MCP 客户端。

下面是一个完整可运行的示例。

custom_mcp_client.py

import os
import re
import json
import asyncio
import httpx
import google.generativeai as genai
from typing import Dict, Any, List

MCP_SERVER_URL = "https://mcp.scrapingbee.com/mcp"
SCRAPINGBEE_API_KEY = os.environ.get("SCRAPINGBEE_API_KEY")
GEMINI_API_KEY = os.environ.get("GEMINI_API_KEY")

MCP 客户端实现

class MCPClient:
    def __init__(self, base_url: str, api_key: str):
        self.base_url = base_url
        self.session_id = None
        self.http_client = httpx.AsyncClient(timeout=300)
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
            "Accept": "application/json, text/event-stream",
            "User-Agent": "Minimal-MCP-Client/1.0",
        }

    async def _send_request(self, payload: Dict[str, Any]):
        headers = self.headers.copy()
        if self.session_id:
            headers["Mcp-Session-Id"] = self.session_id

        response = await self.http_client.post(self.base_url, json=payload, headers=headers)
        response.raise_for_status()

        text = (await response.aread()).decode("utf-8")

        if "data:" in text:
            match = re.search(r"data:\s*({.*})", text, re.DOTALL)
            if match:
                return json.loads(match.group(1))

        if response.status_code in [202, 204] or not text:
            return {"status": "ok"}

        return json.loads(text)

    async def initialize(self):
        init_payload = {
            "jsonrpc": "2.0",
            "id": "1",
            "method": "initialize",
            "params": {
                "protocolVersion": "2024-11-05",
                "capabilities": {},
                "clientInfo": {"name": "MinimalClient"},
            },
        }

        async with self.http_client.stream(
            "POST", self.base_url, json=init_payload, headers=self.headers
        ) as response:
            self.session_id = response.headers.get("Mcp-Session-Id")

        notify_payload = {
            "jsonrpc": "2.0",
            "method": "notifications/initialized",
            "params": {}
        }

        await self._send_request(notify_payload)

        list_payload = {
            "jsonrpc": "2.0",
            "id": "2",
            "method": "tools/list",
            "params": {}
        }

        return await self._send_request(list_payload)

    async def call_tool(self, tool_name: str, arguments: Dict[str, Any]):
        payload = {
            "jsonrpc": "2.0",
            "id": "3",
            "method": "tools/call",
            "params": {
                "name": tool_name,
                "arguments": arguments
            }
        }

        return await self._send_request(payload)

将 MCP 工具转换为 Gemini 工具

def format_tools_for_gemini(tools: List[Dict[str, Any]]):
    type_map = {
        "string": genai.protos.Type.STRING,
        "integer": genai.protos.Type.INTEGER,
        "number": genai.protos.Type.NUMBER,
        "boolean": genai.protos.Type.BOOLEAN,
    }

    gemini_tools = []

    for tool in tools:
        properties = {}
        input_schema = tool.get("inputSchema", {})

        if "properties" in input_schema:
            for name, prop in input_schema["properties"].items():
                properties[name] = genai.protos.Schema(
                    type=type_map.get(prop.get("type", "string")),
                    description=prop.get("description", "")
                )

        func = genai.protos.FunctionDeclaration(
            name=tool["name"],
            description=tool["description"],
            parameters=genai.protos.Schema(
                type=genai.protos.Type.OBJECT,
                properties=properties,
                required=input_schema.get("required", [])
            ),
        )

        gemini_tools.append(
            genai.protos.Tool(function_declarations=[func])
        )

    return gemini_tools

完整的 AI → MCP → 工具执行流程

async def main():
    mcp_client = MCPClient(MCP_SERVER_URL, SCRAPINGBEE_API_KEY)
    tools_response = await mcp_client.initialize()

    genai.configure(api_key=GEMINI_API_KEY)
    gemini_tools = format_tools_for_gemini(tools_response["result"]["tools"])

    model = genai.GenerativeModel(
        model_name="gemini-2.5-pro",
        tools=gemini_tools
    )

    user_prompt = input("Ask the AI to perform a task:\n> ")

    response = model.generate_content(user_prompt)
    function_call = response.candidates[0].content.parts[0].function_call

    tool_name = function_call.name
    tool_args = dict(function_call.args)

    result = await mcp_client.call_tool(tool_name, tool_args)

    print(json.dumps(result, indent=2))

这能实现什么

使用此架构，你可以：

• 构建能够根据可用工具进行推理的 AI Agent
• 允许大模型动态选择抓取操作
• 执行结构化网页数据提取
• 将真实世界的数据注入模型上下文
• 创建可扩展、AI 原生的网页抓取系统

结论

ScrapingBee MCP 服务 将网页抓取转变为面向 AI 系统的结构化、工具驱动数据层。

无需脆弱的提示词抓取，你可以获得：

• 符合模型上下文协议
• 基于 JSON-RPC 的结构化执行
• AI Agent 网页抓取
• 可扩展的工具编排

这就是 AI 原生的网页数据访问。

立即开始构建 AI 驱动的 Agent，获取免费 API 密钥，几分钟内即可连接到 ScrapingBee MCP 服务。