企业官网建设流程全解析

Haystack 2.23 MCP 集成实战：用 MCPTool 与 MCPToolset 将外部工具生态接入 Agent 与 Pipeline

【免费下载链接】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

Haystack 通过mcp-haystack集成包把 Model Context Protocol（MCP）这一标准化协议引入其工具生态，让流水线与 Agent 可以直接调用任意 MCP 兼容服务器暴露的工具（时间服务、Git 操作、文件系统、数据库等）。本文将基于 version-2.23 的 MCP API 参考，结合 MCPTool 与 MCPToolset 使用指南以及仓库内Tool/Toolset/Agent的源码实现，完整讲解 ServerInfo 配置、三种传输方式（StdIO / Streamable HTTP / SSE）、错误模型、State 映射与序列化机制，并给出可直接运行的 Pipeline 示例。读完本文，你将掌握如何在 Haystack 2.23 中连接本地或远程 MCP 服务器、按需筛选工具，并把它们无缝集成到 Agent 的工具调用循环中。

MCP 集成在 Haystack 中的定位

MCP 是一个开放协议，它标准化了应用向 LLM 提供上下文的方式，官方文档常用"USB-C 之于设备连接、MCP 之于 AI 工具连接"来类比这一层标准化。在 Haystack 中，MCP 集成以独立包mcp-haystack的形式提供，导入路径为haystack_integrations.tools.mcp，与核心库（haystack/tools 目录下的Tool、Toolset）通过标准接口对接。

从当前仓库的 Agent 文档可以看到，Agent组件的tools参数明确支持MCPTool与MCPToolset类型（见 agent.mdx），并且 OpenAPI Connector 等旧式外部 API 连接方案在文档中被标注为 legacy，推荐用MCPTool作为接入外部工具的现代标准化方式（见 openapiconnector.mdx）。

集成层支持三种传输方式：

安装与前置条件

安装 MCP 集成包：

pip install mcp-haystack

以文档中最常用的时间服务为例，还需要安装 MCP server 与 uvx 工具，并准备 LLM 的 API Key：

# 安装 MCP server 与执行工具 pip install uvx mcp-server-time # 设置 LLM API Key（示例使用 OpenAI） export OPENAI_API_KEY="your-api-key"

第一层：ServerInfo 配置对象

所有 MCP 连接的第一步都是构造一个MCPServerInfo子类对象。MCPServerInfo是抽象基类，定义了三种 ServerInfo 的共同接口：

• create_client()：根据当前配置创建对应的MCPClient实例；
• to_dict()/from_dict()：将服务器连接参数序列化为字典（或从字典还原），这是工具整体序列化的基础。

仓库中Toolset基类的设计理念与之呼应：对于动态加载工具的 Toolset 子类，序列化时应保存"端点描述符"（如 server_info）而非已加载的 Tool 实例，以保证反序列化后能准确重建（见 toolset.py）。

StdioServerInfo：本地子进程

封装本地命令式 MCP 服务器的连接参数，核心字段为command（要运行的命令，如"uvx"、"python"、"node"）与args（传给命令的参数列表）。对于敏感的环境变量，可以使用Secret对象安全地处理序列化：

server_info = StdioServerInfo( command="uvx", args=["run", "my-mcp-server"], env={ "WORKSPACE_PATH": "/path/to/workspace", # Plain string "API_KEY": Secret.from_env_var("API_KEY"), # Secret object } )

Secret对象在序列化/反序列化时不会暴露其值，而普通字符串会原样保留——凡是涉及敏感数据的场景都应使用Secret。

StreamableHttpServerInfo：远程 HTTP

封装 streamable HTTP 传输的连接参数，字段包括：

• url：MCP 服务器的完整 URL（streamable HTTP 端点）；
• token：可选认证令牌，提供后会生成Authorization: Bearer <token>请求头；
• headers：自定义 HTTP 头，优先于token参数；
• timeout：连接超时秒数。

# 使用 Secret 处理令牌 server_info = StreamableHttpServerInfo( url="https://my-mcp-server.com", token=Secret.from_env_var("API_KEY"), ) # 自定义请求头（非标准认证场景） server_info = StreamableHttpServerInfo( url="https://my-mcp-server.com", headers={ "X-API-Key": Secret.from_env_var("API_KEY"), "X-Client-ID": "my-client-id", }, )

SSEServerInfo：旧式远程 HTTP（已弃用）

字段与 StreamableHttpServerInfo 基本一致，区别在于url指向/sse端点，且base_url参数已弃用（应使用url）。MCP 规范已弃用 SSE 传输，仅用于连接尚不支持 Streamable HTTP 的既有服务器：

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

第二层：MCPClient 传输客户端

MCPClient是抽象基类，定义所有传输客户端的共同接口：

• connect() -> list[types.Tool]：连接 MCP 服务器，返回服务器上可用工具列表；连接失败抛MCPConnectionError；
• call_tool(tool_name, tool_args) -> str：调用服务器上的工具，返回工具调用结果的 JSON 字符串；未连接时抛MCPConnectionError，调用失败抛MCPInvocationError；
• aclose()：关闭连接并释放资源，保证即使出错也会正确清理。

共有三个具体实现，均支持指数退避重连机制（max_retries最大重连次数、base_delay基础延迟秒数、max_delay最大延迟秒数）：

需要注意：SSE 与 Streamable HTTP 客户端在同时提供自定义 headers 和 token 时，自定义 headers 优先。

AsyncExecutor：同步上下文中的异步桥

MCP 客户端底层基于asyncio，而 Haystack 的Tool.invoke()是同步接口，二者通过AsyncExecutor衔接。它是一个线程安全的事件循环执行器，以全局单例形式存在：

• get_instance()：获取或创建全局单例；
• run(coro, timeout=None)：在事件循环中执行协程，可传超时秒数，超时抛TimeoutError；
• get_loop()：获取底层asyncio.AbstractEventLoop；
• run_background(coro_factory, timeout=None)：不阻塞调用线程地调度协程，返回(future, stop_event)二元组——协程工厂接收一个asyncio.Event用于协作式关闭，调用方可通过 stop_event 发出终止信号；
• shutdown(timeout=2)：关闭后台事件循环与线程。

第三层：错误模型

集成层定义了分层异常体系，根类是MCPError（继承Exception）：

这一分层设计让上层（Agent、Pipeline）能够精确地区分"连不上"、"没有这个工具"、"调用出错"三种故障，分别采取重连、改选工具、重试等不同策略。

MCPTool：单个 MCP 工具的 Haystack 封装

MCPTool继承自Tool，用官方 MCP SDK 处理协议层，同时保持与 Haystack 工具生态的兼容。每个MCPTool实例对应服务器上的一个具体工具。

初始化参数

MCPTool( name: str, # 工具名（必填） server_info: MCPServerInfo, # 服务器连接信息（必填） description: str | None = None, # 自定义描述，None 时用服务器描述 connection_timeout: int = 30, # 连接超时（秒） invocation_timeout: int = 30, # 单次调用默认超时（秒） eager_connect: bool = False, # True 时在初始化阶段就连服务器 outputs_to_string: dict[str, Any] | None = None, inputs_from_state: dict[str, str] | None = None, outputs_to_state: dict[str, dict[str, Any]] | None = None, )

连接时机（eager_connect）：默认为False，延迟到warm_up()或首次工具调用时（谁先到谁触发）才连接服务器；设为True则在初始化时立即连接。初始化或连接阶段可能抛出MCPConnectionError、MCPToolNotFoundError（服务器无工具或找不到指定工具）、TimeoutError（连接超时）。

State 映射参数（与Tool基类的能力对齐，见 tool.py）：

• outputs_to_string：定义工具输出如何转成字符串。提供source时只把指定的输出键发给 handler；省略source时把整个工具结果发给 handler。示例：{"source": "docs", "handler": my_custom_function}；
• inputs_from_state：把 Agent State 的键映射为工具参数名。示例：{"repository": "repo"}表示把 State 中的repository映射给工具的repo参数；
• outputs_to_state：定义工具输出如何写入 State 键及可选 handler。带source示例：{"documents": {"source": "docs", "handler": custom_handler}}；不带source示例：{"documents": {"handler": custom_handler}}（此时发送整个工具结果）。

在Tool基类中，这些配置在初始化时会被严格校验：outputs_to_state的 source 必须是字符串且必须存在于工具输出中、handler 必须可调用、inputs_from_state引用的参数必须真实存在，否则抛出TypeError或ValueError。

调用与生命周期

• invoke(**kwargs) -> str：同步调用工具，返回结果的JSON 字符串，需用json.loads()解析为字典；
• ainvoke(**kwargs) -> str | dict：异步调用。当配置了outputs_to_state时返回字典（以支持 State 更新），否则返回 JSON 字符串；失败抛MCPInvocationError或TimeoutError；
• warm_up()：在eager_connect=False时连接服务器并拉取工具 schema；
• to_dict()：序列化为{"type": 完整限定类名, "data": {参数}}格式，保留服务器连接参数、超时设置与 State 映射参数，但活动连接不保留；
• from_dict()：从字典重建MCPTool，会重建 server_info 与 State 映射参数，并在初始化时重新建立连接；
• close()：同步关闭工具。

三种传输的使用示例

Streamable HTTP：

import json from haystack_integrations.tools.mcp import MCPTool, StreamableHttpServerInfo # Create tool instance tool = MCPTool( name="multiply", server_info=StreamableHttpServerInfo(url="http://localhost:8000/mcp") ) # Use the tool and parse result result_json = tool.invoke(a=5, b=3) result = json.loads(result_json)

SSE（已弃用）：

import json from haystack.tools import MCPTool, SSEServerInfo # Create tool instance tool = MCPTool( name="add", server_info=SSEServerInfo(url="http://localhost:8000/sse") ) # Use the tool and parse result result_json = tool.invoke(a=5, b=3) result = json.loads(result_json)

StdIO：

import json from haystack.tools import MCPTool, StdioServerInfo # Create tool instance tool = MCPTool( name="get_current_time", server_info=StdioServerInfo(command="python", args=["path/to/server.py"]) ) # Use the tool and parse result result_json = tool.invoke(timezone="America/New_York") result = json.loads(result_json)

关于响应内容：MCP 服务器返回的TextContent、ImageContent与EmbeddedResource内容类型均受支持，统一以 JSON 字符串形式返回，解析后即为结构化的工具调用结果。

MCPToolset：动态发现与批量加载

MCPToolset继承自Toolset，连接到 MCP 服务器后动态发现并加载其全部工具，同时支持远程网络传输（Streamable HTTP、SSE）与本地进程传输（StdIO）。使用指南（mcptoolset.mdx）明确提醒：如果不传tool_names，服务器上所有工具都会被加载，当工具数量达到 20～30 个以上时可能压垮 LLM 的工具解析逻辑，因此强烈建议按需筛选。

初始化参数

MCPToolset( server_info: MCPServerInfo, # 服务器连接信息（必填） tool_names: list[str] | None = None, # 只加载指定名称的工具 connection_timeout: float = 30.0, # 连接超时（秒） invocation_timeout: float = 30.0, # 调用超时（秒） eager_connect: bool = False, # 初始化时即连接 inputs_from_state: dict[str, dict[str, str]] | None = None, # 按工具名的 State 输入映射 outputs_to_state: dict[str, dict[str, dict[str, Any]]] | None = None, # 按工具名的 State 输出映射 outputs_to_string: dict[str, dict[str, Any]] | None = None, # 按工具名的输出转字符串配置 )

注意tool_names、inputs_from_state、outputs_to_state、outputs_to_string的键都是工具名，它们应该与服务器上真实存在的工具名匹配：未匹配到的工具名会记录 warning；而inputs_from_state中参数名校验的时机与 Haystack 版本相关——Haystack >= 2.22.0会在初始化时校验参数名并抛ValueError，更早版本则只在运行时失败。若指定的工具名在服务器上不存在，抛MCPToolNotFoundError。

带 State 配置的完整示例

from haystack_integrations.tools.mcp import MCPToolset, StdioServerInfo # Create the toolset with per-tool state configuration # This enables tools to read from and write to the Agent's State toolset = MCPToolset( server_info=StdioServerInfo(command="uvx", args=["mcp-server-git"]), tool_names=["git_status", "git_diff", "git_log"], # Maps the state key "repository" to the tool parameter "repo_path" for each tool inputs_from_state={ "git_status": {"repository": "repo_path"}, "git_diff": {"repository": "repo_path"}, "git_log": {"repository": "repo_path"}, }, # Map tool outputs to state keys for each tool outputs_to_state={ "git_status": {"status_result": {"source": "status"}}, # Extract "status" from output "git_diff": {"diff_result": {}}, # use full output with default handling }, )

生命周期方法

• warm_up()：在eager_connect=False时连接并加载工具。文档明确说明该方法会被Agent.warm_up()与Pipeline.warm_up()自动调用，也可以在使用前手动调用，以便在不实际调用工具的情况下确保所有工具 schema 可用。这与 toolset.py 中Toolset.warm_up()的设计一致——基类文档建议动态加载型子类把"加载工具"放在warm_up()中，并以自身状态（如连接对象是否为None）做幂等保护，因为该方法可能在每次 run 前被多次调用；
• to_dict()/from_dict()：序列化/反序列化整个 Toolset（序列化描述符而非 Tool 实例是动态型 Toolset 的推荐策略，避免大对象序列化开销并保证重建准确性）；
• close()：安全关闭底层 MCP 客户端。

在 Pipeline 与 Agent 中集成

用 MCPToolset + Agent 搭建时间查询流水线

# Prerequisites: # 1. pip install uvx mcp-server-time # Install required MCP server and tools # 2. export OPENAI_API_KEY="your-api-key" # Set up your 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 # Create server info for the time service (can also use SSEServerInfo for remote servers) server_info = StdioServerInfo(command="uvx", args=["mcp-server-time", "--local-timezone=Europe/Berlin"]) # Create the toolset - this will automatically discover all available tools # You can optionally specify which tools to include mcp_toolset = MCPToolset( server_info=server_info, tool_names=["get_current_time"] # Only include the get_current_time tool ) # Create a pipeline with an Agent that owns the tool-calling loop. # The Agent passes the toolset to the chat generator, executes any requested # tool calls, and continues until a final answer is produced. pipeline = Pipeline() pipeline.add_component("agent", Agent(chat_generator=OpenAIChatGenerator(model="gpt-4o-mini"), tools=mcp_toolset)) # Run the pipeline with a user question 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)

用 Streamable HTTP 连接远程服务器

from haystack_integrations.tools.mcp import MCPToolset, StreamableHttpServerInfo # Create the toolset with streamable HTTP connection toolset = MCPToolset( server_info=StreamableHttpServerInfo(url="http://localhost:8000/mcp"), tool_names=["multiply"] # Optional: only include specific tools ) # Use the toolset as shown in the pipeline example above

使用 MCPTool 直连单个工具

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 MCPTool, StdioServerInfo time_tool = MCPTool( name="get_current_time", server_info=StdioServerInfo( command="uvx", args=["mcp-server-time", "--local-timezone=Europe/Berlin"], ), ) pipeline = Pipeline() pipeline.add_component( "agent", Agent( chat_generator=OpenAIChatGenerator(model="gpt-4o-mini"), tools=[time_tool], ), ) user_input = "What is the time in New York? Be brief." # can be any city user_input_msg = ChatMessage.from_user(text=user_input) result = pipeline.run({"agent": {"messages": [user_input_msg]}}) print(result["agent"]["last_message"].text) # The current time in New York is 1:57 PM.

直接使用 Agent 组件

from haystack.components.generators.chat import OpenAIChatGenerator from haystack.dataclasses import ChatMessage from haystack.components.agents import Agent from haystack_integrations.tools.mcp import MCPTool, StdioServerInfo time_tool = MCPTool( name="get_current_time", server_info=StdioServerInfo( command="uvx", args=["mcp-server-time", "--local-timezone=Europe/Berlin"], ), ) # Agent Setup agent = Agent( chat_generator=OpenAIChatGenerator(), tools=[time_tool], exit_conditions=["text"], ) # Run the Agent response = agent.run( messages=[ChatMessage.from_user("What is the time in New York? Be brief.")], ) # Output print(response["messages"][-1].text)

源码视角：MCP 集成如何嵌入 Haystack 工具生态

从当前仓库源码可以进一步理解 MCP 集成与核心库的衔接方式：

• Toolset 基类明确为 MCP 预留扩展点：toolset.py 的文档字符串明确指出Toolset的第二个用途就是"作为动态工具加载的基类——从 OpenAPI URL、MCP 服务器等外部来源加载工具"，并给出了MCPToolset式的warm_up()覆盖范式：以if self.mcp_connection is not None: return做幂等保护，再建立连接并执行self.tools = self.mcp_connection.fetch_tools()。同时Toolset实现了__iter__、__contains__、__len__、__getitem__集合接口，因此可以像普通工具列表一样被Agent与 Chat Generator 消费。

• Agent 对动态 Toolset 的运行期支持：agent.py 中，Agent 的tools参数接受"Tool 与/或 Toolset 对象列表，或单个 Toolset"；每次 run 前会调用warm_up_tools（warm_up()被设计为幂等，重复预热开销可忽略）；在工具循环的每个步骤还会重新扁平化工具集合，以便动态 Toolset（如SearchableToolset）能够暴露出运行期间新发现的工具，而不是冻结一份快照。

• State 映射能力来自 Tool 基类：tool.py 中Tool的outputs_to_string、inputs_from_state、outputs_to_state三个字段与MCPTool同名参数一一对应，且带完整的初始化校验（source 存在性、handler 可调用性、参数存在性）与序列化支持（handler 会转换为可序列化字符串表示再还原）。MCPTool直接继承这套能力，因此 MCP 工具天然具备读写 Agent State 的能力。

最佳实践与注意事项

1. 优先用 Streamable HTTP 与 StdIO：SSE 已被 MCP 规范弃用，仅在连接旧式 SSE-only 服务器时使用SSEServerInfo，并在服务器支持后迁移到StreamableHttpServerInfo。
2. 用tool_names限制工具数量：服务器上工具过多（20～30+）会显著增加 LLM 工具解析负担，按需筛选能提升 Agent 的工具选择质量与响应稳定性。
3. 敏感信息一律用Secret：env、token、headers中的密钥应通过Secret.from_env_var(...)传入，序列化时不会泄露明文。
4. 理解eager_connect与warm_up的配合：默认延迟连接模式下，Agent.warm_up()/Pipeline.warm_up()会自动完成连接与 schema 加载；如需在初始化时立即校验服务器可用性，再开启eager_connect=True。
5. 注意版本相关的参数校验差异：inputs_from_state的参数名校验仅在 Haystack >= 2.22.0 生效，旧版本会在运行时才暴露问题，升级后请重新验证既有配置。
6. 响应是 JSON 字符串：invoke/ainvoke返回的是 JSON 字符串（配置outputs_to_state时ainvoke返回字典），记得用json.loads()解析后再消费。
7. 序列化不保留活动连接：MCPTool/MCPToolset反序列化后会重新建立连接，因此在 pipeline 反序列化到实际运行之间会有一段连接建立过程，属于预期行为。

总结

mcp-haystack为 Haystack 2.23 提供了一条完整的 MCP 接入链路：从StdioServerInfo/StreamableHttpServerInfo/SSEServerInfo描述连接，到StdioClient/StreamableHttpClient/SSEClient承载传输（配合AsyncExecutor桥接同步与异步、指数退避保证重连稳健），再到MCPError分层异常提供可区分的故障语义，最后通过MCPTool（单工具）与MCPToolset（批量动态发现）接入Agent的工具调用循环。结合 State 映射参数，MCP 工具不仅能被调用，还能读写 Agent 运行状态，实现"工具即服务、服务即生态"的编排方式。无论你的 MCP 服务器是本地进程还是远程 HTTP 服务，都可以用本文给出的模板在数分钟内接入 Haystack Pipeline。

【免费下载链接】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