Haystack 2.22 OpenAPI 连接器详解：用 OpenAPIConnector 与 OpenAPIServiceConnector 将任意 REST 服务接入 Pipeline

Haystack 2.22 OpenAPI 连接器详解：用 OpenAPIConnector 与 OpenAPIServiceConnector 将任意 REST 服务接入 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 的Connectors组件模块是连接外部服务与编排管道的桥梁，其中OpenAPIConnector与OpenAPIServiceConnector专门用于对接遵循 OpenAPI（原 Swagger）规范的 REST 服务。本文以 version-2.22 的 connectors_api.md 为骨架，结合 v2.22 组件指南 与仓库中的发布记录，完整讲解这两个组件的初始化参数、run()调用方式、认证支持、序列化机制与完整流水线示例，并梳理其后续的弃用与迁移路径。

一、Connectors 模块定位：管道的“外部世界接口”

在 Haystack 中，Pipeline 组件负责检索、路由、记忆与生成，而Connectors是一类特殊的集成组件——它们把管道连接到外部服务商提供的 API。在 v2.22 组件索引 中可以看到，该模块包含 GitHub 系列组件（文件编辑、Issue 查看/评论、PR 创建、仓库 Fork 等）、JinaReader、Langfuse、Weave 追踪组件，以及两个 OpenAPI 相关的组件：

组件	描述
OpenAPIConnector	使用显式输入参数，在 Haystack 生态与 OpenAPI 服务之间充当接口
OpenAPIServiceConnector	作为 Haystack 生态与 OpenAPI 服务之间的接口（面向 LLM 函数调用）

两者的共同点是不需要为每个外部服务编写专用代码：只要该服务提供 OpenAPI 规范文件，就能动态解析规范、处理认证与参数、调用对应端点。区别在于调用方式——前者由用户/上游组件显式传入operation_id与arguments；后者则从ChatMessage中解析由 LLM 生成的函数调用载荷（tool call）来触发服务方法。这正是参考文档connectors_api.md中两个模块openapi与openapi_service的设计意图。

二、OpenAPIConnector：显式参数驱动的 REST 端点直调

OpenAPIConnector使管道可以直接调用 OpenAPI 规范中定义的 REST 端点。它动态解读 API 规范，并提供执行 API 操作的接口。它通常通过两种方式被调用：从 Haystack 管道的run()方法传入输入参数，或由管道中的其他组件向其传递输入参数。

参考文档原文：connectors_api.md

2.1 依赖安装

使用OpenAPIConnector前需要先安装可选的openapi-llm依赖（它提供了 OpenAPI 规范解析与客户端生成能力）：

pip install openapi-llm

2.2 构造参数（__init__）

def __init__( openapi_spec: str, credentials: Secret | None = None, service_kwargs: dict[str, Any] | None = None, )

各参数说明：

• openapi_spec（必填）：OpenAPI 规范的来源，支持三种形式——URL（如https://bit.ly/serperdev_openapi）、本地文件路径、或规范的原始字符串。
• credentials（可选）：封装在Secret中的 API Key 或服务凭证，用于对目标服务进行认证。
• service_kwargs（可选）：传递给OpenAPIClient.from_spec()的额外关键字参数，例如自定义的config_factory或其他客户端配置选项。

参考文档特别强调了两点：parameters参数（即run()的入参）对该组件是必需的；service_kwargs可选，用于向 OpenAPIClient 传递附加选项。

2.3 调用接口（run）

@component.output_types(response=dict[str, Any]) def run(operation_id: str, arguments: dict[str, Any] | None = None) -> dict[str, Any]

• operation_id（必填）：要调用的端点在 OpenAPI 规范中声明的operationId。
• arguments（可选）：端点的参数，包括 query、path 或 body 参数。

返回值为一个包含服务响应的字典，键为response，值为 REST 端点返回的 JSON。参考文档给出的输出结构为：

{ "response": { // 这里是 REST 端点返回的 JSON } }

2.4 独立使用示例

参考文档中的最小示例（以 Serper 搜索引擎为例，将“Who was Nikola Tesla?”作为查询）：

from haystack.utils import Secret from haystack.components.connectors.openapi import OpenAPIConnector connector = OpenAPIConnector( openapi_spec="https://bit.ly/serperdev_openapi", credentials=Secret.from_env_var("SERPERDEV_API_KEY"), service_kwargs={"config_factory": my_custom_config_factory} ) response = connector.run( operation_id="search", arguments={"q": "Who was Nikola Tesla?"} )

要点：

• Secret.from_env_var("SERPERDEV_API_KEY")从环境变量安全地读取密钥，避免在代码中硬编码凭证；
• operation_id="search"对应 Serper OpenAPI 规范中的搜索操作；
• arguments={"q": ...}中的q是 Serper API 的查询参数。

2.5 集成进 Pipeline

在 v2.22 的 openapiconnector.mdx 中给出了完整的管道集成方式。该组件在管道中的典型位置是“任意位置，但需位于能为其 run 参数提供输入的组件之后”：

from haystack import Pipeline from haystack.components.connectors.openapi import OpenAPIConnector from haystack.dataclasses.chat_message import ChatMessage from haystack.utils import Secret # 初始化 OpenAPIConnector connector = OpenAPIConnector( openapi_spec="https://bit.ly/serperdev_openapi", credentials=Secret.from_env_var("SERPERDEV_API_KEY"), ) # 创建用户 ChatMessage user_message = ChatMessage.from_user(text="Who was Nikola Tesla?") # 定义管道 pipeline = Pipeline() pipeline.add_component("openapi_connector", connector) # 运行管道 response = pipeline.run( data={ "openapi_connector": { "operation_id": "search", "arguments": {"q": user_message.text}, }, }, ) # 从响应中提取答案 answer = response.get("openapi_connector", {}).get("response", {}) print(answer)

组件速览（来自 v2.22 组件文档）：

项目	说明
管道中最常见位置	任意位置，位于能提供其 run 参数的组件之后
必填 init 变量	openapi_spec：服务 OpenAPI 规范（URL / 文件路径 / 原始字符串）
必填 run 变量	operation_id：要调用的规范中的 operationId
输出变量	response：REST 服务响应

三、OpenAPIServiceConnector：面向 LLM 函数调用的服务连接器

OpenAPIServiceConnector将 Haystack 框架连接到 OpenAPI 服务，使其能够按服务的 OpenAPI 规范调用其中定义的操作。它的工作方式与OpenAPIConnector有本质区别：它与ChatMessage数据类集成，消息中的载荷（payload）被用于确定要调用的方法以及要传递的参数。

参考文档原文：connectors_api.md

3.1 工作原理

参考文档明确了其调用协议：

• 消息载荷应为OpenAI JSON 格式的函数调用字符串，包含要调用的方法名和参数；
• 组件解析出方法名与参数后，在 OpenAPI 服务上执行对应方法；
• 服务返回的响应被封装为一个ChatMessage返回。

组件本身没有必填的 init 参数，仅提供一个可选的ssl_verify配置（见下）。在使用前，用户通常需要借助OpenAPIServiceToFunctions组件来解析服务端点参数——该组件把 OpenAPI 规范转换成 OpenAI 函数调用机制可理解的格式。

3.2 依赖安装

pip install openapi3

3.3 构造参数（__init__）

def __init__(ssl_verify: bool | str | None = None)

• ssl_verify：决定请求是否启用 SSL 校验。传bool控制开关；若传入字符串，则该字符串会被用作CA 证书路径。默认为None（使用默认校验行为）。

3.4 调用接口（run）

@component.output_types(service_response=dict[str, Any]) def run( messages: list[ChatMessage], service_openapi_spec: dict[str, Any], service_credentials: dict | str | None = None, ) -> dict[str, list[ChatMessage]]

• messages（必填）：ChatMessage列表，其中最后一条消息应包含 tool calls（函数调用）。组件解析该消息来执行服务方法。
• service_openapi_spec（必填）：目标服务的 OpenAPI JSON 规范对象，所有$ref引用必须已经解析完成（即传入的是已展开引用的完整规范）。
• service_credentials（可选）：与服务进行认证的凭证。参考文档明确指出，目前仅支持 OpenAPI 规范 v3 中的两类安全方案：
  1. http—— 用于 Basic、Bearer 及其他 HTTP 认证方案；
  2. apiKey—— 用于 API Key 与 cookie 认证。

异常：若最后一条消息不是来自 assistant（即不是ChatRole.ASSISTANT），或不包含 tool calls，则抛出ValueError。

返回值：字典包含键service_response，其值为ChatMessage列表，每个消息对应一次函数调用；若用户指定了多次函数调用请求，则会有多个响应。响应的content属性中保存 JSON 字符串形式的结果。

3.5 独立使用示例

参考文档给出的独立示例直接构造函数调用载荷（注意：真实场景中该载荷通常由OpenAIChatGenerator生成）：

import json import requests from haystack.components.connectors import OpenAPIServiceConnector from haystack.dataclasses import ChatMessage fc_payload = [{'function': {'arguments': '{"q": "Why was Sam Altman ousted from OpenAI?"}', 'name': 'search'}, 'id': 'call_PmEBYvZ7mGrQP5PUASA5m9wO', 'type': 'function'}] serper_token = <your_serper_dev_token> serperdev_openapi_spec = json.loads(requests.get("https://bit.ly/serper_dev_spec").text) service_connector = OpenAPIServiceConnector() result = service_connector.run( messages=[ChatMessage.from_assistant(json.dumps(fc_payload))], service_openapi_spec=serperdev_openapi_spec, service_credentials=serper_token, ) print(result)

输出示例（截取自参考文档）：

>> {'service_response': [ChatMessage(_role=<ChatRole.ASSISTANT: 'assistant'>, _content=[TextContent(text= >> '{"searchParameters": {"q": "Why was Sam Altman ousted from OpenAI?", >> "type": "search", "engine": "google"}, "answerBox": {"snippet": "Concerns over AI safety and OpenAI's role >> in protecting were at the center of Altman's brief ouster from the company."...

可以看到：

• 函数调用载荷采用 OpenAI 函数调用 JSON 格式：function.name是要调用的方法名（这里是search），function.arguments是 JSON 字符串形式的参数；
• ChatMessage.from_assistant(json.dumps(fc_payload))将载荷封装为 assistant 消息——这是run()校验的前提（最后一条消息必须是 assistant 消息）；
• 服务响应以 JSON 字符串形式包在ChatMessage的content中返回。

3.6 在管道中与 LLM 协同（完整示例）

v2.22 的 openapiserviceconnector.mdx 给出了完整流水线示例。其典型链路是：OpenAPIServiceToFunctions（将 OpenAPI 规范转换为函数调用 schema）→OpenAIChatGenerator（LLM 决策调用哪个函数）→OpenAPIServiceConnector（实际调用服务）→ 另一 LLM 综合结果作答。

import json import requests from typing import Dict, Any, List from haystack import Pipeline from haystack.components.generators.utils import print_streaming_chunk from haystack.components.converters import OpenAPIServiceToFunctions, OutputAdapter from haystack.components.generators.chat import OpenAIChatGenerator from haystack.components.connectors import OpenAPIServiceConnector from haystack.components.fetchers import LinkContentFetcher from haystack.dataclasses import ChatMessage, ByteStream from haystack.utils import Secret def prepare_fc_params(openai_functions_schema: Dict[str, Any]) -> Dict[str, Any]: return { "tools": [{ "type": "function", "function": openai_functions_schema }], "tool_choice": { "type": "function", "function": {"name": openai_functions_schema["name"]} } } system_prompt = requests.get("https://bit.ly/serper_dev_system_prompt").text serper_spec = requests.get("https://bit.ly/serper_dev_spec").text pipe = Pipeline() pipe.add_component("spec_to_functions", OpenAPIServiceToFunctions()) pipe.add_component("functions_llm", OpenAIChatGenerator(api_key=Secret.from_token(llm_api_key), model="gpt-3.5-turbo-0613")) pipe.add_component("openapi_container", OpenAPIServiceConnector()) pipe.add_component("a1", OutputAdapter("{{functions[0] | prepare_fc}}", Dict[str, Any], {"prepare_fc": prepare_fc_params})) pipe.add_component("a2", OutputAdapter("{{specs[0]}}", Dict[str, Any])) pipe.add_component("a3", OutputAdapter("{{system_message + service_response}}", List[ChatMessage])) pipe.add_component("llm", OpenAIChatGenerator(api_key=Secret.from_token(llm_api_key), model="gpt-4-1106-preview", streaming_callback=print_streaming_chunk)) pipe.connect("spec_to_functions.functions", "a1.functions") pipe.connect("spec_to_functions.openapi_specs", "a2.specs") pipe.connect("a1", "functions_llm.generation_kwargs") pipe.connect("functions_llm.replies", "openapi_container.messages") pipe.connect("a2", "openapi_container.service_openapi_spec") pipe.connect("openapi_container.service_response", "a3.service_response") pipe.connect("a3", "llm.messages") user_prompt = "Why was Sam Altman ousted from OpenAI?" result = pipe.run( data={ "functions_llm": {"messages": [ChatMessage.from_system("Only do function calling"), ChatMessage.from_user(user_prompt)]}, "openapi_container": {"service_credentials": serper_dev_key}, "spec_to_functions": {"sources": [ByteStream.from_string(serper_spec)]}, "a3": {"system_message": [ChatMessage.from_system(system_prompt)]}, } )

管道数据流的关键点：

• spec_to_functions.functions→ 经a1包装为 OpenAItools/tool_choice结构 → 注入functions_llm.generation_kwargs；
• spec_to_functions.openapi_specs→ 经a2直接传给openapi_container.service_openapi_spec；
• functions_llm.replies（LLM 生成的函数调用）→openapi_container.messages；
• openapi_container.service_response→ 经a3与系统提示拼接 → 交给最终llm生成自然语言回答。

组件速览（来自 v2.22 组件文档）：

项目	说明
管道中最常见位置	灵活
必填 run 变量	messages（末条需含函数调用载荷）；service_openapi_spec（YAML/JSON，ref 需已解析）；service_credentials（支持 http 与 apiKey 两类安全方案）
输出变量	service_response：ChatMessage列表，每个消息对应一次函数调用；多次调用则多个响应

注意：示例使用了 Serper 与 OpenAI 的 API Key，运行时需自行准备；Serper 仅是示例，任意符合 OpenAPI 规范的服务均可接入。

四、序列化机制：to_dict 与 from_dict

两个组件都实现了 Haystack 组件的标准序列化协议，便于将组件配置保存为 YAML/JSON 并在之后重建：

• OpenAPIConnector.to_dict() -> dict[str, Any]：将组件序列化为字典。
• OpenAPIConnector.from_dict(cls, data: dict[str, Any]) -> "OpenAPIConnector"：从字典反序列化重建组件。
• OpenAPIServiceConnector.to_dict() -> dict[str, Any]：序列化为字典。
• OpenAPIServiceConnector.from_dict(cls, data: dict[str, Any]) -> "OpenAPIServiceConnector"：从字典反序列化。

这意味着包含这两个组件的管道可以借助 Haystack 的Pipeline.dumps()/loads()机制整体序列化，实现组件配置的版本化保存与跨环境复用。需要留意的是，凭证类信息（如Secret）在序列化时通常以引用（如环境变量名）形式保存，而非明文密钥。

五、演进与迁移：从核心组件到 openapi-haystack 集成包

从仓库的发布记录（releasenotes/notes）可以还原这两个组件在 Haystack 中的完整生命周期：

1. 引入（add-openapi-connector-ebaa97cfa95b6c3e.yaml）：引入OpenAPIConnector，支持直接调用 OpenAPI 规范中定义的 REST 端点，无需 LLM 生成载荷，调用参数需显式传入。
2. 弃用（deprecate-openapi-components-e5f0f7470218fcc4.yaml）：OpenAPIConnector、OpenAPIServiceConnector与OpenAPIServiceToFunctions被标记弃用，计划在 Haystack 3.0 中移除，并迁往独立的openapi-haystack集成包。
3. 移除（remove-openapi-components-2f1de8f6c1b2787f.yaml）：这三个组件从 Haystack 核心中移除，迁入专门的openapi-haystack集成包。

迁移方式：安装openapi-haystack包并更新导入路径。

# 之前（Haystack 2.x 核心导入） from haystack.components.connectors import OpenAPIConnector, OpenAPIServiceConnector from haystack.components.converters import OpenAPIServiceToFunctions # 之后（openapi-haystack 集成包） from haystack_integrations.components.connectors.openapi import OpenAPIConnector, OpenAPIServiceConnector from haystack_integrations.components.converters.openapi import OpenAPIServiceToFunctions

官方发布记录同时指出：这些组件是连接 Haystack 与外部 API 的传统方式。对大多数用例，官方建议改用MCPTool——它是给管道与 Agent 提供外部工具和服务访问的现代化、标准化方式。这意味着：如果你现在才开始设计新的管道，应优先评估MCPTool（基于 Model Context Protocol）而非沿用 OpenAPI 连接器；若维护的是基于 Haystack 2.x 的存量管道，则可按上述路径迁移到openapi-haystack包继续使用。

六、选型建议与总结

维度	OpenAPIConnector	OpenAPIServiceConnector
调用驱动	显式参数：operation_id+arguments	LLM 函数调用载荷（tool call），从ChatMessage解析
适用场景	管道逻辑明确知道要调用哪个端点	需要 LLM 自主决策调用哪个服务方法的 Agent 化场景
必填依赖	pip install openapi-llm	pip install openapi3
认证支持	通过credentials（Secret）	service_credentials，支持http与apiKey两类 OpenAPI v3 安全方案
输出	{"response": <REST JSON>}	{"service_response": [ChatMessage, ...]}
是否需 ref 解析	规范交由客户端解析	传入的规范必须已解析全部$ref

参考文档与相关源码的进一步阅读入口：

• API 参考：version-2.22 connectors_api.md（openapi与openapi_service两个模块的完整签名）
• 组件指南：openapiconnector.mdx、openapiserviceconnector.mdx
• 发布记录：引入、弃用、移除

简而言之：OpenAPIConnector适合“确定性地调用”已知端点的场景，OpenAPIServiceConnector则适合“让 LLM 决定调用哪个函数”的智能体链路。二者都是把任意 OpenAPI 规范服务接入 Haystack 的通用桥梁，理解了operation_id/arguments与messages/service_openapi_spec/service_credentials这两组核心入参，即可快速对接包括搜索、数据库、支付等在内的各类 REST 服务。

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