Genkit Python 中间件实战用 ToolApproval、Skills 与 Filesystem 构建带人工审批的沙箱编码代理【免费下载链接】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 示例py/samples/middleware-coding-agent展开演示如何用genkit-middleware插件的三件中间件——ToolApproval、Skills、Filesystem——在约 100 行代码内搭出一个能读写沙箱workspace/目录、写操作前暂停等待 y/N 人工确认、并能按需加载SKILL.md技能库的编码代理coding agent。读完后你可以掌握Genkit Python 中间件的安装与注册方式、use[...]参数中多个中间件的组合顺序、response.interruptsrestart_tool的人工审批闭环以及沙箱文件系统工具的路径逃逸防护与错误自纠正机制。示例定位与目录结构py/samples/middleware-coding-agent/README.md对示例的概括非常精炼The model reads and writes a sandboxedworkspace/. Writes pause for y/N. Skills load fromskills/.即模型可以读写一个受沙箱限制的workspace/目录所有写操作都会暂停等待用户输入 y/N 决定批准与否技能skills从skills/目录加载。示例的实际目录结构为py/samples/middleware-coding-agent/ ├── README.md ├── pyproject.toml ├── skills/ │ ├── python-expert/ │ │ └── SKILL.md # Python 编码规范技能 │ └── test-writer/ │ └── SKILL.md # pytest 测试编写技能 └── src/ └── main.py # 编码代理入口workspace/目录在运行时由main.py自动创建workspace.mkdir(parentsTrue, exist_okTrue)首次运行前不存在属正常现象。环境准备与运行方式pyproject.toml 声明了示例的全部依赖[project] name middleware-coding-agent version 0.1.0 requires-python 3.10 dependencies [ genkit, genkit-google-genai, genkit-middleware, pydantic2.10.5, structlog25.2.0, ]运行前提与步骤来自 READMEexport GEMINI_API_KEYyour-api-key uv sync uv run src/main.py需要 Python 3.10genkit-google-genai提供模型插件GoogleAI因此必须配置GEMINI_API_KEYgenkit-middleware即仓库内的 genkit-middleware 包提供Retry、Fallback、ToolApproval、Skills、Filesystem、Artifacts六类中间件见 插件注册文件运行后进入交互终端输入自然语言请求输入exit退出。核心实现逐段解析1. Genkit 实例与插件注册main.py 的前置部分here Path(__file__).resolve().parent.parent workspace here / workspace skills here / skills ai Genkit( plugins[GoogleAI(), Middleware()], modelGoogleAI.gemini_model(gemini-flash-latest), )workspace与skills路径都锚定在示例根目录src/的上一级因此沙箱与技能目录始终位于示例内部不会误伤外部文件Middleware()插件负责把ToolApproval、Skills、Filesystem等中间件描述符注册到运行时注册逻辑见init.py使它们可用于use[...]并能在 Dev UI 中展示默认模型为gemini-flash-latest通过GoogleAI.gemini_model(...)创建模型引用。2. 三个中间件实例的组合# ToolApproval lets read-only tools run; write_file / edit_file interrupt. middleware [ ToolApproval(allowed_tools[read_file, list_files, use_skill]), Skills(skill_paths[str(skills)]), Filesystem(root_dirstr(workspace), allow_write_accessTrue), ]三个中间件各司其职中间件关键配置职责ToolApprovalallowed_tools[read_file, list_files, use_skill]白名单内的工具直接放行其余工具即write_file/edit_file触发审批中断Skillsskill_paths[str(skills)]扫描skills/*/SKILL.md把技能清单注入系统提示并提供use_skill工具按需加载全文Filesystemroot_dirstr(workspace),allow_write_accessTrue提供list_files/read_file/write_file/edit_file四个沙箱文件工具注意ToolApproval的allowed_tools中包含了use_skill——这个工具由Skills中间件贡献把它加入白名单意味着“加载技能”属于无副作用的只读行为不需要人工确认而真正改文件的write_file/edit_file不在名单里必然走审批流程。3. 系统提示与代理主循环async def main() - None: workspace.mkdir(parentsTrue, exist_okTrue) messages [ Message( roleRole.SYSTEM, content[ Part( rootTextPart( text( fYou are a coding agent. Working directory is {workspace}. Use plain filenames relative to that root. Read a file before you edit it. Start by listing the workspace. ) ) ) ], ), ] print(Type a request. exit to quit.) while True: ...系统提示见 main.py明确约束了模型的行为模式工作目录就是沙箱根、使用相对于该根的普通文件名、编辑前先读文件、首轮先列目录。主循环的核心是“生成 → 检查中断 → 审批 → 续跑”的迭代过程restart None prompt user_input while True: response await ai.generate( promptprompt, messagesmessages, resume_restartrestart, max_turns20, usemiddleware, ) messages response.messages if not response.interrupts: print(response.text) break # Each interrupt is a write. Approve restarts that tool. approved [] for interrupt in response.interrupts: print(f{interrupt.tool_request.name}: {interrupt.tool_request.input}) if input(Approve? (y/N): ).strip().lower() in (y, yes): approved.append(restart_tool(interruptinterrupt, resumed_metadata{tool_approved: True})) if not approved: print(Denied.) break restart approved prompt None这段循环见 main.py体现了 Genkit Python 的工具中断interrupt协议max_turns20限制模型与工具交替执行的最大轮数防止代理无限循环烧 tokenresume_restartrestart把上一轮批准的重启请求传给引擎让被中断的工具带着审批元数据重新执行response.interrupts非空时循环打印每个中断对应的工具名与完整入参用户能看清模型准备写的文件路径和内容逐个询问 y/N用户答 y 时restart_tool(interruptinterrupt, resumed_metadata{tool_approved: True})构造一个新的ToolRequestPart——其metadata[resumed]中携带tool_approved: True全部拒绝则本轮会话终止prompt None续跑时不再注入新用户文本只靠历史消息推进。restart_tool的实现在 genkit/_ai/_tools.py它把resumed_metadata写入返回 part 的metadata[resumed]供下一轮工具执行时读取。源码级原理三个中间件如何协作Filesystem沙箱边界与错误自纠正Filesystem 通过FilesystemConfig约束行为class FilesystemConfig(PydanticBaseModel): Sandbox root and write/tool naming options. root_dir: str allow_write_access: bool False tool_name_prefix: str root_dir为必填项构造时校验非空即沙箱根目录allow_write_access默认False——不开启时只注册list_files和read_file示例中显式传了True才有write_file/edit_file见 tools() 方法tool_name_prefix支持给工具名加前缀便于多实例共存。路径逃逸防护是沙箱的核心。每个工具调用都经过_resolve_safe见 _filesystem.pydef _resolve_safe(self, rel: str) - str: Resolve rel to an absolute path, raising ValueError if it escapes root. rel rel.strip().lstrip(/).lstrip(\\) ... candidate os.path.realpath(os.path.join(self._root_abs, rel)) ... if c_norm ! root_norm and not c_norm.startswith(root_norm os.sep): raise ValueError(fPath {rel!r} escapes the root directory.)它对相对路径做os.path.realpath展开符号链接与..后用前缀匹配校验结果是否仍在根目录内../../etc/passwd之类的越界请求会直接抛错。读取限制与内容入队read_file有 10 MB 文件上限和 256 KB 单片上限_MAX_FILE_SIZE_BYTES/_MAX_READ_SLICE_BYTES支持offset/limit分页读取。更微妙的是文件内容并不作为工具返回值而是被“排队”为一条用户消息——wrap_generate 在下一轮模型调用前把这些排队消息注入请求。这样做让工具响应本身保持很短避免把大段文件内容挤占在 tool response 位置。错误自纠正wrap_tool 捕获文件工具抛出的异常但放行Interrupt保证审批中断不受影响把Tool xxx failed: ...也作为用户消息入队模型下一轮就能看到错误并自行修正——例如old_string未命中时edit_file会报错模型可重新读取文件再试。ToolApproval白名单放行 中断审批ToolApproval.wrap_tool 的逻辑只有三步if tool_name in self.config.allowed_tools: return await next_fn(params, ctx) metadata params.tool_request_part.metadata or {} resumed metadata.get(resumed) if isinstance(resumed, dict) and (resumed.get(toolApproved) or resumed.get(tool_approved)): return await next_fn(params, ctx) # ... 记录 span 后 raise Interrupt({message: fTool not in approved list: {tool_name}})工具名在allowed_tools白名单中 → 直接执行工具请求的metadata.resumed里带有tool_approved或toolApproved标记 → 说明这是用户批准后的重启执行放行否则抛出Interrupt生成流程暂停interrupts出现在响应中等待宿主程序处理。示例主循环中restart_tool(..., resumed_metadata{tool_approved: True})写入的正是第 2 步检查的字段二者首尾呼应构成完整的人工审批human-in-the-loop闭环。被拒绝的工具调用则永远不会真正执行。SkillsSKILL.md 技能库与按需加载Skills 中间件 的工作方式扫描_scan_skills遍历skill_paths示例传的是skills目录默认值为[skills]下的每个子目录读取其中的SKILL.md解析 YAML front matter 中的name与description两个字段注入系统提示wrap_generate在每次生成前把技能清单以skills.../skills形式注入系统消息带skills-instructions标记的 part重复注入时原地替换而非叠加提供use_skill工具模型看到描述后调用use_skill(skill_name...)加载技能全文未找到技能时返回可用技能列表模型可据此自我纠正。这种“目录里只放索引、正文按需加载”的设计避免了把所有技能全文塞进系统提示。示例自带两个技能python-expert/SKILL.mdPython 编码规范全量类型注解、优先 dataclass、抛具体异常、注释解释“为什么”、小步聚焦编辑等描述为 “Load whenever you read, edit, or write Python source files”test-writer/SKILL.mdpytest 测试规范一模块一测试文件、test_unit_scenario_expected命名、Arrange/Act/Assert、断言行为而非实现等。front matter 必须是name/description的 YAML 字典否则该技能会被静默跳过解析见 _parse_skill_file——这是编写自定义技能时最容易踩的坑。一次完整交互的执行链路把三层机制串起来用户输入“给 cart.py 加一个 discount 方法”时的链路大致是基于上述源码结构推断Skills.wrap_generate注入技能索引系统提示告知存在python-expert模型调用list_files/read_file——命中ToolApproval白名单直接执行文件内容经Filesystem排队为下轮用户消息模型调用use_skill(skill_namepython-expert)——白名单放行拿到规范全文模型调用edit_file/write_file——不在白名单ToolApproval抛Interruptgenerate返回并带interrupts主循环打印工具名与入参等待 y/N批准后用restart_tool重发wrap_tool检测到resumed.tool_approved放行执行文件落入workspace/沙箱过程中任何工具报错如old_string未命中会被Filesystem.wrap_tool转成用户消息模型在max_turns20的额度内自行重试。小结与可调整点这个示例展示了 Genkit Python 中间件组合的最小完整形态几个值得注意的设计选择安全边界分两层Filesystem负责“写到哪里”沙箱根 逃逸防护ToolApproval负责“什么时候允许写”人工审批两者正交、可独立开关只读先行把read_file/list_files/use_skill放入白名单让模型能自由探索仅在变更动作上设卡兼顾了代理自主性与可控性可调整点把max_turns调大以支持更长的编辑会话给Filesystem传tool_name_prefix以区分多沙箱实例在skills/下新增子目录并写好 front matter 即可扩展技能若希望写操作也免审批把write_file/edit_file加入allowed_tools即可相应地失去人工确认。相关源码入口示例入口 src/main.py、中间件包 py/packages/genkit-middleware、restart_tool/中断协议 genkit/_ai/_tools.py。【免费下载链接】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），仅供参考 Genkit Python 中间件实战：用 ToolApproval、Skills 与 Filesystem 构建带人工审批的沙箱编码代理
Genkit Python 中间件实战用 ToolApproval、Skills 与 Filesystem 构建带人工审批的沙箱编码代理【免费下载链接】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 示例py/samples/middleware-coding-agent展开演示如何用genkit-middleware插件的三件中间件——ToolApproval、Skills、Filesystem——在约 100 行代码内搭出一个能读写沙箱workspace/目录、写操作前暂停等待 y/N 人工确认、并能按需加载SKILL.md技能库的编码代理coding agent。读完后你可以掌握Genkit Python 中间件的安装与注册方式、use[...]参数中多个中间件的组合顺序、response.interruptsrestart_tool的人工审批闭环以及沙箱文件系统工具的路径逃逸防护与错误自纠正机制。示例定位与目录结构py/samples/middleware-coding-agent/README.md对示例的概括非常精炼The model reads and writes a sandboxedworkspace/. Writes pause for y/N. Skills load fromskills/.即模型可以读写一个受沙箱限制的workspace/目录所有写操作都会暂停等待用户输入 y/N 决定批准与否技能skills从skills/目录加载。示例的实际目录结构为py/samples/middleware-coding-agent/ ├── README.md ├── pyproject.toml ├── skills/ │ ├── python-expert/ │ │ └── SKILL.md # Python 编码规范技能 │ └── test-writer/ │ └── SKILL.md # pytest 测试编写技能 └── src/ └── main.py # 编码代理入口workspace/目录在运行时由main.py自动创建workspace.mkdir(parentsTrue, exist_okTrue)首次运行前不存在属正常现象。环境准备与运行方式pyproject.toml 声明了示例的全部依赖[project] name middleware-coding-agent version 0.1.0 requires-python 3.10 dependencies [ genkit, genkit-google-genai, genkit-middleware, pydantic2.10.5, structlog25.2.0, ]运行前提与步骤来自 READMEexport GEMINI_API_KEYyour-api-key uv sync uv run src/main.py需要 Python 3.10genkit-google-genai提供模型插件GoogleAI因此必须配置GEMINI_API_KEYgenkit-middleware即仓库内的 genkit-middleware 包提供Retry、Fallback、ToolApproval、Skills、Filesystem、Artifacts六类中间件见 插件注册文件运行后进入交互终端输入自然语言请求输入exit退出。核心实现逐段解析1. Genkit 实例与插件注册main.py 的前置部分here Path(__file__).resolve().parent.parent workspace here / workspace skills here / skills ai Genkit( plugins[GoogleAI(), Middleware()], modelGoogleAI.gemini_model(gemini-flash-latest), )workspace与skills路径都锚定在示例根目录src/的上一级因此沙箱与技能目录始终位于示例内部不会误伤外部文件Middleware()插件负责把ToolApproval、Skills、Filesystem等中间件描述符注册到运行时注册逻辑见init.py使它们可用于use[...]并能在 Dev UI 中展示默认模型为gemini-flash-latest通过GoogleAI.gemini_model(...)创建模型引用。2. 三个中间件实例的组合# ToolApproval lets read-only tools run; write_file / edit_file interrupt. middleware [ ToolApproval(allowed_tools[read_file, list_files, use_skill]), Skills(skill_paths[str(skills)]), Filesystem(root_dirstr(workspace), allow_write_accessTrue), ]三个中间件各司其职中间件关键配置职责ToolApprovalallowed_tools[read_file, list_files, use_skill]白名单内的工具直接放行其余工具即write_file/edit_file触发审批中断Skillsskill_paths[str(skills)]扫描skills/*/SKILL.md把技能清单注入系统提示并提供use_skill工具按需加载全文Filesystemroot_dirstr(workspace),allow_write_accessTrue提供list_files/read_file/write_file/edit_file四个沙箱文件工具注意ToolApproval的allowed_tools中包含了use_skill——这个工具由Skills中间件贡献把它加入白名单意味着“加载技能”属于无副作用的只读行为不需要人工确认而真正改文件的write_file/edit_file不在名单里必然走审批流程。3. 系统提示与代理主循环async def main() - None: workspace.mkdir(parentsTrue, exist_okTrue) messages [ Message( roleRole.SYSTEM, content[ Part( rootTextPart( text( fYou are a coding agent. Working directory is {workspace}. Use plain filenames relative to that root. Read a file before you edit it. Start by listing the workspace. ) ) ) ], ), ] print(Type a request. exit to quit.) while True: ...系统提示见 main.py明确约束了模型的行为模式工作目录就是沙箱根、使用相对于该根的普通文件名、编辑前先读文件、首轮先列目录。主循环的核心是“生成 → 检查中断 → 审批 → 续跑”的迭代过程restart None prompt user_input while True: response await ai.generate( promptprompt, messagesmessages, resume_restartrestart, max_turns20, usemiddleware, ) messages response.messages if not response.interrupts: print(response.text) break # Each interrupt is a write. Approve restarts that tool. approved [] for interrupt in response.interrupts: print(f{interrupt.tool_request.name}: {interrupt.tool_request.input}) if input(Approve? (y/N): ).strip().lower() in (y, yes): approved.append(restart_tool(interruptinterrupt, resumed_metadata{tool_approved: True})) if not approved: print(Denied.) break restart approved prompt None这段循环见 main.py体现了 Genkit Python 的工具中断interrupt协议max_turns20限制模型与工具交替执行的最大轮数防止代理无限循环烧 tokenresume_restartrestart把上一轮批准的重启请求传给引擎让被中断的工具带着审批元数据重新执行response.interrupts非空时循环打印每个中断对应的工具名与完整入参用户能看清模型准备写的文件路径和内容逐个询问 y/N用户答 y 时restart_tool(interruptinterrupt, resumed_metadata{tool_approved: True})构造一个新的ToolRequestPart——其metadata[resumed]中携带tool_approved: True全部拒绝则本轮会话终止prompt None续跑时不再注入新用户文本只靠历史消息推进。restart_tool的实现在 genkit/_ai/_tools.py它把resumed_metadata写入返回 part 的metadata[resumed]供下一轮工具执行时读取。源码级原理三个中间件如何协作Filesystem沙箱边界与错误自纠正Filesystem 通过FilesystemConfig约束行为class FilesystemConfig(PydanticBaseModel): Sandbox root and write/tool naming options. root_dir: str allow_write_access: bool False tool_name_prefix: str root_dir为必填项构造时校验非空即沙箱根目录allow_write_access默认False——不开启时只注册list_files和read_file示例中显式传了True才有write_file/edit_file见 tools() 方法tool_name_prefix支持给工具名加前缀便于多实例共存。路径逃逸防护是沙箱的核心。每个工具调用都经过_resolve_safe见 _filesystem.pydef _resolve_safe(self, rel: str) - str: Resolve rel to an absolute path, raising ValueError if it escapes root. rel rel.strip().lstrip(/).lstrip(\\) ... candidate os.path.realpath(os.path.join(self._root_abs, rel)) ... if c_norm ! root_norm and not c_norm.startswith(root_norm os.sep): raise ValueError(fPath {rel!r} escapes the root directory.)它对相对路径做os.path.realpath展开符号链接与..后用前缀匹配校验结果是否仍在根目录内../../etc/passwd之类的越界请求会直接抛错。读取限制与内容入队read_file有 10 MB 文件上限和 256 KB 单片上限_MAX_FILE_SIZE_BYTES/_MAX_READ_SLICE_BYTES支持offset/limit分页读取。更微妙的是文件内容并不作为工具返回值而是被“排队”为一条用户消息——wrap_generate 在下一轮模型调用前把这些排队消息注入请求。这样做让工具响应本身保持很短避免把大段文件内容挤占在 tool response 位置。错误自纠正wrap_tool 捕获文件工具抛出的异常但放行Interrupt保证审批中断不受影响把Tool xxx failed: ...也作为用户消息入队模型下一轮就能看到错误并自行修正——例如old_string未命中时edit_file会报错模型可重新读取文件再试。ToolApproval白名单放行 中断审批ToolApproval.wrap_tool 的逻辑只有三步if tool_name in self.config.allowed_tools: return await next_fn(params, ctx) metadata params.tool_request_part.metadata or {} resumed metadata.get(resumed) if isinstance(resumed, dict) and (resumed.get(toolApproved) or resumed.get(tool_approved)): return await next_fn(params, ctx) # ... 记录 span 后 raise Interrupt({message: fTool not in approved list: {tool_name}})工具名在allowed_tools白名单中 → 直接执行工具请求的metadata.resumed里带有tool_approved或toolApproved标记 → 说明这是用户批准后的重启执行放行否则抛出Interrupt生成流程暂停interrupts出现在响应中等待宿主程序处理。示例主循环中restart_tool(..., resumed_metadata{tool_approved: True})写入的正是第 2 步检查的字段二者首尾呼应构成完整的人工审批human-in-the-loop闭环。被拒绝的工具调用则永远不会真正执行。SkillsSKILL.md 技能库与按需加载Skills 中间件 的工作方式扫描_scan_skills遍历skill_paths示例传的是skills目录默认值为[skills]下的每个子目录读取其中的SKILL.md解析 YAML front matter 中的name与description两个字段注入系统提示wrap_generate在每次生成前把技能清单以skills.../skills形式注入系统消息带skills-instructions标记的 part重复注入时原地替换而非叠加提供use_skill工具模型看到描述后调用use_skill(skill_name...)加载技能全文未找到技能时返回可用技能列表模型可据此自我纠正。这种“目录里只放索引、正文按需加载”的设计避免了把所有技能全文塞进系统提示。示例自带两个技能python-expert/SKILL.mdPython 编码规范全量类型注解、优先 dataclass、抛具体异常、注释解释“为什么”、小步聚焦编辑等描述为 “Load whenever you read, edit, or write Python source files”test-writer/SKILL.mdpytest 测试规范一模块一测试文件、test_unit_scenario_expected命名、Arrange/Act/Assert、断言行为而非实现等。front matter 必须是name/description的 YAML 字典否则该技能会被静默跳过解析见 _parse_skill_file——这是编写自定义技能时最容易踩的坑。一次完整交互的执行链路把三层机制串起来用户输入“给 cart.py 加一个 discount 方法”时的链路大致是基于上述源码结构推断Skills.wrap_generate注入技能索引系统提示告知存在python-expert模型调用list_files/read_file——命中ToolApproval白名单直接执行文件内容经Filesystem排队为下轮用户消息模型调用use_skill(skill_namepython-expert)——白名单放行拿到规范全文模型调用edit_file/write_file——不在白名单ToolApproval抛Interruptgenerate返回并带interrupts主循环打印工具名与入参等待 y/N批准后用restart_tool重发wrap_tool检测到resumed.tool_approved放行执行文件落入workspace/沙箱过程中任何工具报错如old_string未命中会被Filesystem.wrap_tool转成用户消息模型在max_turns20的额度内自行重试。小结与可调整点这个示例展示了 Genkit Python 中间件组合的最小完整形态几个值得注意的设计选择安全边界分两层Filesystem负责“写到哪里”沙箱根 逃逸防护ToolApproval负责“什么时候允许写”人工审批两者正交、可独立开关只读先行把read_file/list_files/use_skill放入白名单让模型能自由探索仅在变更动作上设卡兼顾了代理自主性与可控性可调整点把max_turns调大以支持更长的编辑会话给Filesystem传tool_name_prefix以区分多沙箱实例在skills/下新增子目录并写好 front matter 即可扩展技能若希望写操作也免审批把write_file/edit_file加入allowed_tools即可相应地失去人工确认。相关源码入口示例入口 src/main.py、中间件包 py/packages/genkit-middleware、restart_tool/中断协议 genkit/_ai/_tools.py。【免费下载链接】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），仅供参考