Haystack OAuth 集成实战OAuthTokenResolver 与三种 Token Source 全解析【免费下载链接】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 生态中OAuth 是打通 SharePoint、Google Drive 等企业级数据源的钥匙。本指南以 docs-website/reference/integrations-api/oauth.md 为骨架结合 OAuthTokenResolver 使用文档系统讲解OAuthTokenResolver组件及其可插拔的三种 token source并给出在 Pipeline 中与 Retriever/Fetcher 串联的完整实战方案。一、为什么需要 OAuthTokenResolver在构建企业级 LLM 应用时检索 SharePoint、Google Drive 等云存储中的文档是高频需求。这些服务普遍使用 OAuth 2.0 进行身份认证而 token 的获取、刷新、缓存是一个脏活token 有生命周期access token 通常几分钟到一小时即过期需要定期用 refresh token 换取新 token不同场景 token 来源不同单固定身份、多用户并发、多副本部署各有各的最佳实践下游组件不该关心 token 怎么来的Retriever、Fetcher 只需要一个合法的access_token。OAuthTokenResolver正是为此设计的它在Pipeline 运行时解析 OAuth access token并从access_token输出 socket 发射出去供下游组件如MSSharePointRetriever、MSSharePointFetcher、GoogleDriveRetriever、GoogleDriveFetcher通过普通连接消费。二、设计思想薄封装 可插拔 Token Source从 oauth.md 的定义可以看出OAuthTokenResolver本身是一个极薄的包装器The resolver component is a thin wrapper over a pluggable token source that decideswherethe token comes from真正干活的是你传入的token source它决定了 token 从哪里来。这种分离让认证策略refresh-token grant、逐请求 token 交换、静态长寿命 token可以随意切换而无需改动 Pipeline 的其余部分。所有 token source 都可以从haystack_integrations.utils.oauth导入并且在haystack核心库的haystack.utils中提供了Secret类型用于安全地承载凭据详见 haystack/utils/auth.py。三种 Token Source 速览Source适用场景每请求输入OAuthRefreshTokenSource单一固定身份存有 refresh token希望自动换取短期 access token 并缓存无OAuthTokenExchangeSource多用户或多副本部署用传入的逐请求用户断言换取下游 token无需持久化存储。实现 RFC 8693 token exchange 和 Microsoft on-behalf-of 流程subject_tokenOAuthStaticTokenSource提供方签发不过期 token如 Slack、Notion在带外管理无协议层TokenSource 与 SubjectTokenSource在 oauth.md 中定义了两个 ProtocolTokenSource无逐请求输入的 token source凭据在构造时固定如OAuthRefreshTokenSource、OAuthStaticTokenSource。它们设置类属性requires_subject_token FalseOAuthTokenResolver会将它们作为源节点运行无 run 输入。协议方法包括resolve()、resolve_async()、to_dict()、from_dict()。SubjectTokenSource通过交换逐请求 subject token 来解析 access token。subject_token是由控制器/应用注入的逐请求凭据例如传入的用户断言不是终端用户自行选择的。OAuthTokenExchangeSource实现该协议设置requires_subject_token True这会让OAuthTokenResolver声明一个强制的subject_tokenrun 输入。错误类型oauth.md 还定义了异常体系全部位于haystack_integrations.utils.oauth.errorsOAuthErrorOAuth 集成抛出的所有错误的基类继承ExceptionOAuthConfigErrorOAuth 组件或 token source 配置错误时抛出TokenRefreshError无法在身份提供方解析或刷新 token 时抛出。三、OAuthTokenResolver 组件详解3.1__init____init__(token_source: TokenSource | SubjectTokenSource) - None初始化解析器。唯一必填参数是token_source——决定 access token 解析策略的对象。若它设置requires_subject_token True例如OAuthTokenExchangeSource解析器声明强制subject_tokenrun 输入否则解析器无 run 输入。注意若传入的token_source未实现 token-source 协议将抛出OAuthConfigError。3.2runrun(**kwargs: Any) - dict[str, str]解析 access token 并发射。当配置的 source 需要 subject token 时kwargs携带subject_token作为强制输入由应用/控制器逐请求注入对于仅配置型 source不声明输入kwargs为空。返回仅含单个access_token键的字典值为 bearer token 字符串。异常若 source 需要subject_token但缺失或为空抛出OAuthConfigError。3.3run_asyncrun_async(**kwargs: Any) - dict[str, str]run的异步版本签名、返回与异常语义一致。注意同一个 source 实例要么用于同步模式、要么用于异步模式不要两者混用见OAuthRefreshTokenSource.resolve_async的说明。3.4 序列化to_dict/from_dictto_dict() - dict[str, Any]将组件序列化为字典from_dict(data: dict[str, Any]) - OAuthTokenResolver从字典反序列化。若序列化中的token_source类型无法导入抛出ImportError。这保证了包含OAuthTokenResolver的 Pipeline 可以完整地保存为 YAML 并重新加载。四、三种 Token Source 深度剖析4.1 OAuthRefreshTokenSource单固定身份的 refresh-token grant核心行为通过向 OAuth token 端点发起RFC 6749 refresh-token grant来解析 access token。给定存储的 refresh token 与客户端凭据换取 access token 并在进程内缓存至临近过期。若身份提供方在交换时轮换 refresh token新值在进程生命周期内保留并通过可选的on_rotate回调暴露出来以便持久化。构造参数__init__( token_url: str, client_id: str, *, refresh_token: Secret Secret.from_env_var(OAUTH_REFRESH_TOKEN), client_secret: Secret | None None, scopes: list[str] | None None, scope_delimiter: str , expiry_buffer_seconds: int DEFAULT_EXPIRY_BUFFER_SECONDS, timeout: float DEFAULT_TIMEOUT_SECONDS, on_rotate: Callable[[str], None] | None None ) - None参数说明token_urlOAuth 2.0 token 端点地址必填client_idOAuth 客户端标识必填refresh_token要交换的 refresh token。默认读取OAUTH_REFRESH_TOKEN环境变量client_secret机密客户端的客户端密钥公开客户端可省略scopes请求的 OAuth 作用域用scope_delimiter连接。作用域值由提供方决定需查阅其文档scope_delimiter连接 scopes 的分隔符默认空格部分提供方用逗号expiry_buffer_seconds在声明过期前多少秒刷新缓存 access token默认值见常量DEFAULT_EXPIRY_BUFFER_SECONDStimeout请求 token 端点的超时秒数on_rotate提供方轮换 refresh token 时、携带新值的可选回调。用它把轮换后的 token 持久化source 本身只在进程内保存⚠️ 多副本部署的陷阱该 source 是单身份的——每个实例对应一个 refresh token且进程内缓存不跨进程共享。在多副本部署中每个副本维护自己的缓存对于轮换签发一次性refresh token 的提供方各副本可能互相使彼此的 token 失效——除非通过on_rotate将轮换持久化到共享存储并由单一所有者驱动刷新。选择建议单固定身份 refresh grant 场景选它长寿命不过期 token 用OAuthStaticTokenSource多副本/多用户后端用OAuthTokenExchangeSource。4.2 OAuthTokenExchangeSource多用户无状态 token 交换核心行为在 OAuth token 端点交换逐请求的 subject token。实现RFC 8693 token exchange并通过配置支持Microsoft on-behalf-of 流程。与OAuthRefreshTokenSource不同它是无需任何持久化存储的多用户方案逐请求的subject_token传入的用户断言本身就是用户身份被实时交换为下游 token。解析出的 token 按 subject token在内存中缓存有界 LRU至临近过期。由于不持久化任何实例状态它也是多副本部署的正确选择。构造参数__init__( token_url: str, client_id: str, *, client_secret: Secret | None None, grant_type: str DEFAULT_TOKEN_EXCHANGE_GRANT, subject_token_param: str subject_token, subject_token_type: str | None None, requested_token_type: str | None None, scopes: list[str] | None None, scope_delimiter: str , extra_token_params: dict[str, str] | None None, expiry_buffer_seconds: int DEFAULT_EXPIRY_BUFFER_SECONDS, cache_max_size: int DEFAULT_CACHE_MAX_SIZE, timeout: float DEFAULT_TIMEOUT_SECONDS ) - None参数说明token_urlOAuth 2.0 token 端点地址必填client_idOAuth 客户端标识必填client_secret机密客户端的客户端密钥公开客户端可省略grant_type作为grant_type表单参数发送的授权类型。默认 RFC 8693 token-exchange grantMicrosoft on-behalf-of 需设为urn:ietf:params:oauth:grant-type:jwt-bearersubject_token_param承载逐请求 subject token 的表单参数名。默认subject_tokenRFC 8693。部分提供方用不同名字如 Microsoft 用assertionsubject_token_typeRFC 8693 中 subject token 的类型标识作为subject_token_type表单参数发送未设置时省略。RFC 8693 token exchange 必需如urn:ietf:params:oauth:token-type:access_tokenMicrosoft on-behalf-of 流程不使用requested_token_typeRFC 8693 中期望返回的 token 类型标识作为requested_token_type表单参数发送未设置时省略。可选scopes请求的作用域用scope_delimiter连接。值由提供方决定只有线上格式是标准化的RFC 6749 §3.3scope_delimiter连接 scopes 的分隔符默认空格extra_token_params每个请求中原样包含的额外表单参数如{requested_token_use: on_behalf_of}。最后应用因此任何键都会覆盖从其他参数推导出的对应表单参数如grant_type、subject_token_type、requested_token_type、scope、client_secretexpiry_buffer_seconds在声明过期前多少秒刷新缓存的 access tokencache_max_size内存缓存中保留的每用户 token 最大数量满时逐出最久未使用LRU条目timeout请求 token 端点的超时秒数配置要点提供方差异全部以配置表达——grant_type、subject_token_param如 Microsoft 的assertion、scopes、extra_token_params如{requested_token_use: on_behalf_of}。4.3 OAuthStaticTokenSource静态长寿命 token核心行为原样返回配置好的长寿命 access token。适合签发不过期 token的提供方如 Slack、Notion无需刷新流程token 在带外管理。若提供方签发需要刷新的短期 token则应改用OAuthRefreshTokenSource。无逐请求输入。__init__(token: Secret) - None参数说明token要返回的长寿命 access tokenSecret类型方法resolve()、resolve_async()直接返回配置的 tokento_dict()/from_dict()支持序列化。五、Secret API凭据安全承载所有 token source 的敏感参数refresh token、client secret、静态 token都使用haystack.utils.Secret类型实现见 haystack/utils/auth.py这是 Haystack 的凭据管理标准详见 secret-management 概念文档Secret.from_env_var(MS_REFRESH_TOKEN)从环境变量读取强烈推荐。Pipeline 序列化为 YAML 时只保存环境变量名不泄露明文Secret.from_token(...)直接内联 token不可序列化安全特性防止敏感数据意外暴露。EnvVarSecret还支持传入多个候选环境变量并按顺序取第一个已设置的strictFalse时未设置也不会报错见 haystack/utils/auth.py。六、独立使用快速验证三种场景6.1 用 refresh token 解析Microsoft Graph 示例from haystack.utils import Secret from haystack_integrations.components.connectors.oauth import OAuthTokenResolver from haystack_integrations.utils.oauth import OAuthRefreshTokenSource resolver OAuthTokenResolver( token_sourceOAuthRefreshTokenSource( token_urlhttps://login.microsoftonline.com/common/oauth2/v2.0/token, client_idaaa-bbb-ccc, refresh_tokenSecret.from_env_var(MS_REFRESH_TOKEN), scopes[ https://graph.microsoft.com/Files.Read.All, offline_access, ], ), ) access_token resolver.run()[access_token]作用域是提供方特定的Microsoft Graph 需要https://graph.microsoft.com/Files.Read.All这类作用域Google Drive 则需https://www.googleapis.com/auth/drive.readonly。务必查阅身份提供方文档确认准确的作用域值。Microsoft 的刷新流程通常还需要offline_access以获取 refresh token。6.2 用静态长寿命 token 解析Slack / Notion 场景from haystack.utils import Secret from haystack_integrations.components.connectors.oauth import OAuthTokenResolver from haystack_integrations.utils.oauth import OAuthStaticTokenSource resolver OAuthTokenResolver( token_sourceOAuthStaticTokenSource(tokenSecret.from_env_var(SERVICE_TOKEN)), ) access_token resolver.run()[access_token]6.3 多用户后端逐请求 token 交换Microsoft on-behalf-offrom haystack_integrations.components.connectors.oauth import OAuthTokenResolver from haystack_integrations.utils.oauth import OAuthTokenExchangeSource resolver OAuthTokenResolver( token_sourceOAuthTokenExchangeSource( token_urlhttps://login.microsoftonline.com/tenant/oauth2/v2.0/token, client_idaaa-bbb-ccc, subject_token_paramassertion, grant_typeurn:ietf:params:oauth:grant-type:jwt-bearer, scopes[https://graph.microsoft.com/Files.Read.All], extra_token_params{requested_token_use: on_behalf_of}, ), ) # subject_token 是传入的逐请求用户断言由你的应用注入。 access_token resolver.run(subject_tokenincoming-user-assertion)[access_token]七、接入 Pipeline检索 SharePoint 全流程在 Pipeline 中把解析器的access_token输出连接到下游一个或多个组件的access_token输入即可。下面的例子把解析器接入MSSharePointRetriever使运行时只需提供 query见 mssharepointretriever.mdxfrom haystack import Pipeline from haystack.utils import Secret from haystack_integrations.components.connectors.oauth import OAuthTokenResolver from haystack_integrations.utils.oauth import OAuthRefreshTokenSource from haystack_integrations.components.retrievers.microsoft_sharepoint import ( MSSharePointRetriever, ) pipeline Pipeline() pipeline.add_component( resolver, OAuthTokenResolver( token_sourceOAuthRefreshTokenSource( token_urlhttps://login.microsoftonline.com/common/oauth2/v2.0/token, client_idaaa-bbb-ccc, refresh_tokenSecret.from_env_var(MS_REFRESH_TOKEN), scopes[ https://graph.microsoft.com/Files.Read.All, https://graph.microsoft.com/Sites.Read.All, offline_access, ], ), ), ) pipeline.add_component(retriever, MSSharePointRetriever(top_k5)) pipeline.connect(resolver.access_token, retriever.access_token) result pipeline.run({retriever: {query: quarterly roadmap}}) documents result[retriever][documents]检索 下载 转换的端到端管线OAuthTokenResolver的单个access_token输出可以连接到多个下游输入。以下管线把一个 token 同时喂给 Retriever 和 Fetcher再经FileTypeRouter分流到对应转换器完整示例见 mssharepointfetcher.mdxfrom haystack import Pipeline from haystack.utils import Secret from haystack.components.routers import FileTypeRouter from haystack.components.converters import PyPDFToDocument, DOCXToDocument from haystack_integrations.components.connectors.oauth import OAuthTokenResolver from haystack_integrations.utils.oauth import OAuthRefreshTokenSource from haystack_integrations.components.retrievers.microsoft_sharepoint import ( MSSharePointRetriever, ) from haystack_integrations.components.fetchers.microsoft_sharepoint import ( MSSharePointFetcher, ) pipeline Pipeline() pipeline.add_component( resolver, OAuthTokenResolver( token_sourceOAuthRefreshTokenSource( token_urlhttps://login.microsoftonline.com/common/oauth2/v2.0/token, client_idaaa-bbb-ccc, refresh_tokenSecret.from_env_var(MS_REFRESH_TOKEN), scopes[ https://graph.microsoft.com/Files.Read.All, https://graph.microsoft.com/Sites.Read.All, offline_access, ], ), ), ) pipeline.add_component(retriever, MSSharePointRetriever(top_k5)) pipeline.add_component(fetcher, MSSharePointFetcher()) pipeline.add_component( router, FileTypeRouter( mime_types[ application/pdf, application/vnd.openxmlformats-officedocument.wordprocessingml.document, ], ), ) pipeline.add_component(pdf_converter, PyPDFToDocument()) pipeline.add_component(docx_converter, DOCXToDocument()) # 同一个 token 同时喂给 Retriever 和 Fetcher。 pipeline.connect(resolver.access_token, retriever.access_token) pipeline.connect(resolver.access_token, fetcher.access_token) # 检索到的文档成为 Fetcher 的目标。 pipeline.connect(retriever.documents, fetcher.targets) # 将每个获取的 ByteStream 路由到匹配的转换器。 pipeline.connect(fetcher.streams, router.sources) pipeline.connect(router.application/pdf, pdf_converter.sources) pipeline.connect( router.application/vnd.openxmlformats-officedocument.wordprocessingml.document, docx_converter.sources, ) result pipeline.run({retriever: {query: quarterly roadmap}})Google Drive 场景同样模式适用于 Google DriveGoogleDriveRetriever通过 Drive API v3 的files.list做全文检索默认只返回元数据include_contentTrue时可导出原生 Google 文档文本GoogleDriveFetcher下载完整内容为ByteStream。token 需携带委托的 Google OAuth 作用域如https://www.googleapis.com/auth/drive.readonly仅元数据的drive.metadata.readonly作用域无法搜索内容或导出文档详见 googledriveretriever.mdx 与 google_drive.mdfrom haystack import Pipeline from haystack.utils import Secret from haystack_integrations.components.connectors.oauth import OAuthTokenResolver from haystack_integrations.utils.oauth import OAuthRefreshTokenSource from haystack_integrations.components.retrievers.google_drive import ( GoogleDriveRetriever, ) pipeline Pipeline() pipeline.add_component( resolver, OAuthTokenResolver( token_sourceOAuthRefreshTokenSource( token_urlhttps://oauth2.googleapis.com/token, client_idaaa-bbb-ccc, refresh_tokenSecret.from_env_var(GOOGLE_REFRESH_TOKEN), scopes[https://www.googleapis.com/auth/drive.readonly], ), ), ) pipeline.add_component(retriever, GoogleDriveRetriever(top_k5)) pipeline.connect(resolver.access_token, retriever.access_token) result pipeline.run({retriever: {query: quarterly roadmap}}) documents result[retriever][documents]八、选型与部署决策指南决策维度OAuthRefreshTokenSourceOAuthTokenExchangeSourceOAuthStaticTokenSource身份模型单固定身份多用户每请求身份单固定身份持久化无进程内缓存 可选on_rotate持久化无有界 LRU 内存缓存无多副本部署⚠️ 需on_rotate持久化 单一刷新所有者✅ 天然适配✅ 天然适配适用提供方Microsoft Graph、Google Drive 等短期 token企业 SSO、Microsoft on-behalf-ofSlack、Notion 等不过期 token每请求输入无subject_token强制无关键部署提醒对轮换 refresh token 的提供方多副本 OAuthRefreshTokenSource存在 token 互相失效的风险务必通过on_rotate回调把轮换后的 token 写入共享存储并保证同一时刻只有一个副本驱动刷新。多用户/多副本场景优先考虑OAuthTokenExchangeSource——它无持久化状态逐请求交换天然规避了该问题。九、安装与更多资料pip install oauth-haystack配套集成各自独立安装SharePoint/OneDrivepip install microsoft-sharepoint-haystack组件见 mssharepointfetcher.mdxGoogle Drivepip install google-drive-haystack组件见 googledriveretriever.mdxOAuthTokenResolver的组件级说明见 oauthtokenresolver.mdxOAuth 集成的完整 API 参考见 oauth.mdSecret的序列化行为细节可查阅 secret-management.mdx。结合以上内容你可以在 Haystack Pipeline 中把 OAuth 认证完全前置化运行时只需注入 query或 subject tokenaccess token 的获取、缓存与刷新全部由OAuthTokenResolver与对应的 token source 透明完成下游的检索、下载、转换组件无需感知任何认证细节。【免费下载链接】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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 Haystack OAuth 集成实战：OAuthTokenResolver 与三种 Token Source 全解析
Haystack OAuth 集成实战OAuthTokenResolver 与三种 Token Source 全解析【免费下载链接】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 生态中OAuth 是打通 SharePoint、Google Drive 等企业级数据源的钥匙。本指南以 docs-website/reference/integrations-api/oauth.md 为骨架结合 OAuthTokenResolver 使用文档系统讲解OAuthTokenResolver组件及其可插拔的三种 token source并给出在 Pipeline 中与 Retriever/Fetcher 串联的完整实战方案。一、为什么需要 OAuthTokenResolver在构建企业级 LLM 应用时检索 SharePoint、Google Drive 等云存储中的文档是高频需求。这些服务普遍使用 OAuth 2.0 进行身份认证而 token 的获取、刷新、缓存是一个脏活token 有生命周期access token 通常几分钟到一小时即过期需要定期用 refresh token 换取新 token不同场景 token 来源不同单固定身份、多用户并发、多副本部署各有各的最佳实践下游组件不该关心 token 怎么来的Retriever、Fetcher 只需要一个合法的access_token。OAuthTokenResolver正是为此设计的它在Pipeline 运行时解析 OAuth access token并从access_token输出 socket 发射出去供下游组件如MSSharePointRetriever、MSSharePointFetcher、GoogleDriveRetriever、GoogleDriveFetcher通过普通连接消费。二、设计思想薄封装 可插拔 Token Source从 oauth.md 的定义可以看出OAuthTokenResolver本身是一个极薄的包装器The resolver component is a thin wrapper over a pluggable token source that decideswherethe token comes from真正干活的是你传入的token source它决定了 token 从哪里来。这种分离让认证策略refresh-token grant、逐请求 token 交换、静态长寿命 token可以随意切换而无需改动 Pipeline 的其余部分。所有 token source 都可以从haystack_integrations.utils.oauth导入并且在haystack核心库的haystack.utils中提供了Secret类型用于安全地承载凭据详见 haystack/utils/auth.py。三种 Token Source 速览Source适用场景每请求输入OAuthRefreshTokenSource单一固定身份存有 refresh token希望自动换取短期 access token 并缓存无OAuthTokenExchangeSource多用户或多副本部署用传入的逐请求用户断言换取下游 token无需持久化存储。实现 RFC 8693 token exchange 和 Microsoft on-behalf-of 流程subject_tokenOAuthStaticTokenSource提供方签发不过期 token如 Slack、Notion在带外管理无协议层TokenSource 与 SubjectTokenSource在 oauth.md 中定义了两个 ProtocolTokenSource无逐请求输入的 token source凭据在构造时固定如OAuthRefreshTokenSource、OAuthStaticTokenSource。它们设置类属性requires_subject_token FalseOAuthTokenResolver会将它们作为源节点运行无 run 输入。协议方法包括resolve()、resolve_async()、to_dict()、from_dict()。SubjectTokenSource通过交换逐请求 subject token 来解析 access token。subject_token是由控制器/应用注入的逐请求凭据例如传入的用户断言不是终端用户自行选择的。OAuthTokenExchangeSource实现该协议设置requires_subject_token True这会让OAuthTokenResolver声明一个强制的subject_tokenrun 输入。错误类型oauth.md 还定义了异常体系全部位于haystack_integrations.utils.oauth.errorsOAuthErrorOAuth 集成抛出的所有错误的基类继承ExceptionOAuthConfigErrorOAuth 组件或 token source 配置错误时抛出TokenRefreshError无法在身份提供方解析或刷新 token 时抛出。三、OAuthTokenResolver 组件详解3.1__init____init__(token_source: TokenSource | SubjectTokenSource) - None初始化解析器。唯一必填参数是token_source——决定 access token 解析策略的对象。若它设置requires_subject_token True例如OAuthTokenExchangeSource解析器声明强制subject_tokenrun 输入否则解析器无 run 输入。注意若传入的token_source未实现 token-source 协议将抛出OAuthConfigError。3.2runrun(**kwargs: Any) - dict[str, str]解析 access token 并发射。当配置的 source 需要 subject token 时kwargs携带subject_token作为强制输入由应用/控制器逐请求注入对于仅配置型 source不声明输入kwargs为空。返回仅含单个access_token键的字典值为 bearer token 字符串。异常若 source 需要subject_token但缺失或为空抛出OAuthConfigError。3.3run_asyncrun_async(**kwargs: Any) - dict[str, str]run的异步版本签名、返回与异常语义一致。注意同一个 source 实例要么用于同步模式、要么用于异步模式不要两者混用见OAuthRefreshTokenSource.resolve_async的说明。3.4 序列化to_dict/from_dictto_dict() - dict[str, Any]将组件序列化为字典from_dict(data: dict[str, Any]) - OAuthTokenResolver从字典反序列化。若序列化中的token_source类型无法导入抛出ImportError。这保证了包含OAuthTokenResolver的 Pipeline 可以完整地保存为 YAML 并重新加载。四、三种 Token Source 深度剖析4.1 OAuthRefreshTokenSource单固定身份的 refresh-token grant核心行为通过向 OAuth token 端点发起RFC 6749 refresh-token grant来解析 access token。给定存储的 refresh token 与客户端凭据换取 access token 并在进程内缓存至临近过期。若身份提供方在交换时轮换 refresh token新值在进程生命周期内保留并通过可选的on_rotate回调暴露出来以便持久化。构造参数__init__( token_url: str, client_id: str, *, refresh_token: Secret Secret.from_env_var(OAUTH_REFRESH_TOKEN), client_secret: Secret | None None, scopes: list[str] | None None, scope_delimiter: str , expiry_buffer_seconds: int DEFAULT_EXPIRY_BUFFER_SECONDS, timeout: float DEFAULT_TIMEOUT_SECONDS, on_rotate: Callable[[str], None] | None None ) - None参数说明token_urlOAuth 2.0 token 端点地址必填client_idOAuth 客户端标识必填refresh_token要交换的 refresh token。默认读取OAUTH_REFRESH_TOKEN环境变量client_secret机密客户端的客户端密钥公开客户端可省略scopes请求的 OAuth 作用域用scope_delimiter连接。作用域值由提供方决定需查阅其文档scope_delimiter连接 scopes 的分隔符默认空格部分提供方用逗号expiry_buffer_seconds在声明过期前多少秒刷新缓存 access token默认值见常量DEFAULT_EXPIRY_BUFFER_SECONDStimeout请求 token 端点的超时秒数on_rotate提供方轮换 refresh token 时、携带新值的可选回调。用它把轮换后的 token 持久化source 本身只在进程内保存⚠️ 多副本部署的陷阱该 source 是单身份的——每个实例对应一个 refresh token且进程内缓存不跨进程共享。在多副本部署中每个副本维护自己的缓存对于轮换签发一次性refresh token 的提供方各副本可能互相使彼此的 token 失效——除非通过on_rotate将轮换持久化到共享存储并由单一所有者驱动刷新。选择建议单固定身份 refresh grant 场景选它长寿命不过期 token 用OAuthStaticTokenSource多副本/多用户后端用OAuthTokenExchangeSource。4.2 OAuthTokenExchangeSource多用户无状态 token 交换核心行为在 OAuth token 端点交换逐请求的 subject token。实现RFC 8693 token exchange并通过配置支持Microsoft on-behalf-of 流程。与OAuthRefreshTokenSource不同它是无需任何持久化存储的多用户方案逐请求的subject_token传入的用户断言本身就是用户身份被实时交换为下游 token。解析出的 token 按 subject token在内存中缓存有界 LRU至临近过期。由于不持久化任何实例状态它也是多副本部署的正确选择。构造参数__init__( token_url: str, client_id: str, *, client_secret: Secret | None None, grant_type: str DEFAULT_TOKEN_EXCHANGE_GRANT, subject_token_param: str subject_token, subject_token_type: str | None None, requested_token_type: str | None None, scopes: list[str] | None None, scope_delimiter: str , extra_token_params: dict[str, str] | None None, expiry_buffer_seconds: int DEFAULT_EXPIRY_BUFFER_SECONDS, cache_max_size: int DEFAULT_CACHE_MAX_SIZE, timeout: float DEFAULT_TIMEOUT_SECONDS ) - None参数说明token_urlOAuth 2.0 token 端点地址必填client_idOAuth 客户端标识必填client_secret机密客户端的客户端密钥公开客户端可省略grant_type作为grant_type表单参数发送的授权类型。默认 RFC 8693 token-exchange grantMicrosoft on-behalf-of 需设为urn:ietf:params:oauth:grant-type:jwt-bearersubject_token_param承载逐请求 subject token 的表单参数名。默认subject_tokenRFC 8693。部分提供方用不同名字如 Microsoft 用assertionsubject_token_typeRFC 8693 中 subject token 的类型标识作为subject_token_type表单参数发送未设置时省略。RFC 8693 token exchange 必需如urn:ietf:params:oauth:token-type:access_tokenMicrosoft on-behalf-of 流程不使用requested_token_typeRFC 8693 中期望返回的 token 类型标识作为requested_token_type表单参数发送未设置时省略。可选scopes请求的作用域用scope_delimiter连接。值由提供方决定只有线上格式是标准化的RFC 6749 §3.3scope_delimiter连接 scopes 的分隔符默认空格extra_token_params每个请求中原样包含的额外表单参数如{requested_token_use: on_behalf_of}。最后应用因此任何键都会覆盖从其他参数推导出的对应表单参数如grant_type、subject_token_type、requested_token_type、scope、client_secretexpiry_buffer_seconds在声明过期前多少秒刷新缓存的 access tokencache_max_size内存缓存中保留的每用户 token 最大数量满时逐出最久未使用LRU条目timeout请求 token 端点的超时秒数配置要点提供方差异全部以配置表达——grant_type、subject_token_param如 Microsoft 的assertion、scopes、extra_token_params如{requested_token_use: on_behalf_of}。4.3 OAuthStaticTokenSource静态长寿命 token核心行为原样返回配置好的长寿命 access token。适合签发不过期 token的提供方如 Slack、Notion无需刷新流程token 在带外管理。若提供方签发需要刷新的短期 token则应改用OAuthRefreshTokenSource。无逐请求输入。__init__(token: Secret) - None参数说明token要返回的长寿命 access tokenSecret类型方法resolve()、resolve_async()直接返回配置的 tokento_dict()/from_dict()支持序列化。五、Secret API凭据安全承载所有 token source 的敏感参数refresh token、client secret、静态 token都使用haystack.utils.Secret类型实现见 haystack/utils/auth.py这是 Haystack 的凭据管理标准详见 secret-management 概念文档Secret.from_env_var(MS_REFRESH_TOKEN)从环境变量读取强烈推荐。Pipeline 序列化为 YAML 时只保存环境变量名不泄露明文Secret.from_token(...)直接内联 token不可序列化安全特性防止敏感数据意外暴露。EnvVarSecret还支持传入多个候选环境变量并按顺序取第一个已设置的strictFalse时未设置也不会报错见 haystack/utils/auth.py。六、独立使用快速验证三种场景6.1 用 refresh token 解析Microsoft Graph 示例from haystack.utils import Secret from haystack_integrations.components.connectors.oauth import OAuthTokenResolver from haystack_integrations.utils.oauth import OAuthRefreshTokenSource resolver OAuthTokenResolver( token_sourceOAuthRefreshTokenSource( token_urlhttps://login.microsoftonline.com/common/oauth2/v2.0/token, client_idaaa-bbb-ccc, refresh_tokenSecret.from_env_var(MS_REFRESH_TOKEN), scopes[ https://graph.microsoft.com/Files.Read.All, offline_access, ], ), ) access_token resolver.run()[access_token]作用域是提供方特定的Microsoft Graph 需要https://graph.microsoft.com/Files.Read.All这类作用域Google Drive 则需https://www.googleapis.com/auth/drive.readonly。务必查阅身份提供方文档确认准确的作用域值。Microsoft 的刷新流程通常还需要offline_access以获取 refresh token。6.2 用静态长寿命 token 解析Slack / Notion 场景from haystack.utils import Secret from haystack_integrations.components.connectors.oauth import OAuthTokenResolver from haystack_integrations.utils.oauth import OAuthStaticTokenSource resolver OAuthTokenResolver( token_sourceOAuthStaticTokenSource(tokenSecret.from_env_var(SERVICE_TOKEN)), ) access_token resolver.run()[access_token]6.3 多用户后端逐请求 token 交换Microsoft on-behalf-offrom haystack_integrations.components.connectors.oauth import OAuthTokenResolver from haystack_integrations.utils.oauth import OAuthTokenExchangeSource resolver OAuthTokenResolver( token_sourceOAuthTokenExchangeSource( token_urlhttps://login.microsoftonline.com/tenant/oauth2/v2.0/token, client_idaaa-bbb-ccc, subject_token_paramassertion, grant_typeurn:ietf:params:oauth:grant-type:jwt-bearer, scopes[https://graph.microsoft.com/Files.Read.All], extra_token_params{requested_token_use: on_behalf_of}, ), ) # subject_token 是传入的逐请求用户断言由你的应用注入。 access_token resolver.run(subject_tokenincoming-user-assertion)[access_token]七、接入 Pipeline检索 SharePoint 全流程在 Pipeline 中把解析器的access_token输出连接到下游一个或多个组件的access_token输入即可。下面的例子把解析器接入MSSharePointRetriever使运行时只需提供 query见 mssharepointretriever.mdxfrom haystack import Pipeline from haystack.utils import Secret from haystack_integrations.components.connectors.oauth import OAuthTokenResolver from haystack_integrations.utils.oauth import OAuthRefreshTokenSource from haystack_integrations.components.retrievers.microsoft_sharepoint import ( MSSharePointRetriever, ) pipeline Pipeline() pipeline.add_component( resolver, OAuthTokenResolver( token_sourceOAuthRefreshTokenSource( token_urlhttps://login.microsoftonline.com/common/oauth2/v2.0/token, client_idaaa-bbb-ccc, refresh_tokenSecret.from_env_var(MS_REFRESH_TOKEN), scopes[ https://graph.microsoft.com/Files.Read.All, https://graph.microsoft.com/Sites.Read.All, offline_access, ], ), ), ) pipeline.add_component(retriever, MSSharePointRetriever(top_k5)) pipeline.connect(resolver.access_token, retriever.access_token) result pipeline.run({retriever: {query: quarterly roadmap}}) documents result[retriever][documents]检索 下载 转换的端到端管线OAuthTokenResolver的单个access_token输出可以连接到多个下游输入。以下管线把一个 token 同时喂给 Retriever 和 Fetcher再经FileTypeRouter分流到对应转换器完整示例见 mssharepointfetcher.mdxfrom haystack import Pipeline from haystack.utils import Secret from haystack.components.routers import FileTypeRouter from haystack.components.converters import PyPDFToDocument, DOCXToDocument from haystack_integrations.components.connectors.oauth import OAuthTokenResolver from haystack_integrations.utils.oauth import OAuthRefreshTokenSource from haystack_integrations.components.retrievers.microsoft_sharepoint import ( MSSharePointRetriever, ) from haystack_integrations.components.fetchers.microsoft_sharepoint import ( MSSharePointFetcher, ) pipeline Pipeline() pipeline.add_component( resolver, OAuthTokenResolver( token_sourceOAuthRefreshTokenSource( token_urlhttps://login.microsoftonline.com/common/oauth2/v2.0/token, client_idaaa-bbb-ccc, refresh_tokenSecret.from_env_var(MS_REFRESH_TOKEN), scopes[ https://graph.microsoft.com/Files.Read.All, https://graph.microsoft.com/Sites.Read.All, offline_access, ], ), ), ) pipeline.add_component(retriever, MSSharePointRetriever(top_k5)) pipeline.add_component(fetcher, MSSharePointFetcher()) pipeline.add_component( router, FileTypeRouter( mime_types[ application/pdf, application/vnd.openxmlformats-officedocument.wordprocessingml.document, ], ), ) pipeline.add_component(pdf_converter, PyPDFToDocument()) pipeline.add_component(docx_converter, DOCXToDocument()) # 同一个 token 同时喂给 Retriever 和 Fetcher。 pipeline.connect(resolver.access_token, retriever.access_token) pipeline.connect(resolver.access_token, fetcher.access_token) # 检索到的文档成为 Fetcher 的目标。 pipeline.connect(retriever.documents, fetcher.targets) # 将每个获取的 ByteStream 路由到匹配的转换器。 pipeline.connect(fetcher.streams, router.sources) pipeline.connect(router.application/pdf, pdf_converter.sources) pipeline.connect( router.application/vnd.openxmlformats-officedocument.wordprocessingml.document, docx_converter.sources, ) result pipeline.run({retriever: {query: quarterly roadmap}})Google Drive 场景同样模式适用于 Google DriveGoogleDriveRetriever通过 Drive API v3 的files.list做全文检索默认只返回元数据include_contentTrue时可导出原生 Google 文档文本GoogleDriveFetcher下载完整内容为ByteStream。token 需携带委托的 Google OAuth 作用域如https://www.googleapis.com/auth/drive.readonly仅元数据的drive.metadata.readonly作用域无法搜索内容或导出文档详见 googledriveretriever.mdx 与 google_drive.mdfrom haystack import Pipeline from haystack.utils import Secret from haystack_integrations.components.connectors.oauth import OAuthTokenResolver from haystack_integrations.utils.oauth import OAuthRefreshTokenSource from haystack_integrations.components.retrievers.google_drive import ( GoogleDriveRetriever, ) pipeline Pipeline() pipeline.add_component( resolver, OAuthTokenResolver( token_sourceOAuthRefreshTokenSource( token_urlhttps://oauth2.googleapis.com/token, client_idaaa-bbb-ccc, refresh_tokenSecret.from_env_var(GOOGLE_REFRESH_TOKEN), scopes[https://www.googleapis.com/auth/drive.readonly], ), ), ) pipeline.add_component(retriever, GoogleDriveRetriever(top_k5)) pipeline.connect(resolver.access_token, retriever.access_token) result pipeline.run({retriever: {query: quarterly roadmap}}) documents result[retriever][documents]八、选型与部署决策指南决策维度OAuthRefreshTokenSourceOAuthTokenExchangeSourceOAuthStaticTokenSource身份模型单固定身份多用户每请求身份单固定身份持久化无进程内缓存 可选on_rotate持久化无有界 LRU 内存缓存无多副本部署⚠️ 需on_rotate持久化 单一刷新所有者✅ 天然适配✅ 天然适配适用提供方Microsoft Graph、Google Drive 等短期 token企业 SSO、Microsoft on-behalf-ofSlack、Notion 等不过期 token每请求输入无subject_token强制无关键部署提醒对轮换 refresh token 的提供方多副本 OAuthRefreshTokenSource存在 token 互相失效的风险务必通过on_rotate回调把轮换后的 token 写入共享存储并保证同一时刻只有一个副本驱动刷新。多用户/多副本场景优先考虑OAuthTokenExchangeSource——它无持久化状态逐请求交换天然规避了该问题。九、安装与更多资料pip install oauth-haystack配套集成各自独立安装SharePoint/OneDrivepip install microsoft-sharepoint-haystack组件见 mssharepointfetcher.mdxGoogle Drivepip install google-drive-haystack组件见 googledriveretriever.mdxOAuthTokenResolver的组件级说明见 oauthtokenresolver.mdxOAuth 集成的完整 API 参考见 oauth.mdSecret的序列化行为细节可查阅 secret-management.mdx。结合以上内容你可以在 Haystack Pipeline 中把 OAuth 认证完全前置化运行时只需注入 query或 subject tokenaccess token 的获取、缓存与刷新全部由OAuthTokenResolver与对应的 token source 透明完成下游的检索、下载、转换组件无需感知任何认证细节。【免费下载链接】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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考