Genkit FastAPI 插件实战：把 Python Flow 与 Agent 变成标准 HTTP 端点

Genkit FastAPI 插件实战把 Python Flow 与 Agent 变成标准 HTTP 端点【免费下载链接】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 官方 Python SDK 中的genkit-fastapi插件文档讲解如何把 Genkit 的 Flow 和 Agent 一键挂载到已有的 FastAPI 应用上包括serve_flow/serve_agent两种挂载方式、自定义路由路径与 FastAPI 依赖注入、genkit_fastapi_handler装饰器、SSE 流式输出协议以及用genkit start启动 Dev UI 的完整运行流程。读完后你可以直接在自己的 FastAPI 服务中暴露 Genkit 能力并理解底层请求/响应线格式wire format与源码实现细节。插件定位与安装genkit-fastapi是 Genkit Python SDK 的 FastAPI 集成插件官方描述为Hang flows and agents on the FastAPI app you already have把 Flow 和 Agent 挂到你已有的 FastAPI 应用上。它把 Genkit 的 flow / agent 包装成符合 Genkit 统一 HTTP 协议的APIRouter让你无需手写请求解析和 SSE 格式化逻辑。安装方式使用 uvuv add genkit-fastapi genkit-google-genai从 pyproject.toml 可以看到该包的实现事实当前版本0.11.0开发状态为BetaDevelopment Status :: 4 - Beta要求 Python3.10声明支持 3.10 ~ 3.14运行时依赖为genkit、pydantic2.10.5、fastapi0.100.0。插件入口模块 genkit_fastapi 对外导出 5 个符号__all__ [ genkit_fastapi_handler, handle_genkit_request, package_name, serve_agent, serve_flow, ]其中serve_flow、serve_agent是最常用的挂载函数genkit_fastapi_handler用于自定义路由handle_genkit_request则是所有路由共同依赖的底层请求处理器后文会展开。另外模块 docstring 中说明当设置GENKIT_ENVdev时Dev UI 的 reflection 服务会自动在后台线程启动无需手动做 lifespan 接线。用 serve_flow 暴露 Flowserve_flow把单个 flow 注册为一个 FastAPI 端点默认路由路径为/flow 名称这一点可从源码 handler.py 中resolved_base_path f/{flow.name} if base_path is None else base_path得到印证。官方 README 给出的完整示例from fastapi import FastAPI from genkit import Genkit from genkit_fastapi import serve_flow from genkit_google_genai import GoogleAI ai Genkit(plugins[GoogleAI()], modelGoogleAI.gemini_model(gemini-flash-latest)) app FastAPI() ai.tool(descriptionGet current weather for a location) async def get_weather(location: str) - str: return fSunny in {location} ai.flow() async def chat_flow(prompt: str) - str: response await ai.generate( promptprompt, tools[get_weather], ) return response.text # Mount flow endpoint at POST /api/chat_flow app.include_router(serve_flow(chat_flow), prefix/api)这个例子的要点flow 定义与 HTTP 层完全解耦——chat_flow本身不感知 HTTPserve_flow(chat_flow)返回一个APIRouter再通过app.include_router(..., prefix/api)挂载因此最终端点是POST /api/chat_flowget_weather通过ai.tool注册为工具并在ai.generate的tools参数中声明模型可在生成过程中发起工具调用请求体使用 Genkit 统一线格式{data: 输入}响应体为{result: 输出}。仓库中还有一个可直接运行的完整样例 fastapi-bugbot它用asyncio.gather并行发起三次ai.generate安全 / 缺陷 / 风格各一次以 Pydantic 模型Analysis作为output_schema约束结构化输出最终通过app.include_router(serve_flow(review_code, base_path/review))暴露在POST /review上。其调用方式为curl -X POST http://localhost:8080/review \ -H Content-Type: application/json \ -d {data:{code:eval(user_input),language:python}}可以看到请求体中data字段内嵌的正是 flow 入参对象CodeInput的code与language字段这正是 Genkit 线格式“输入包在data里”的直观体现。用 serve_agent 暴露 Agent含 getSnapshot / abortAgent 相比 flow 带有会话session语义serve_agent在挂载 agent 的 turn对话轮次路由之外还会附加快照与中止端点。README 示例from fastapi import FastAPI from genkit.exp import Genkit from genkit_fastapi import serve_agent from genkit_google_genai import GoogleAI ai Genkit(plugins[GoogleAI()], modelGoogleAI.gemini_model(gemini-flash-latest)) app FastAPI() ai.tool(descriptionGet current weather for a location) async def get_weather(location: str) - str: return fSunny in {location} weather_agent ai.define_agent( nameweatherAgent, modelGoogleAI.gemini_model(gemini-flash-latest), systemYou are a helpful weather assistant., tools[get_weather], ) # Mount agent turn route at POST /api/weatherAgent (plus /getSnapshot and /abort companion endpoints) app.include_router(serve_agent(weather_agent), prefix/api)注意这里Genkit从genkit.exp导入——agent 能力当前位于 Python SDK 的实验性子模块中genkit.exp与测试文件 agent_handler_test.py 开头pytest.importorskip(genkit.exp.agent)的防御性导入一致。从 serve_agent 源码 可以确认端点挂载的精确行为默认路径同样是/agent 名称本例即POST /api/weatherAgent这是 agent 的 turn 路由仅当agent.store is not None即启用了会话状态存储时才会额外注册两个伴生端点POST /api/weatherAgent/getSnapshot通过snapshotId或snapshot_id/sessionId或session_id二选一查询会话快照查不到时返回 404测试test_get_snapshot_missing_returns_404验证了这一点POST /api/weatherAgent/abort按snapshotId中止一次运行中的 agent返回{snapshotId: ..., status: ...}context_dependency如果有会同时应用到 turn、getSnapshot、abort 三条路由上即鉴权依赖对整个 agent 作用域生效。getSnapshot的入参解析由 parse_snapshot_lookup_input 实现payload 可以是裸字符串直接作为 snapshot ID也可以是 dictdict 中snapshotId/sessionId恰好必须提供一个两者同时给或都不给都会抛INVALID_ARGUMENT错误。请求线格式data / input / message 与 session initserve_flow/serve_agent之下的所有路由共用一个请求处理器handle_genkit_request。从 extract_action_input 的源码可以完整梳理出客户端可发送的多种请求体形态请求体形态处理逻辑{data: ...}标准线格式data即 action 输入{input: ...}等价写法input即 action 输入{message: ...}便捷写法自动包装为{message: {role: user, content: [{text: ...}]}}的对话消息结构{snapshotId/sessionId: ...}透传整个 body供 getSnapshot / abort 使用{}空对象视为无输入data缺省与 JS 版 Express 插件行为对齐允许 void flow 正常执行其他未知键抛INVALID_ARGUMENTHTTP 400测试 fastapi_test.py 对以上行为逐一做了断言test_void_flow_accepts_empty_body空 body 运行无参 flow 返回 200、test_unknown_body_shape_still_returns_400未知键返回 400 且响应体含message/status/details三个字段。对于 agent会话身份可以通过init字段传递resolve_session_init 还支持从查询参数?session_id或?thread_id中注入会话 ID前提是 body 里的init未显式指定方便 URL 直接携带会话标识。非流式响应的结构固定为{result: ...}action 内部抛出的异常会转成 HTTP 500并以{message, status, details}的标准 JSON 错误格式返回见 json_error_response 与测试test_500_flow_exception_returns_valid_json。自定义 base_path 与 FastAPI 依赖注入serve_flow/serve_agent都接受两个关键字参数base_path覆盖默认的/名称路由路径和context_dependency一个 FastAPI 依赖其解析结果作为 action 的context传入。README 示例from fastapi import Depends, Header from genkit_fastapi import serve_flow async def user_context(authorization: str Header(...)): return {uid: parse_token(authorization)} app.include_router( serve_flow( chat_flow, base_path/chat, context_dependencyuser_context, ), prefix/api, )工作机制在 _mount_action 中当提供了context_dependency时生成的路由函数签名会携带context: Any Depends(context_dependency)FastAPI 会按正常依赖图解析它包括其自身的子依赖、security scheme 等解析得到的 dict 随后原样作为context传给 actionaction 内部即可通过ctx: ActionRunContext读取ctx.context。测试 test_context_dependency_value_reaches_action 验证了整条链路带子依赖的user_context依赖解析出{uid: user-123}flow 中通过ctx.context.get(uid)成功读取。另外 agent_handler_test.py 中的test_context_dependency_gates_the_turn还验证了一个安全语义context_dependency抛出HTTPException(401)时agent turn 在开始流式执行前即被拦截可直接实现“鉴权不过不产生任何模型调用”的网关效果。装饰器方式genkit_fastapi_handler如果你需要在完全自定义的路由上运行 Genkit action可以叠加genkit_fastapi_handler(ai)装饰器放在app.post(...)之下、ai.flow()之上装饰器自下而上读取最内层是 flow 定义from genkit_fastapi import genkit_fastapi_handler app.post(/custom-chat, response_modelNone) genkit_fastapi_handler(ai) ai.flow() async def custom_chat(prompt: str) - str: response await ai.generate( promptprompt, tools[get_weather], ) return response.text从 genkit_fastapi_handler 源码 可确认其支持两种被装饰对象直接装饰一个 Action如上例ai.flow()返回的就是 Action装饰一个 async 包装函数该函数返回 Action——适用于 flow 在别处定义、稍后注入的场景源码会显式检查包装函数必须是 async否则抛INVALID_ARGUMENT这也是 docstring 中示例要求async def chat(): return my_flow的原因。装饰器还支持可选的context_provider参数与context_dependency走 FastAPI 依赖图不同context_provider接收一个RequestData由 FastAPIRequestData 封装的 method / headers / input可同步或异步地返回 context dict。源码注释明确了两者的分工装饰器路线自己从请求读取 context而需要 FastAPI 完整依赖图鉴权方案、数据库会话的场景应改用serve_flow/serve_agent的context_dependency。还有一种“逃生舱口”如果你想要最大控制权可以写任意app.post端点、使用任意Depends(...)参数自己构造context与init然后直接调用handle_genkit_request(request, action..., context..., init...)获得标准 Genkit 线格式响应而不用重新实现协议细节见其 docstring。运行Dev UI 与生产模式README 给出两种运行方式# With Genkit Dev UI genkit start -- uvicorn main:app --reload # Production (no Dev UI) uvicorn main:appgenkit start --之后的参数会原样透传给 uvicorn同时由 CLI 拉起 Genkit Dev UI用于在开发期可视化调用链与 flow 执行生产环境直接用uvicorn main:app即可插件不强制任何额外服务。流式输出SSE端点自动支持流式客户端只要发送Accept: text/event-stream请求头或追加查询参数?streamtrue响应即切换为 SSE 流判定逻辑见 wants_stream。curl -X POST http://localhost:8000/api/chat_flow \ -H Content-Type: application/json \ -H Accept: text/event-stream \ -d {data: Tell me a joke}SSE 帧格式在源码中固定为三种事件format_stream_chunk / format_stream_result / format_stream_error且 JSON 使用紧凑分隔符以减小线上体积流式块data: {message: chunk}\n\n——action 流中的每个 chunk最终结果data: {result: response}\n\n——流结束后补发的一条终结事件流内错误data: {error: {message, status, details}}\n\n——异常时以 SSE 事件形式下发而非断开连接。测试 agent_handler_test.py 的test_turn_streams_sse_and_final_result验证了 agent turn 的完整流式行为POST /api/chat?streamtrue返回text/event-stream响应头事件序列以含result字段的最终事件收尾。小结genkit-fastapi的核心价值在于“零协议代码”serve_flow/serve_agent两行代码即可把 Genkit 能力接入既有 FastAPI 路由体系同时保留 FastAPI 原生的 prefix、依赖注入与 security 能力SSE 流式、统一错误格式、会话快照/中止端点全部由插件按 Genkit 统一线格式实现实现集中在 handler.py。需要注意的前提agent 相关 APIdefine_agent、serve_agent的 getSnapshot/abort依赖genkit.exp实验子模块与会话存储配置插件本身处于 Beta 状态生产使用前建议关注genkit-fastapi的版本变更。【免费下载链接】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），仅供参考