企业官网建设流程全解析

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 格式化逻辑。

安装方式（使用 uv）：

uv add genkit-fastapi genkit-google-genai

从 pyproject.toml 可以看到该包的实现事实：

• 当前版本0.11.0，开发状态为Beta（Development Status :: 4 - Beta）；
• 要求 Python>=3.10（声明支持 3.10 ~ 3.14）；
• 运行时依赖为genkit、pydantic>=2.10.5、fastapi>=0.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_ENV=dev时，Dev UI 的 reflection 服务会自动在后台线程启动，无需手动做 lifespan 接线。

用 serve_flow 暴露 Flow

serve_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()], model=GoogleAI.gemini_model('gemini-flash-latest')) app = FastAPI() @ai.tool(description='Get current weather for a location') async def get_weather(location: str) -> str: return f'Sunny in {location}' @ai.flow() async def chat_flow(prompt: str) -> str: response = await ai.generate( prompt=prompt, tools=[get_weather], ) return response.text # Mount flow endpoint at POST /api/chat_flow app.include_router(serve_flow(chat_flow), prefix='/api')

这个例子的要点：

1. flow 定义与 HTTP 层完全解耦——chat_flow本身不感知 HTTP，serve_flow(chat_flow)返回一个APIRouter，再通过app.include_router(..., prefix='/api')挂载，因此最终端点是POST /api/chat_flow；
2. get_weather通过@ai.tool注册为工具，并在ai.generate的tools参数中声明，模型可在生成过程中发起工具调用；
3. 请求体使用 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 / abort）

Agent 相比 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()], model=GoogleAI.gemini_model('gemini-flash-latest')) app = FastAPI() @ai.tool(description='Get current weather for a location') async def get_weather(location: str) -> str: return f'Sunny in {location}' weather_agent = ai.define_agent( name='weatherAgent', model=GoogleAI.gemini_model('gemini-flash-latest'), system='You 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），也可以是 dict；dict 中snapshotId/sessionId恰好必须提供一个，两者同时给或都不给都会抛INVALID_ARGUMENT错误。

请求线格式：data / input / message 与 session init

serve_flow/serve_agent之下的所有路由共用一个请求处理器handle_genkit_request。从 extract_action_input 的源码可以完整梳理出客户端可发送的多种请求体形态：

测试 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_dependency=user_context, ), prefix='/api', )

工作机制在 _mount_action 中：当提供了context_dependency时，生成的路由函数签名会携带context: Any = Depends(context_dependency)，FastAPI 会按正常依赖图解析它（包括其自身的子依赖、security scheme 等），解析得到的 dict 随后原样作为context传给 action，action 内部即可通过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_model=None) @genkit_fastapi_handler(ai) @ai.flow() async def custom_chat(prompt: str) -> str: response = await ai.generate( prompt=prompt, tools=[get_weather], ) return response.text

从 genkit_fastapi_handler 源码 可确认其支持两种被装饰对象：

1. 直接装饰一个 Action（如上例，@ai.flow()返回的就是 Action）；
2. 装饰一个 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:app

• genkit start --之后的参数会原样透传给 uvicorn，同时由 CLI 拉起 Genkit Dev UI，用于在开发期可视化调用链与 flow 执行；
• 生产环境直接用uvicorn main:app即可，插件不强制任何额外服务。

流式输出（SSE）

端点自动支持流式：客户端只要发送Accept: text/event-stream请求头，或追加查询参数?stream=true，响应即切换为 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?stream=true返回text/event-stream响应头，事件序列以含result字段的最终事件收尾。

小结

genkit-fastapi的核心价值在于“零协议代码”：serve_flow/serve_agent两行代码即可把 Genkit 能力接入既有 FastAPI 路由体系，同时保留 FastAPI 原生的 prefix、依赖注入与 security 能力；SSE 流式、统一错误格式、会话快照/中止端点全部由插件按 Genkit 统一线格式实现（实现集中在 handler.py）。需要注意的前提：agent 相关 API（define_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