基于 Web Fetch API 的 Genkit Action 服务化深入解析 genkit-ai/fetch 插件【免费下载链接】genkitOpen-source framework for building agentic apps in JavaScript, Go, Dart, and Python, built and used in production by Google项目地址: https://gitcode.com/GitHub_Trending/ge/genkit本篇技术指南围绕 Genkit 官方 Web Fetch 插件genkit-ai/fetch展开讲解如何将 Genkit 的 Flow、模型等 Action 以标准 Fetch APIRequest/Response形式暴露为 HTTP 端点适用于 Hono、Bun、Cloudflare Workers、Deno、Node 18、Vercel Edge、Netlify Edge、Elysia、SvelteKit 等任意支持 Web Fetch 标准的运行时与框架。读完本文你将掌握fetchHandler/fetchHandlers/withActionOptions的核心用法、基于ContextProvider的鉴权机制、可断线重连的 Durable StreamingBeta能力以及服务端与客户端完整的数据协议与错误语义。插件定位与设计理念genkit-ai/fetch是一个框架无关的 Genkit 插件它只依赖genkit核心与标准 Web API不含任何特定 Web 框架的绑定。这一点在插件源码的 package.json 中体现得很直接——peerDependencies仅声明了genkit一个依赖而 Hono、Cloudflare 等全部是可选的使用场景而非编译期依赖。其核心 API 采用 Express 风格先把 Action 传给fetchHandler再用返回的处理器去消费请求。处理器的签名统一为(request: Request) PromiseResponse因此它可以被任何能接收 Web 标准Request并返回Response的框架直接调用例如 Hono 的c.req.raw这也是它能横跨 Hono、Bun、Cloudflare Workers、Deno、Node 18、Vercel Edge、Netlify Edge、Elysia、SvelteKit 等运行时与框架的根本原因。从源码结构看插件本体非常精简全部实现集中在 js/plugins/fetch/src/index.ts另外有一个处理错误提取的工具文件 js/plugins/fetch/src/utils.ts。这也印证了其设计目标尽可能薄只做Web 请求 ↔ Genkit Action之间的协议适配不引入任何运行时专属逻辑。安装在任意支持 Web Fetch 的运行时项目中通过 npm 安装npm i genkit-ai/fetch安装后无需注册插件配置直接按需导入fetchHandler、fetchHandlers、withActionOptions等函数即可使用。插件当前版本为 0.5.0许可证为 Apache-2.0。将单个 Action 暴露为端点fetchHandlerfetchHandler(action, options?)接收一个 Genkit ActionFlow、模型等均可返回(request: Request) PromiseResponse的处理器。以 Hono 为例定义一个 Flow 并将其挂载到路由上import { fetchHandler } from genkit-ai/fetch; import { Hono } from hono; const simpleFlow ai.defineFlow(simpleFlow, async (input, { sendChunk }) { const { text } await ai.generate({ model: googleAI.model(gemini-2.0-flash), prompt: input, onChunk: (c) sendChunk(c.text), }); return text; }); const app new Hono(); app.all(/simpleFlow, (c) fetchHandler(simpleFlow)(c.req.raw));注意这里用的是app.all因为客户端请求统一为 POST但服务端不关心具体 HTTP 方法下文协议细节部分会说明请求体约定。模型同样可以直接作为 Action 暴露先从插件中解析出模型实例再交给fetchHandlerconst gai googleAI(); const model await gai.model(gemini-2.0-flash); app.post(/models/gemini-flash, (c) fetchHandler(model)(c.req.raw));从 fetchHandler 源码实现 可以看到fetchHandler本质上是把 Action 与可选 options 闭包进handleActionRequest之后每次请求都走同一套请求解析与 Action 执行流程。一次挂载多个 ActionfetchHandlers 与路径路由fetchHandlers(actions, pathPrefix?)允许把多个 ActionFlow、模型等挂载在同一路径前缀下按请求路径选择要执行的 Action——例如请求/api/hello会执行名为hello的 Actionimport { fetchHandlers } from genkit-ai/fetch; const actions [helloFlow, greetingFlow, streamingFlow]; app.all(/api/*, (c) fetchHandlers(actions, /api)(c.req.raw));客户端调用约定POST /api/actionName请求体为{ data: input }。其路径匹配逻辑在 fetchHandlers 源码 中非常清晰值得展开说明前缀剥离若传入pathPrefix如/api会先对pathname做归一化——/api变为空串/api/xxx剥掉/api/前缀如果路径根本不以前缀开头直接返回 404NOT_FOUND名字匹配去掉开头的/后遍历 actions 数组逐个与 Action 名__action.name比对若是withActionOptions包装过的条目则优先使用options.path未指定时才回落为 Action 名未命中没有任何 Action 匹配时返回404响应体为{ status: NOT_FOUND, message: No action matched the request path. }。对于使用withActionOptions包装的条目其options.path还可以覆盖默认的 Action 名路由这在不想暴露内部 Action 名时非常有用。通过 ContextProvider 实现鉴权ContextProvider是 Genkit 中跨框架统一的请求上下文提取机制。类型定义位于 js/core/src/context.tsexport interface RequestDataT any { method: GET | PUT | POST | DELETE | OPTIONS | QUERY; headers: Recordstring, string; // 必须为小写键保证跨框架可移植 input: T; } export type ContextProviderC extends ActionContext ActionContext, T any (request: RequestDataT) C | PromiseC;将 ContextProvider 与 Action 组合即可实现鉴权。文档给出了完整的鉴权示例import { UserFacingError } from genkit; import type { ContextProvider, RequestData } from genkit/context; import { fetchHandler, fetchHandlers, withActionOptions, } from genkit-ai/fetch; const authContext: ContextProvider{ userId: string } (req: RequestData) { if (req.headers[authorization] ! Bearer open-sesame) { throw new UserFacingError(PERMISSION_DENIED, not authorized); } return { userId: authenticated-user }; }; // 单个 Action 附加鉴权 app.all(/secureFlow, (c) fetchHandler(secureFlow, { contextProvider: authContext })(c.req.raw) ); // 或者包装 Action用于 fetchHandlers const actions [ publicFlow, withActionOptions(secureFlow, { contextProvider: authContext }), ]; app.all(/api/*, (c) fetchHandlers(actions, /api)(c.req.raw));底层执行细节头部小写化插件内部通过headersToObject将Headers转换为全小写键的普通对象见 js/plugins/fetch/src/index.ts这与RequestData接口的headers 必须小写约定一致——所以文档示例中用req.headers[authorization]而非Authorization上下文合并getContext在调用 ContextProvider 前会先组装{ method, headers, input }Provider 的返回值即为传给 Action 的context见 js/plugins/fetch/src/index.ts异常语义Provider 抛错时插件会捕获并调用getCallableJSON/getHttpStatus将错误序列化为 JSON 响应状态码由错误类型决定。在测试 js/plugins/fetch/tests/web_test.ts 中PERMISSION_DENIED被映射为 HTTP 403未授权请求会收到not authorized的错误信息。一个可运行的验证实验在仓库的测试代码中ContextProvider 使用了一个更具可读性的示例见 js/plugins/fetch/tests/web_test.ts校验authorization头是否为open sesame通过后返回{ auth: { user: Ali Baba } }。测试中还覆盖了带正确凭证调用 Flow 成功、错误凭证返回 403两条路径可以直接作为自测参考。Durable StreamingBeta可断线重连的流式输出普通流式响应一旦客户端断开连接流就中断了。Durable Streaming 通过StreamManager将流状态持久化客户端可以断开后携带streamId重新连接不会丢失已产生的流数据。在fetchHandler或withActionOptions的 options 中传入streamManager即可启用import { InMemoryStreamManager } from genkit/beta; import { fetchHandler, fetchHandlers, withActionOptions, } from genkit-ai/fetch; app.all(/myDurableFlow, (c) fetchHandler(myFlow, { streamManager: new InMemoryStreamManager(), })(c.req.raw) ); // 或与 fetchHandlers 搭配 const actions [ withActionOptions(myFlow, { streamManager: new InMemoryStreamManager(), }), ]; app.all(/api/*, (c) fetchHandlers(actions, /api)(c.req.raw));开发环境InMemoryStreamManager来自genkit/beta状态存在内存中适合本地调试生产环境使用持久化实现例如genkit-ai/firebase提供的FirestoreStreamManager、RtdbStreamManager或自行实现StreamManager接口该接口包含open、subscribe以及流的write/done/error等方法定义在genkit/beta。客户端断线重连客户端通过genkit/beta/client的streamFlow发起流式请求拿到streamId保存之后随时重连import { streamFlow } from genkit/beta/client; // 开启一个新流 const result streamFlow({ url: http://localhost:3780/api/myDurableFlow, input: tell me a long story, }); const streamId await result.streamId; // 保存用于重连 // 稍后重连 const reconnected streamFlow({ url: http://localhost:3780/api/myDurableFlow, streamId, });服务端实现要点从 源码 可以看到 Durable Streaming 的关键机制每次请求会通过globalThis.crypto.randomUUID()生成新的streamId并在响应头x-genkit-stream-id中返回见 index.tsAction 的每个输出 chunk 会被写入 SSE 帧data: {message: ...}同时通过AsyncTaskQueue异步持久化到StreamManager保证先推送给当前客户端、再落盘的顺序正确性客户端携带streamId重连时服务端通过streamManager.subscribe(streamId, ...)从已持久化的流中重放数据见 index.ts重连时若流不存在StreamNotFoundError返回HTTP 204若流已超时DEADLINE_EXCEEDED返回一个带错误事件的 200 SSE 响应流结束时会写入data: {result: ...}帧并关闭 writer。测试 js/plugins/fetch/tests/web_test.ts 验证了三条核心路径创建新流拿到streamId、携带streamId重连获得与原始流一致的全部 chunk、以及订阅进行中的流也能收到完整输出不存在流的场景则返回NOT_FOUND: Stream not found.。客户端调用协议runFlow 与 streamFlow客户端使用genkit/beta/client提供的runFlow/streamFlow与 Express 插件使用同一套协议import { runFlow, streamFlow } from genkit/beta/client; // 普通调用 const result await runFlow({ url: http://localhost:3780/api/hello, input: world, }); console.log(result); // 携带鉴权头 const result await runFlow({ url: http://localhost:3780/api/secureGreeting, headers: { Authorization: Bearer open-sesame }, input: { name: Alex }, }); // 流式调用 const result streamFlow({ url: http://localhost:3780/api/streaming, input: { prompt: Say hello in chunks }, }); for await (const chunk of result.stream) { console.log(chunk); } console.log(await result.output);客户端协议实现细节streamFlow的签名定义在 js/genkit/src/client/client.ts返回{ output, stream, streamId }三个成员其中streamId通过读取响应头x-genkit-stream-id解析。其底层__flowRunEnvelopeclient.ts展示了与服务端对应的完整协议客户端总是POST请求体为JSON.stringify({ data: input, ...(init ! undefined { init }) })请求头固定携带Accept: text/event-stream与Content-Type: application/json自定义 headers 追加其后重连时还会带上x-genkit-stream-id头响应为 SSE 流帧格式为data: json按\n\n分隔message键表示流式 chunkresult键表示最终输出error键表示错误携带status/message/detailsHTTP 204 视为流不存在非 200 状态码会抛出带服务端响应文本的错误。测试中还验证了 Accept 头的健壮性即使客户端或代理发送混合大小写的多值 Accept如Text/Event-Stream, */*服务端也应判定为流式请求并返回text/event-stream响应见 js/plugins/fetch/tests/web_test.ts。Initialization DatainitSchema如果 Flow 或 Action 通过initSchema声明了初始化数据例如会话 ID、模型配置客户端可在调用时通过init字段传入const result await runFlow({ url: http://localhost:3780/api/myFlow, input: say hello, init: { sessionId: abc123, config: { temperature: 0.7 } }, }); // 同样支持流式 const streamed streamFlow({ url: http://localhost:3780/api/myFlow, input: say hello, init: { sessionId: abc123 }, }); for await (const chunk of streamed.stream) { console.log(chunk); } console.log(await streamed.output);协议层面的约定init与data一起放在请求体中{ data: ..., init: ... }服务端从请求体提取见 index.ts服务端在 Flow 运行前会依据 Action 的initSchema校验init不符合 schema 时请求会在 Flow 执行前失败返回400 INVALID_ARGUMENT测试 js/plugins/fetch/tests/web_test.ts 验证了合法init会被透传给 Action 的执行上下文{ init }而非法类型如sessionId传数字会被拒绝并返回 400。API 速查表Export说明fetchHandler(action, options?)为单个 ActionFlow、模型等返回(request) PromiseResponse处理器fetchHandlers(actions, pathPrefix?)返回一个按路径分发到多个 Action 的处理器withActionOptions(action, options)用contextProvider、streamManager或自定义path包装 ActionActionWithOptionsAction options 的类型FetchHandlerOptionsfetchHandler的 options 类型contextProvider、streamManager请求 / 响应协议速查请求体必须是 JSON且包含data字段{ data: input }init为可选字段触发流式请求头Accept: text/event-stream或查询参数?streamtrue流式响应头Content-Type: text/event-stream、Cache-Control: no-cache、Connection: keep-alive、Transfer-Encoding: chunked启用 Durable Streaming 时额外返回x-genkit-stream-id非流式响应200{ result: output }并携带x-genkit-trace-id与x-genkit-span-id两个遥测响应头见 index.ts测试见 web_test.ts错误响应状态码由错误映射而来如PERMISSION_DENIED→ 403、INVALID_ARGUMENT→ 400响应体为getCallableJSON序列化后的错误对象请求体不是合法 JSON、或缺少data字段时固定返回400 INVALID_ARGUMENT见 index.ts。调试与测试插件自带一套不依赖任何 Web 框架的端到端测试位于 js/plugins/fetch/tests/web_test.ts。其巧妙之处在于用 Node 原生http模块起一个服务通过辅助函数把IncomingMessage转换为标准Request、把Response写回ServerResponse从而在纯 Node 环境下完整演练fetchHandler/fetchHandlers的全部行为见 web_test.ts。运行测试cd js/plugins/fetch npm test测试脚本定义在 package.json 的test字段基于 Node 内置 test runner tsx。测试覆盖的典型场景包括void/string/object 三种输入形态的 Flow、带 schema 的输入校验、鉴权成功与失败、init透传与initSchema校验、单 Action 与多 Action 路径路由、普通流式与 Durable Streaming 重连、直接调用模型等。这些用例既是行为规范也是读者理解该插件协议语义的最佳注释。总结genkit-ai/fetch用极薄的适配层把 Genkit 的 Action 体系无缝接入整个 Web Fetch 生态fetchHandler解决单个 Action 一个端点fetchHandlers解决一批 Action 按路径分发ContextProvider解决跨框架统一的请求上下文与鉴权StreamManager解决流式输出的持久化与断线重连。其Express 风格、先传 Action 再消费请求的 API 设计加上仅依赖标准 Web API 的约束使其成为在边缘计算、Serverless 与轻量运行时中部署 Genkit 应用的通用方案。更深层的协议细节SSE 帧格式、错误映射、遥测头、init 校验都可以在上述源码与测试文件中逐一验证供你在实际集成时按需对照。【免费下载链接】genkitOpen-source framework for building agentic apps in JavaScript, Go, Dart, and Python, built and used in production by Google项目地址: https://gitcode.com/GitHub_Trending/ge/genkit创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考