在 Haystack 中集成 Bright Data：自定义工具与 MCP

在本教程中，你将学习：

• 什么是 Haystack，以及为什么集成 Bright Data 能让其 AI 流水线和 Agent 更上一层楼。
• 如何开始上手。
• 如何通过自定义工具将 Haystack 与 Bright Data 集成。
• 如何通过 Web MCP 将 Haystack 连接到 60+ 个工具。

让我们开始吧！

Haystack 简介及其对网页数据检索工具的需求

Haystack 是一个开源 AI 框架，用于构建可用于生产环境的 LLM 应用。它可以通过将模型、向量数据库和工具等组件连接成模块化工作流，帮助你创建 RAG 系统、AI Agent 和高级数据流水线。

Haystack 提供了从概念到部署所需的灵活性、自定义能力和可扩展性。这一切都集中在一个开源库中，并在 GitHub 上收获了超过 23k 颗星。

然而，无论你的 Haystack 应用有多么复杂，它仍然会面临 LLM 的核心限制：来自静态训练数据的过时知识，以及缺乏对实时网络的访问！

解决方案是与像 Bright Data 这样的 AI 网页数据提供商集成，它提供网页抓取、搜索、浏览器自动化等工具——从而释放你的 AI 系统的全部潜力！

前置条件

要跟随本教程，你需要：

• 本地安装 Python 3.9+。
• 一个已配置 API 密钥的Bright Data 账户。
• 一个 OpenAI API 密钥（或来自任意 Haystack 支持的其他 LLM的 API 密钥）。

如果你还没有完成这些，请按照官方指南设置账号并生成 Bright Data API 密钥。请将它安全保存，因为很快就会用到。

对Bright Data 产品和服务有一些了解也会有所帮助，同时还需要对Haystack 中工具和 MCP 集成的工作方式有基础认识。

为简单起见，我们假设你已经有一个配置好虚拟环境的 Python 项目。使用以下命令安装 Haystack：

pip install haystack-ai

现在你已经具备了在 Haystack 中开始集成 Bright Data 所需的一切。在这里，我们将探讨两种方式：

• 使用 @tool 注解定义自定义工具。
• 从 Bright Data Web MCP 服务器加载 MCPTool。

在 Haystack 中定义由 Bright Data 驱动的自定义工具

在 Haystack 中访问 Bright Data 能力的一种方式是定义自定义工具。这些工具通过在自定义函数中调用 API 与 Bright Data 产品集成。

为简化这一过程，我们将依赖 Bright Data Python SDK，它提供了一个 Python API，可以轻松调用：

• Web Unlocker API：通过一次请求抓取任意网站并返回干净的 HTML 或 JSON，同时自动处理代理、解封、请求头以及验证码。
• SERP API：在不必担心封锁的情况下，大规模采集来自 Google、Bing 等搜索引擎的结果。
• Web Scraping API：从 Amazon、Instagram、LinkedIn、Yahoo Finance 等热门网站中获取结构化、已解析的数据。
• 以及其他 Bright Data 解决方案……

我们会把主要的 SDK 方法转换为适配 Haystack 的工具，让任意 AI Agent 或流水线都能利用由 Bright Data 驱动的网页数据检索能力！

步骤一：安装并配置 Bright Data Python SDK

首先通过 brightdata-sdk PyPI 包安装 Bright Data Python SDK：

pip install brightdata-sdk

导入库并初始化一个 BrightDataClient 实例：

import os
from brightdata import BrightDataClient

os.environ["BRIGHTDATA_API_TOKEN"] = "<YOUR_BRIGHT_DATA_API_KEY>" # Replace it with your Bright Data API key

# Initialize the Bright Data Python SDK client
client = BrightDataClient(
    token=BRIGHT_DATA_API_KEY,
)

将 <YOUR_BRIGHT_DATA_API_KEY> 占位符替换为你在“前置条件”部分生成的 API 密钥。

在生产代码中，请避免在脚本中硬编码 API 密钥。Bright Data Python SDK 期望从 BRIGHTDATA_API_TOKEN 环境变量中读取它，所以可在全局环境中设置该变量，或使用 python-dotenv 包从 .env 文件加载。

BrightDataClient 会在你的 Bright Data 账户中自动创建默认的 Web Unlocker 和 SERP API Zone：

这两个 Zone 是 SDK 用于暴露其 60+ 个工具所必需的。

如果你已经配置了自定义 Zone，请按文档中的说明进行指定：

client = BrightDataClient(
    serp_zone="serp_api",              # Replace with the name of your SERP API zone
    web_unlocker_zone="web_unlocker",  # Replace with the name of your Web Unlocker API zone
)

太好了！你现在已经准备好将 Bright Data Python SDK 方法转换为 Haystack 工具。

步骤二：将 SDK 函数转换为工具

本节将演示如何把 Bright Data Python SDK 中的 SERP API 和 Web Unlocker 方法转换为 Haystack 工具。掌握这一点后，你就可以轻松地把任意其他 SDK 方法或直接 API 调用转换为 Haystack 工具。

首先，把 SERP API 方法转换为一个可供 AI 调用的工具：

from brightdata import SearchResult
from typing import Union, List
import json
from haystack.tools import Tool

parameters = {
    "type": "object",
    "properties": {
        "query": {
            "type": ["string", "array"],
            "items": {"type": "string"},
            "description": "The search query or a list of queries to run on Google."
        },
        "kwargs": {
            "type": "object",
            "description": "Additional optional parameters for the search (e.g., location, language, device, num_results)."
        }
    },
    "required": ["query"]
}

def serp_api_output_handler(results: Union[SearchResult, List[SearchResult]]) -> str:
    if isinstance(results, list):
        # Handle a list of SearchResult instances
        output = [result.data for result in results]
    else:
        # Handle a single SearchResult
        output = results.data

    return json.dumps(output)

serp_api_tool = Tool(
    name="serp_api_tool",
    description="Calls the Bright Data SERP API to perform web searches and retrieve SERP data from Google.",
    parameters=parameters,
    function=client.search.google,
    outputs_to_string={ "handler": serp_api_output_handler },
    outputs_to_state= {
            "documents": {"handler": serp_api_output_handler }
        }
)

上述代码定义了一个 Haystack Tool，用于调用 Bright Data SERP API。构造该工具时需要提供名称、描述、与输入参数匹配的 JSON Schema，以及要转换为工具的函数。

由于 client.search.google() 返回的是一个特殊对象，因此需要一个自定义输出处理函数将函数输出转换为字符串。该处理器会将结果转换为 JSON，并同时映射为字符串输出和 Agent 状态。

你刚刚定义的这个 Tool 现在可以被 AI Agent 或流水线用来执行 Google 搜索并获取结构化的 SERP 数据。

类似地，为 Web Unlocker 方法创建一个工具：

parameters = {
    "type": "object",
    "properties": {
        "url": {
            "type": ["string", "array"],
            "items": {"type": "string"},
            "description": "The URL or list of URLs to scrape."
        },
        "country": {
            "type": "string",
            "description": "Optional country code to localize the scraping."
        },
    },
    "required": ["url"]
}

def web_unlocker_output_handler(results: Union[ScrapeResult, List[ScrapeResult]]) -> str:
    if isinstance(results, list):
        # Handle a list of ScrapeResult instances
        output = [result.data for result in results]
    else:
        # Handle a single ScrapeResult
        output = results.data

    return json.dumps(output)

web_unlocker_tool = Tool(
    name="web_unlocker_tool",
    description="Calls the Bright Data Web Unlocker API to scrape web pages and retrieve their content.",
    parameters=parameters,
    function=client.scrape.generic.url,
    outputs_to_string={"handler": web_unlocker_output_handler},
    outputs_to_state={"scraped_data": {"handler": web_unlocker_output_handler}}
)

这个新的 Tool 允许 AI Agent 抓取网页并访问其内容，即使这些网页受到反爬虫或反机器人方案的保护。

很好！你现在已经拥有两个可用的 Bright Data Haystack 工具。

步骤三：将工具传入 Haystack AI Agent

上述工具可以被直接调用、传递给对话生成器、用于 Haystack 流水线，或集成到 AI Agent 中。这里我们展示 AI Agent 的用法，你也可以通过文档轻松测试其他三种方式。

首先，Haystack AI Agent 需要一个 LLM 引擎。本示例中我们使用 OpenAI 模型，但任何其他受支持的 LLM 都可以：

from haystack.components.generators.chat import OpenAIChatGenerator

os.environ["OPENAI_API_KEY"] = "<YOUR_OPENAI_API_KEY>" # Replace with your OpenAI API key


# Initialize the LLM engine
chat_generator = OpenAIChatGenerator(
    model="gpt-5-mini"
)

如前所述，在生产脚本中，请从环境变量中加载 OpenAI API 密钥。这里我们配置的是 gpt-5-mini 模型，但任何支持工具调用的 OpenAI 模型都可以使用。其他受支持的生成器同样兼容。

接下来，将该 LLM 引擎与工具结合，创建一个Haystack AI Agent：

from haystack.components.agents import Agent

agent = Agent(
    chat_generator=chat_generator,
    tools=[serp_api_tool, web_unlocker_tool], # The Bright Data tools
)

请注意，这两个 Bright Data 工具是如何被传入 Agent 的 tools 参数的。这使得由 OpenAI GPT-5 Mini 驱动的 AI Agent 得以调用自定义 Bright Data 工具。任务完成！

步骤四：运行 AI Agent

为了测试 Haystack 与 Bright Data 的集成，可以考虑一个同时涉及网页搜索与抓取的任务。例如，使用以下提示词：

Identify the top 3 latest stock market news about the Google company on different topics, access the articles, and provide a brief summary for each.

这可以为任何对投资 Google 感兴趣的人提供一个快速概览。

使用下面的代码片段，在你的 Bright Data 驱动的 Haystack Agent 中执行该提示词：

from haystack.dataclasses import ChatMessage

agent.warm_up()
prompt = """
Identify the top 3 latest stock market news about the Google company on different topics, access the articles, and provide a brief summary for each.
"""
chat_message = ChatMessage.from_user(prompt)

response = agent.run(messages=[chat_message])

然后使用以下代码打印 AI Agent 生成的响应及其工具调用详情：

for msg in response["messages"]:
    role = msg._role.value

    if role == "tool":
        # Log tool outputs
        for content in msg._content:
            print("=== Tool Output ===")
            print(json.dumps(content.result, indent=2))
    elif role == "assistant":
        # Log assistant final messages
        for content in msg._content:
            if hasattr(content, "text"):
                print("=== Assistant Response ===")
                print(content.text)

完美！接下来只需要查看完整代码并执行它，以验证一切工作正常。

步骤五：完整代码

下面是你的 Haystack AI Agent 连接 Bright Data 工具的完整代码：

# pip install haystack-ai brightdata-sdk

import os
from brightdata import BrightDataClient, SearchResult, ScrapeResult
from typing import Union, List
import json
from haystack.tools import Tool
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.agents import Agent
from haystack.dataclasses import ChatMessage

# Set the required environment variables
os.environ["BRIGHTDATA_API_TOKEN"] = "<YOUR_BRIGHT_DATA_API_KEY>" # Replace it with your Bright Data API key
os.environ["OPENAI_API_KEY"] = "<YOUR_OPENAI_API_KEY>" # Replace it with your OpenAI API key

# Initialize the Bright Data Python SDK client
client = BrightDataClient(
    serp_zone="serp_api",              # Replace with the name of your SERP API zone
    web_unlocker_zone="web_unlocker",  # Replace with the name of your Web Unlocker API zone
)

# Turn client.search.google() from Bright Data Python SDK into a Haystack tool
parameters = {
    "type": "object",
    "properties": {
        "query": {
            "type": ["string", "array"],
            "items": {"type": "string"},
            "description": "The search query or a list of queries to run on Google."
        },
        "kwargs": {
            "type": "object",
            "description": "Additional optional parameters for the search (e.g., location, language, device, num_results)."
        }
    },
    "required": ["query"]
}

def serp_api_output_handler(results: Union[SearchResult, List[SearchResult]]) -> str:
    if isinstance(results, list):
        # Handle a list of SearchResult instances
        output = [result.data for result in results]
    else:
        # Handle a single SearchResult
        output = results.data

    return json.dumps(output)

serp_api_tool = Tool(
    name="serp_api_tool",
    description="Calls the Bright Data SERP API to perform web searches and retrieve SERP data from Google.",
    parameters=parameters,
    function=client.search.google,
    outputs_to_string={ "handler": serp_api_output_handler },
    outputs_to_state= {
            "documents": {"handler": serp_api_output_handler }
        }
)

# Turn client.scrape.generic.url() from Bright Data Python SDK into a Haystack tool
parameters = {
    "type": "object",
    "properties": {
        "url": {
            "type": ["string", "array"],
            "items": {"type": "string"},
            "description": "The URL or list of URLs to scrape."
        },
        "country": {
            "type": "string",
            "description": "Optional country code to localize the scraping."
        },
    },
    "required": ["url"]
}

def web_unlocker_output_handler(results: Union[ScrapeResult, List[ScrapeResult]]) -> str:
    if isinstance(results, list):
        # Handle a list of ScrapeResult instances
        output = [result.data for result in results]
    else:
        # Handle a single ScrapeResult
        output = results.data

    return json.dumps(output)

web_unlocker_tool = Tool(
    name="web_unlocker_tool",
    description="Calls the Bright Data Web Unlocker API to scrape web pages and retrieve their content.",
    parameters=parameters,
    function=client.scrape.generic.url,
    outputs_to_string={"handler": web_unlocker_output_handler},
    outputs_to_state={"scraped_data": {"handler": web_unlocker_output_handler}}
)

# Initialize the LLM engine
chat_generator = OpenAIChatGenerator(
    model="gpt-5-mini"
)

# Initialize a Haystack AI agent
agent = Agent(
    chat_generator=chat_generator,
    tools=[serp_api_tool, web_unlocker_tool], # The Bright Data tools
)

## Run the agent
agent.warm_up()
prompt = """
Identify the top 3 latest stock market news about the Google company on different topics, access the articles, and provide a brief summary for each.
"""
chat_message = ChatMessage.from_user(prompt)

response = agent.run(messages=[chat_message])

## Print the output in structured format, with info on tool usage
for msg in response["messages"]:
    role = msg._role.value

    if role == "tool":
        # Log tool outputs
        for content in msg._content:
            print("=== Tool Output ===")
            print(json.dumps(content.result, indent=2))
    elif role == "assistant":
        # Log assistant final messages
        for content in msg._content:
            if hasattr(content, "text"):
                print("=== Assistant Response ===")
                print(content.text)

Et voilà！仅用大约 130 行代码，你便构建了一个能够搜索和抓取网页的 AI Agent，可以完成多种任务并覆盖多种使用场景。

步骤六：测试集成

运行你的脚本，你应该会看到类似如下的结果：

这对应于当前日期“Google stock market news”查询的结果，完全符合预期！

需要注意的是，普通 AI Agent 无法独立完成这项任务。原生 LLM 无法在没有外部工具的情况下直接访问实时网络和搜索引擎。即便是内置的 Grounding 工具通常也有限制、速度较慢，且难以像 Bright Data 那样扩展到任意网站。

日志中包含了所有 SERP API 调用的详细信息：

你还会看到针对 Google 搜索结果中所选新闻文章的 Web Unlocker 调用：

Et voilà！你现在拥有了一个与 Bright Data 工具深度集成的 Haystack AI Agent。

在 Haystack 中集成 Bright Data Web MCP

将 Haystack 连接到 Bright Data 的另一种方式是通过 Web MCP。该 MCP 服务器将 Bright Data 的众多强大能力封装为一大批可供 AI 直接调用的工具。

Web MCP 基于 Bright Data 的网页自动化和数据采集基础设施，提供了60+ 个工具。即使是在免费层中，你也可以使用两个非常实用的工具：

工具	描述
search_engine	以 JSON 或 Markdown 格式获取来自 Google、Bing 或 Yandex 的搜索结果。
scrape_as_markdown	将任意网页抓取为干净的 Markdown，同时绕过反机器人机制。

当你启用付费层（Pro 模式）后，Web MCP 会解锁对各大平台（如 Amazon、Zillow、LinkedIn、YouTube、TikTok、Google Maps 等）的结构化数据采集功能，同时还提供自动化浏览器操作工具。

下面我们来演示如何在 Haystack 中使用 Bright Data 的 Web MCP！

前置条件

开源的 Web MCP 包基于 Node.js 构建。这意味着，如果你想在本地运行 Web MCP 并将其连接到 Haystack AI Agent，你需要先在本机安装 Node.js。

或者，你也可以连接到远程 Web MCP 实例，这种方式无需任何本地安装。

步骤一：安装 MCP–Haystack 集成包

在你的 Python 项目中，运行以下命令安装 MCP–Haystack 集成：

pip install mcp-haystack

该包提供了连接本地或远程 MCP 服务器所需的类。

步骤二：在本地测试 Web MCP

在将 Haystack 连接到 Bright Data 的 Web MCP 之前，先确保你的本机可以本地运行 MCP 服务器。

注意：Web MCP 也可以作为远程服务器使用（通过 SSE 和 Streamable HTTP），更适合企业级场景。

首先，创建一个 Bright Data 账户。如果你已经拥有账户，直接登录即可。为快速上手，请按照仪表盘中“MCP”部分的说明进行配置：

当然，如果需要更多指导，也可以参考下文的详细说明。

首先，生成你的 Bright Data API 密钥。请妥善保存，因为稍后你将使用它来为本地 Web MCP 实例进行身份验证。

然后，通过 @brightdata/mcp 包在本机全局安装 Web MCP：

npm install -g @brightdata/mcp

通过执行以下命令验证 MCP 服务器是否正常工作：

API_TOKEN="<YOUR_BRIGHT_DATA_API>" npx -y @brightdata/mcp

或者，在 PowerShell 中：

$Env:API_TOKEN="<YOUR_BRIGHT_DATA_API>"; npx -y @brightdata/mcp

将 <YOUR_BRIGHT_DATA_API> 占位符替换为你的 Bright Data API 密钥。上述两个（等价的）命令会设置必需的 API_TOKEN 环境变量并启动 Web MCP 服务器。

如果一切正常，你应该会看到类似如下的日志输出：

在首次启动时，Web MCP 会在你的 Bright Data 账户中创建两个 Zone：

• mcp_unlocker：用于 Web Unlocker 的 Zone。
• mcp_browser：用于 Browser API 的 Zone。

这两项服务是 Web MCP 提供其 60+ 个工具所必需的。

要验证这些 Zone 是否已创建，请前往仪表盘中的“Proxies & Scraping Infrastructure”页面。你应该可以在列表中看到这两个 Zone：

请记住，Web MCP 免费层仅提供对 search_engine 和 scrape_as_markdown 工具的访问。

要解锁所有工具，请通过设置 PRO_MODE="true" 环境变量启用 Pro 模式：

API_TOKEN="<YOUR_BRIGHT_DATA_API>" PRO_MODE="true" npx -y @brightdata/mcp

或者，在 Windows 上：

$Env:API_TOKEN="<YOUR_BRIGHT_DATA_API>"; $Env:PRO_MODE="true"; npx -y @brightdata/mcp

Pro 模式会解锁全部 60+ 个工具，但不包含在免费层中，可能会产生额外费用。

做得很好！你已经确认 Web MCP 服务器能在本机成功运行。现在可以终止该进程，接下来你将把 Haystack 配置为在本地启动该服务器并连接到它。

步骤三：在 Haystack 中连接 Web MCP

使用以下代码连接 Web MCP：

from haystack_integrations.tools.mcp import StdioServerInfo, MCPToolset

BRIGHT_DATA_API_KEY = "<YOUR_BRIGHT_DATA_API_KEY>" # Replace it with your Bright Data API key

# Configuration to connect to the Web MCP server via STDIO
web_mcp_server_info = StdioServerInfo(
    command="npx",
    args=["-y", "@brightdata/mcp"],
    env={
        "API_TOKEN": BRIGHT_DATA_API_KEY,
        "PRO_MODE": "true"  # Enable Pro tools (optional)
    }
)

上述 StdioServerInfo 对象等价于你之前测试的 npx 命令，只不过将其封装为 Haystack 可调用的形式。它同时设置了配置 Web MCP 服务器所需的环境变量：

• API_TOKEN：必需。设置为你之前生成的 Bright Data API 密钥。
• PRO_MODE：可选。如果你希望保持在免费层，仅访问 search_engine 和 scrape_as_markdown 工具，则删除该变量。

接下来，通过以下代码访问 Web MCP 暴露的所有工具：

web_mcp_toolset = MCPToolset(
    server_info=web_mcp_server_info,
    invocation_timeout=180 # 3 minutes
)

通过加载工具并打印信息，验证集成是否正常工作：

web_mcp_toolset.warm_up()
for tool in web_mcp_toolset.tools:
  print(f"Name: {tool.name}")
  print(f"Description: {tool.name}\n")

如果你启用了 Pro 模式，你应该会看到全部 60+ 个工具：

很好！Bright Data Web MCP 在 Haystack 中的集成已经正常工作。

步骤四：测试集成

在完成工具设置后，可以在AI Agent（如前所示）或Haystack 流水线中使用这些工具。比如，假设你希望 AI Agent 处理以下提示词：

Return a Markdown report with useful insights from the following Crunchbase company URL:
"https://www.crunchbase.com/organization/apple"

这就是一个需要 Web MCP 工具才能完成的任务示例。

在 Agent 中执行该任务：

agent = Agent(
    chat_generator=chat_generator,
    tools=web_mcp_toolset, # The Bright Data Web MCP tools
)

## Run the agent
agent.warm_up()
prompt = """
Return a Markdown report with useful insights from the following Crunchbase company URL:
"https://www.crunchbase.com/organization/apple"
"""
chat_message = ChatMessage.from_user(prompt)

response = agent.run(messages=[chat_message])

结果可能如下所示：

被调用的工具是 Pro 工具 web_data_crunchbase_company：

在底层，该工具依赖 Bright Data Crunchbase Scraper，从指定的 Crunchbase 页面提取结构化信息。

抓取 Crunchbase 显然不是普通 LLM 单独就能完成的任务！这进一步体现了在 Haystack 中集成 Web MCP 的强大之处，它支持大量的使用场景。

步骤五：完整代码

下面是将 Bright Data Web MCP 连接到 Haystack 的完整代码：

# pip install haystack-ai mcp-haystack

from haystack_integrations.tools.mcp import StdioServerInfo, MCPToolset
import os
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.components.agents import Agent
from haystack.dataclasses import ChatMessage
import json

os.environ["OPENAI_API_KEY"] = "<YOUR_OPENAI_API_KEY>" # Replace it with your Bright Data API key

BRIGHT_DATA_API_KEY = "<YOUR_BRIGHT_DATA_API_KEY>" # Replace it with your Bright Data API key

# Configuration to connect to the Web MCP server via STDIO
web_mcp_server_info = StdioServerInfo(
    command="npx",
    args=["-y", "@brightdata/mcp"],
    env={
        "API_TOKEN": BRIGHT_DATA_API_KEY,
        "PRO_MODE": "true"  # Enable Pro tools (optional)
    }
)

# Load the available MCP tools exposed by the Web MCP server
web_mcp_toolset = MCPToolset(
    server_info=web_mcp_server_info,
    invocation_timeout=180, # 3 minutes
    tool_names=["web_data_crunchbase_company"]
)

# Initialize the LLM engine
chat_generator = OpenAIChatGenerator(
    model="gpt-5-mini"
)

# Initialize a Haystack AI agent
agent = Agent(
    chat_generator=chat_generator,
    tools=web_mcp_toolset, # The Bright Data Web MCP tools
)

## Run the agent
agent.warm_up()
prompt = """
Return a Markdown report with useful insights from the following Crunchbase company URL:
"https://www.crunchbase.com/organization/apple"
"""
chat_message = ChatMessage.from_user(prompt)

response = agent.run(messages=[chat_message])

## Print the output in structured format, with info on tool usage
for msg in response["messages"]:
    role = msg._role.value

    if role == "tool":
        # Log tool outputs
        for content in msg._content:
            print("=== Tool Output ===")
            print(json.dumps(content.result, indent=2))
    elif role == "assistant":
        # Log assistant final messages
        for content in msg._content:
            if hasattr(content, "text"):
                print("=== Assistant Response ===")
                print(content.text)

结论

在本指南中，你学习了如何通过自定义工具或 MCP 在 Haystack 中集成 Bright Data。

通过这一套配置，Haystack 中的 AI Agent 和流水线可以执行网页搜索、提取结构化数据、访问实时网页数据流，并自动化 Web 交互。这一切都得益于 Bright Data AI 生态中完整的服务套件。

立即免费创建一个 Bright Data 账户，开始探索我们功能强大的、可直接用于 AI 的网页数据工具吧！