企业官网建设流程全解析

Haystack MCP 集成完全指南：用 MCPTool 与 MCPToolset 接入外部工具生态

【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack

<output文章>

Haystack MCP 集成完全指南：用 MCPTool 与 MCPToolset 接入外部工具生态

导读

本指南围绕 Haystack 的 MCP（Model Context Protocol）集成展开，讲解MCPTool与MCPToolset两大核心类如何让 Haystack Agent / Pipeline 通过Streamable HTTP、SSE、StdIO三种传输方式接入任意 MCP 兼容服务器，自动发现并调用其外部工具。读完本文，你将掌握从ServerInfo配置、MCPClient连接管理、错误处理，到与 Agent State 联动的完整链路，能够把时间服务、Git 操作等外部能力直接注入自己的 RAG 与 Agent 应用。

一、MCP 与 Haystack：为什么需要这个集成

Model Context Protocol（MCP）是当前 LLM 应用生态中标准化"外部工具与上下文"接入方式的核心协议。在 Haystack 中，工具（Tool）是 Agent 调用外部能力的统一抽象，而 MCP 集成将这一抽象延伸到整个 MCP 生态：

• 一个连接、多个工具：MCP 服务器通常会暴露一组相关工具（如mcp-server-time提供时间查询、mcp-server-git提供 Git 操作），Haystack 集成可以一次连接、自动发现全部工具。
• 协议由官方 SDK 承载：根据 MCPTool 参考文档 的说明，本实现使用官方 MCP SDK 处理协议细节，同时保持与 Haystack 工具生态的兼容。
• 三种传输方式：Streamable HTTP（面向远程 HTTP 服务器，当前推荐）、SSE（Server-Sent Events，已被 MCP 规范弃用但仍可连接旧服务器）、StdIO（直接以子进程方式运行本地程序）。

补充说明：MCPTool与MCPToolset属于独立的mcp-haystack集成包，源码维护在 haystack-core-integrations 仓库；本仓库中与之对应的基础抽象（Tool、Toolset）实现在 haystack/tools/tool.py 与 haystack/tools/toolset.py，下文会结合这些基类源码做原理级解读。

安装

pip install mcp-haystack

参考使用文档 mcptool.mdx 与 mcptoolset.mdx，安装后即可从以下命名空间导入：

from haystack_integrations.tools.mcp import ( MCPTool, MCPToolset, StdioServerInfo, SSEServerInfo, StreamableHttpServerInfo, )

二、架构总览：ServerInfo → MCPClient → Tool / Toolset

从 参考 API 文档 的类结构可以看出，MCP 集成分为三层：

1. MCPServerInfo（服务器配置层）：抽象基类，统一定义 MCP 服务器的连接参数，并提供create_client()、to_dict()、from_dict()三个方法。具体实现有三种：SSEServerInfo、StreamableHttpServerInfo、StdioServerInfo。
2. MCPClient（连接层）：抽象基类，定义connect()、call_tool()、aclose()三个核心接口，具体实现为StdioClient、SSEClient、StreamableHttpClient。
3. MCPTool/MCPToolset（工具层）：分别将单个 MCP 工具或整台 MCP 服务器上的工具集合包装成 Haystack 可用的Tool/Toolset。

配置对象通过create_client()工厂方法返回对应的客户端，因此用户只需要面向ServerInfo编程，无需关心底层传输细节：

server_info = StreamableHttpServerInfo(url="http://localhost:8000/mcp") client = server_info.create_client() # -> StreamableHttpClient tools = client.connect() # -> list[types.Tool]

三、ServerInfo 配置对象：三种传输的完整参数说明

3.1 StdioServerInfo（本地进程）

server_info = StdioServerInfo( command="uvx", args=["mcp-server-time", "--local-timezone=Europe/Berlin"], env={ "WORKSPACE_PATH": "/path/to/workspace", # 普通字符串原样传递 "API_KEY": Secret.from_env_var("API_KEY"), # Secret 对象安全处理 }, )

参数说明：

关于Secret：文档明确说明，Secret对象会被安全地序列化与反序列化、不会暴露敏感值，而普通字符串会原样保留。因此凡是 API Key、Token 这类敏感数据都应使用Secret.from_env_var("API_KEY")这样的方式注入。

3.2 StreamableHttpServerInfo（推荐，远程 HTTP）

server_info = StreamableHttpServerInfo( url="https://my-mcp-server.com", # streamable HTTP 端点 token=Secret.from_env_var("API_KEY"), # 可选，自动生成 Authorization: Bearer <token> 头 headers={"X-API-Key": Secret.from_env_var("API_KEY")}, # 自定义头，优先级高于 token timeout=30, # 连接超时（秒） )

参数说明：

3.3 SSEServerInfo（已弃用）

server_info = SSEServerInfo( url="https://my-mcp-server.com/sse", # 含 /sse 端点 token=Secret.from_env_var("API_KEY"), )

参数说明：

注意：SSE 传输已被 MCP 规范弃用，官方建议新集成一律使用 Streamable HTTP。SSEServerInfo仅用于连接尚不支持 streamable HTTP 的旧服务器。

3.4 统一序列化接口

MCPServerInfo抽象基类（参考文档）还定义了：

• create_client() -> MCPClient：根据配置创建对应的 MCP 客户端；
• to_dict() -> dict[str, Any]：序列化为字典；
• from_dict(data) -> MCPServerInfo：从字典反序列化，返回对应类型的配置对象。

这保证了包含ServerInfo的 Tool / Toolset 可以被 Haystack 的序列化机制完整保存与重建。

四、MCPClient 连接层：三种传输客户端与重试策略

MCPClient是抽象基类（继承ABC），定义了所有客户端共用的接口与能力：

三个具体实现：

• StdioClient(command, args=None, env=None, max_retries=3, base_delay=1.0, max_delay=30.0)：以 stdio 传输连接本地进程。
• SSEClient(server_info, max_retries=3, base_delay=1.0, max_delay=30.0)：以 SSE 传输连接远程服务器。注意：若同时提供自定义 headers 和 token，自定义 headers 优先。
• StreamableHttpClient(server_info, max_retries=3, base_delay=1.0, max_delay=30.0)：以 streamable HTTP 传输连接远程服务器，同样遵循"headers 优先于 token"规则。

三者都内置了指数退避重连机制：max_retries为最大重连次数（默认 3），base_delay为退避基础延迟（默认 1.0 秒），max_delay为退避上限（默认 30.0 秒）。

4.1 AsyncExecutor：同步上下文中的事件循环执行器

由于 MCP 客户端基于asyncio，而 Agent 的某些调用路径是同步的，集成内部提供了一个线程安全的AsyncExecutor来桥接：

• get_instance() -> AsyncExecutor：获取（或创建）全局单例执行器；
• run(coro, timeout=None) -> Any：在专用事件循环中运行协程，可选超时；超过超时抛TimeoutError；
• run_background(coro_factory, timeout=None) -> tuple[Future, Event]：不阻塞调用线程地调度协程运行。工厂函数接收一个asyncio.Event（stop_event），可用于协作式关闭协程；方法返回(future, stop_event)，前者用于观察完成或失败，后者用于发出终止信号；
• get_loop() -> asyncio.AbstractEventLoop：获取事件循环；
• shutdown(timeout=2)：关闭后台事件循环与线程，默认 2 秒超时。

从实现看，AsyncExecutor.__init__会初始化一个专用事件循环，保证同步上下文中的协程调度不会污染调用方的 asyncio 状态。

五、错误体系：MCPError 及其派生异常

MCP 集成定义了层级化的错误类型（源码中的ToolInvocationError定义见 haystack/tools/errors.py）：

注意MCPInvocationError的基类是 Haystack 的ToolInvocationError而非MCPError——这是为了让工具调用错误与 Haystack 核心工具体系保持一致，便于上层 Agent 统一捕获处理。

六、MCPTool：把单个 MCP 工具包装成 Haystack Tool

MCPTool继承自 Haystack 的Tool基类（见 haystack/tools/tool.py），代表 MCP 服务器上的单个工具。它使用官方 MCP SDK 处理协议，同时兼容 Haystack 工具生态。

6.1 完整初始化参数

tool = MCPTool( name="multiply", # 必填：工具名 server_info=StreamableHttpServerInfo(url="http://localhost:8000/mcp"), # 必填：服务器配置 description=None, # 可选：自定义描述，None 时使用服务器描述 connection_timeout=30, # 连接服务器超时（秒） invocation_timeout=30, # 工具调用默认超时（秒） eager_connect=False, # True 则初始化时立即连接；False（默认）延迟到 warm_up 或首次使用 outputs_to_string=None, # 定义工具输出如何转成字符串 inputs_from_state=None, # 定义 State 键到工具参数的映射 outputs_to_state=None, # 定义工具输出到 State 键的映射 )

6.2 三种传输方式的完整用法

Streamable HTTP（推荐）：

import json from haystack_integrations.tools.mcp import MCPTool, StreamableHttpServerInfo tool = MCPTool( name="multiply", server_info=StreamableHttpServerInfo(url="http://localhost:8000/mcp"), ) result_json = tool.invoke(a=5, b=3) result = json.loads(result_json) # 用 json.loads 解析为字典

SSE（已弃用）：

import json from haystack_integrations.tools.mcp import MCPTool, SSEServerInfo tool = MCPTool( name="add", server_info=SSEServerInfo(url="http://localhost:8000/sse"), ) result_json = tool.invoke(a=5, b=3) result = json.loads(result_json)

StdIO（本地进程）：

import json from haystack_integrations.tools.mcp import MCPTool, StdioServerInfo tool = MCPTool( name="get_current_time", server_info=StdioServerInfo(command="python", args=["path/to/server.py"]), ) result_json = tool.invoke(timezone="America/New_York") result = json.loads(result_json)

6.3 响应处理规则

根据参考文档，MCPTool的响应处理遵循以下约定：

• 支持文本（TextContent）与图片（ImageContent）内容，统一以 JSON 字符串形式返回；
• JSON 中包含 MCP 服务器返回的结构化响应；
• 使用json.loads()即可把响应解析为字典。

6.4 生命周期：懒连接与显式预热

• eager_connect=False（默认）：连接延迟到warm_up()或首次使用工具时，谁先发生谁触发连接；
• eager_connect=True：初始化时立即连接服务器；
• warm_up()：当eager_connect关闭时，调用它来连接服务器并抓取工具 schema。

在 Haystack 中，Agent.warm_up()与Pipeline.warm_up()会自动调用工具的warm_up()，因此默认配置下无需手动管理连接时机。

6.5 异步调用

result = await tool.ainvoke(**kwargs)

ainvoke(**kwargs) -> str | dict[str, Any]提供异步调用能力，返回 JSON 字符串或字典；当配置了outputs_to_state需要更新 State 时返回字典。失败抛MCPInvocationError，超时抛TimeoutError。

6.6 序列化：to_dict / from_dict

• to_dict() -> dict：返回{"type": 全限定类名, "data": {参数}}格式，完整保留服务器连接参数、超时设置与 state-mapping 配置，以便重建工具；注意活动中的连接不会被序列化。
• from_dict(data) -> Tool：从字典重建MCPTool，包括重建server_info与 state-mapping 参数，并在初始化过程中重新建立到 MCP 服务器的连接；连接失败会抛异常。

6.7 close()

close()同步关闭工具，释放底层 MCP 客户端资源。

七、State-Mapping：让 MCP 工具与 Agent State 联动

MCPTool支持三个与 Agent State 联动的参数（在 haystack/tools/tool.py 的Tool基类中有完整的校验逻辑：inputs_from_state的参数名、outputs_to_state的 source 键都会在构造时校验，非法配置直接抛ValueError/TypeError）。

7.1 inputs_from_state：从 State 注入工具参数

tool = MCPTool( name="git_status", server_info=..., inputs_from_state={"repository": "repo_path"}, # 将 State 中的 "repository" 键映射为工具的 "repo_path" 参数 )

7.2 outputs_to_state：把工具输出写回 State

tool = MCPTool( name="git_diff", server_info=..., outputs_to_state={ "diff_result": {"source": "diff", "handler": custom_handler}, # 指定 source 时，只有 "diff" 这个输出键会交给 handler； # 不指定 source 时，整个工具结果都会交给 handler }, )

两种形态：

• 带source：{"documents": {"source": "docs", "handler": custom_handler}}—— 只把输出中的docs键交给 handler；
• 不带source：{"documents": {"handler": custom_handler}}—— 整个工具结果交给 handler。

7.3 outputs_to_string：自定义输出转字符串

tool = MCPTool( name="git_diff", server_info=..., outputs_to_string={"source": "diff", "handler": format_diff}, # 指定 source 时只把该输出键交给 handler；省略 source 时整个工具结果交给 handler )

八、MCPToolset：自动发现并批量加载 MCP 服务器工具

MCPToolset继承自Toolset（见 haystack/tools/toolset.py），动态发现并加载 MCP 服务器上的所有工具，同时支持远程（Streamable HTTP、SSE）与本地（StdIO）两类连接。

8.1 初始化参数

toolset = MCPToolset( server_info=server_info, # 必填：服务器连接信息 tool_names=None, # 可选：只加载指定名称的工具 connection_timeout=30.0, # 连接超时（秒） invocation_timeout=30.0, # 工具调用默认超时（秒） eager_connect=False, # True 初始化即连接；False 延迟到 warm_up inputs_from_state=None, # 按工具名配置 State -> 参数映射 outputs_to_state=None, # 按工具名配置输出 -> State 映射 outputs_to_string=None, # 按工具名配置输出转字符串 )

关键点：

• tool_names过滤：若不指定，则加载服务器上的全部工具。文档特别提醒：如果工具很多（20~30 个以上），可能压垮 LLM 的工具解析逻辑，建议按需过滤。
• 按工具名的 state 配置：inputs_from_state/outputs_to_state/outputs_to_string都以"工具名 -> 配置"的嵌套字典形式提供。工具名若与服务器上可用工具不匹配，会记录警告日志。
• 参数名校验：Haystack >= 2.22.0 时，inputs_from_state中的参数名会在构造时校验，非法参数直接抛ValueError；更早版本中非法参数会在运行时才失败。
• 错误：指定的工具名在服务器上找不到时抛MCPToolNotFoundError。

8.2 与 Pipeline 集成的完整示例

# 前置条件： # 1. pip install uvx mcp-server-time 安装所需 MCP 服务器与工具 # 2. export OPENAI_API_KEY="your-api-key" 配置 OpenAI API Key from haystack import Pipeline from haystack.components.agents import Agent from haystack.components.generators.chat import OpenAIChatGenerator from haystack.dataclasses import ChatMessage from haystack_integrations.tools.mcp import MCPToolset, StdioServerInfo # 为时间服务创建服务器配置（远程服务器也可用 SSEServerInfo / StreamableHttpServerInfo） server_info = StdioServerInfo(command="uvx", args=["mcp-server-time", "--local-timezone=Europe/Berlin"]) # 创建 toolset——自动发现所有可用工具；也可用 tool_names 指定只包含部分工具 mcp_toolset = MCPToolset( server_info=server_info, tool_names=["get_current_time"], # 只包含 get_current_time 这一个工具 ) # 创建 Pipeline：Agent 拥有工具调用循环，把 toolset 交给 chat generator， # 执行任何请求的工具调用，直到产出最终答案 pipeline = Pipeline() pipeline.add_component( "agent", Agent(chat_generator=OpenAIChatGenerator(model="gpt-4o-mini"), tools=mcp_toolset), ) user_input = "What is the time in New York? Be brief." user_input_msg = ChatMessage.from_user(text=user_input) result = pipeline.run({"agent": {"messages": [user_input_msg]}}) print(result["agent"]["messages"][-1].text)

8.3 通过 Streamable HTTP 连接远程服务器

from haystack_integrations.tools.mcp import MCPToolset, StreamableHttpServerInfo toolset = MCPToolset( server_info=StreamableHttpServerInfo(url="http://localhost:8000/mcp"), tool_names=["multiply"], # 可选：只包含特定工具 ) # 之后用法与上面的 Pipeline 示例一致

8.4 按工具配置 State 联动（Agent 集成）

from haystack_integrations.tools.mcp import MCPToolset, StdioServerInfo toolset = MCPToolset( server_info=StdioServerInfo(command="uvx", args=["mcp-server-git"]), tool_names=["git_status", "git_diff", "git_log"], # 为每个工具映射 State 键到工具参数：把 State 的 "repository" 映射到 "repo_path" inputs_from_state={ "git_status": {"repository": "repo_path"}, "git_diff": {"repository": "repo_path"}, "git_log": {"repository": "repo_path"}, }, # 把工具输出映射到 State 键 outputs_to_state={ "git_status": {"status_result": {"source": "status"}}, # 从输出中提取 "status" "git_diff": {"diff_result": {}}, # 使用完整输出走默认处理 }, )

8.5 SSE 示例（已弃用）

from haystack_integrations.tools.mcp import MCPToolset, SSEServerInfo sse_toolset = MCPToolset( server_info=SSEServerInfo(url="http://some-remote-server.com:8000/sse"), tool_names=["add", "subtract"], # 只包含指定工具 )

8.6 warm_up、序列化与关闭

• warm_up()：当eager_connect=False时，调用此方法连接并加载工具。Agent.warm_up()与Pipeline.warm_up()会自动调用；你也可以在正式调用前手动调用，确保所有工具 schema 可用。
• to_dict()/from_dict()：序列化 / 反序列化整个 toolset。
• close()：安全关闭底层 MCP 客户端。

九、源码级原理：Tool / Toolset 基类如何支撑 MCP 集成

9.1 Tool 基类：invoke / invoke_async / warm_up

MCPTool继承的Tool数据类（haystack/tools/tool.py）提供了统一的调用与生命周期契约：

• invoke(**kwargs)：同步调用工具。若工具只有异步实现（function=None）则抛ToolInvocationError，提示只能通过invoke_async（配合Agent.run_async）调用；
• invoke_async(**kwargs)：若设置了async_function则直接 await；否则通过asyncio.to_thread把同步function派发到工作线程执行；
• warm_up()：默认空实现，留给子类覆盖以建立远程连接、加载模型等重量级初始化，且要求幂等（可能被多次调用）。MCPTool正是在此方法中完成"连接服务器 + 抓取工具 schema"；
• 构造时校验：function/async_function至少一个存在、参数必须是合法 JSON Schema、state-mapping 配置的类型与引用键合法（见 tool.py 的__post_init__）。

9.2 Toolset 基类：懒加载与动态工具发现

Toolset（haystack/tools/toolset.py）在其文档字符串中给出了MCPToolset 的参考实现模式：

class MCPToolset(Toolset): def warm_up(self) -> None: if self.mcp_connection is not None: return # 幂等保护：已连接则直接返回 self.mcp_connection = establish_connection(self.server_url) self.tools = self.mcp_connection.fetch_tools() # 动态加载工具

基类还提醒：由于warm_up()可能在每次运行前被调用，实现必须基于自身状态做幂等保护；动态加载的 Toolset 在序列化时应序列化端点描述符（URL、服务器信息）而不是动态加载的 Tool 实例，这正是MCPToolset.to_dict()保留server_info的原因。

9.3 测试佐证：懒加载在 Agent 中的真实行为

仓库测试 test/components/agents/test_agent.py 中有一个MockMCPToolset用例，验证了懒加载 Toolset 的完整链路：

• 初始化时 toolset 为空（或只含占位工具）；
• Agent.warm_up()触发warm_up()，将真实工具加载进self.tools；
• Agent 运行工具调用循环后，工具结果进入对话消息，Agent 依据exit_conditions产出最终文本答案。

该测试同时印证了 Agent 对"工具集动态变化"的设计：exit_conditions 不再在初始化时校验工具名，因为工具集可能是动态的（如SearchableToolset/MCPToolset）。

十、实践建议与注意事项

1. 新项目优先 Streamable HTTP：SSE 已被 MCP 规范弃用，仅用于连接尚不支持新传输的存量服务器；文档中的旧示例多使用 SSE，注意区分。
2. 善用tool_names过滤：不指定则加载服务器全部工具，工具过多（20~30+）会拖累 LLM 的工具解析。用 mcptoolset.mdx 中的建议，按需加载。
3. 敏感信息用Secret：Token、API Key 一律用Secret.from_env_var(...)包装，避免序列化时泄露；普通路径等非敏感值用普通字符串即可。
4. 连接时机交给 warm_up：默认eager_connect=False，由Pipeline.warm_up()/Agent.warm_up()自动处理连接，既避免初始化开销，也保证运行前 schema 就绪。
5. State 配置会被严格校验：Haystack >= 2.22.0 时，inputs_from_state的参数名在构造期校验，拼写错误会立即抛ValueError而非运行时失败；outputs_to_state的 source 键也会对照工具输出校验。
6. 响应是 JSON 字符串：invoke()返回的 JSON 字符串需用json.loads()解析；文本与图片内容均以 JSON 结构化返回。
7. 重试参数按网络环境调整：三种客户端默认max_retries=3、指数退避 1.0s → 30.0s，弱网或本地进程启动慢的场景可适当调大。

参考文件索引

• API 参考：docs-website/reference/integrations-api/mcp.md
• MCPTool 使用指南：docs-website/docs/tools/mcptool.mdx
• MCPToolset 使用指南：docs-website/docs/tools/mcptoolset.mdx
• Tool 基类实现：haystack/tools/tool.py
• Toolset 基类实现：haystack/tools/toolset.py
• 工具错误定义：haystack/tools/errors.py
• Agent 懒加载 Toolset 测试：test/components/agents/test_agent.py </output文章>

【免费下载链接】haystackOpen-source AI orchestration framework for building context-engineered, production-ready LLM applications. Design modular pipelines and agent workflows with explicit control over retrieval, routing, memory, and generation. Built for scalable agents, RAG, multimodal applications, semantic search, and conversational systems.项目地址: https://gitcode.com/GitHub_Trending/ha/haystack