企业官网建设流程全解析

openai-agents-python MCP 工具过滤实战：用静态白名单/黑名单限制 Filesystem Server 暴露的工具

【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python

导读

本文基于 openai-agents-python 仓库中的 MCP Tool Filter 示例（examples/mcp/tool_filter_example/README.md），完整讲解如何在多 Agent 工作流中接入基于 stdio 传输的 MCP 文件系统服务器，通过create_static_tool_filter静态工具过滤器只向模型暴露指定的安全工具，并配合require_approval="always"与代码内自动批准机制实践 Human-in-the-Loop（HITL）审批流程。读完本文，你将掌握 MCP 服务器的启动参数、静态工具过滤的底层实现原理、拦截被屏蔽工具的验证方法，以及一套可直接复制运行的最小可运行示例。

一、示例概览：这个示例要解决什么问题

该示例是 JS 版examples/mcp/tool-filter-example.ts的 Python 移植，聚焦四个目标：

1. 通过npx在本地启动官方 filesystem MCP 服务器（@modelcontextprotocol/server-filesystem）；
2. 应用静态工具过滤器，只允许read_file、list_directory两个只读工具暴露给模型；
3. 通过实际对话验证被屏蔽的写工具（write_file）确实不可用；
4. 开启require_approval="always"审批策略，并在代码中自动批准所有中断（interruption），从而跑通 HITL 审批路径。

示例目录结构如下（见 examples/mcp/tool_filter_example）：

examples/mcp/tool_filter_example/ ├── README.md # 示例说明 ├── main.py # 可运行的主程序 └── sample_files/ ├── books.txt # 供 filesystem 服务器读取的样例文件 └── favorite_songs.txt

二、运行方式与前置条件

直接运行：

uv run python examples/mcp/tool_filter_example/main.py

前置条件有两个：

• npx必须位于PATH中（示例启动时也会用shutil.which("npx")做显式检查，缺失时抛出RuntimeError，提示先执行npm install -g npx）；
• 必须设置OPENAI_API_KEY环境变量，供模型调用使用。

示例运行时会打印 Trace 链接（基于gen_trace_id()生成的trace_id），方便在 platform.openai.com/logs/trace 上回放整个 MCP 工具调用过程。

三、逐段解析 main.py

3.1 自动批准工具调用的辅助函数

async def run_with_auto_approval(agent: Agent[Any], message: str) -> str | None: """Run and auto-approve interruptions.""" result = await Runner.run(agent, message) while result.interruptions: state = result.to_state() for interruption in result.interruptions: print(f"Approving a tool call... (name: {interruption.name})") state.approve(interruption, always_approve=True) result = await Runner.run(agent, state) return cast(str | None, result.final_output)

这是 HITL 路径的核心循环：

• Runner.run返回结果后，若存在interruptions（即工具调用等待人工审批），则进入循环；
• 通过result.to_state()把运行状态序列化为可恢复的RunState；
• 遍历每个interruption，打印工具名并调用state.approve(interruption, always_approve=True)自动批准（always_approve=True表示对本次工具调用授予永久批准，后续同工具不再中断）；
• 用更新后的state重新Runner.run，直至没有任何中断，最后返回final_output。

由于示例配置了require_approval="always"，每次工具调用都会产生中断，这个循环正是为了在代码中自动完成审批，从而在不引入真实人工交互的情况下验证审批链路是否工作。

3.2 启动带过滤与审批策略的 MCP 服务器

async with MCPServerStdio( name="Filesystem Server with filter", params={ "command": "npx", "args": ["-y", "@modelcontextprotocol/server-filesystem", samples_dir], "cwd": samples_dir, }, require_approval="always", tool_filter=create_static_tool_filter( allowed_tool_names=["read_file", "list_directory"], blocked_tool_names=["write_file"], ), ) as server:

逐项说明：

• MCPServerStdio：基于 stdio 传输的 MCP 服务器实现（定义于 src/agents/mcp/server.py），通过子进程标准输入/输出与 MCP 服务器通信；
• params：MCPServerStdioParamsTypedDict，镜像mcp.client.stdio.StdioServerParameters，支持command、args、env、cwd、encoding等字段（见 server.py）。此处command="npx"、args=["-y", "@modelcontextprotocol/server-filesystem", samples_dir]表示临时下载并启动官方 filesystem 服务器，并以其工作目录（cwd=samples_dir）作为可访问根目录；
• require_approval="always"：审批策略，使服务器上所有工具调用都需要批准。除字符串外还支持"never"、按工具名的字典映射，以及带 always/never 工具列表的对象（见 server.py）；
• tool_filter=create_static_tool_filter(...)：静态工具过滤器，白名单放行read_file、list_directory，黑名单剔除write_file。

MCPServerStdio还提供其他可配置项，例如cache_tools_list（缓存工具列表避免每次往返服务器，显著降低延迟）、client_session_timeout_seconds（ClientSession 读超时，默认 5 秒）、max_retry_attempts与retry_backoff_seconds_base（list_tools/call_tool 失败重试与指数退避）、use_structured_content（是否直接使用tool_result.structured_content）、tool_input_guardrails/tool_output_guardrails（服务器级工具守卫）等，均可按需组合使用。

3.3 绑定 MCP 服务器的 Agent

agent = Agent( name="MCP Assistant", instructions=( "Use only the available filesystem tools. " "All file paths should be absolute paths inside the allowed directory. " "If a user asks for an action that requires an unavailable tool, " "explicitly explain that it is blocked by the tool filter." ), mcp_servers=[server], )

• mcp_servers=[server]把上面创建的 MCP 服务器挂载到 Agent 上，模型即可调用服务器暴露的工具；
• instructions引导模型：只使用可用工具、路径必须是允许目录内的绝对路径；当用户请求需要被过滤工具时，明确说明该操作被工具过滤器屏蔽。这是让“拦截行为可观察”的关键提示词设计。

3.4 两轮验证对话

trace_id = gen_trace_id() with trace(workflow_name="MCP Tool Filter Example", trace_id=trace_id): print(f"View trace: https://platform.openai.com/logs/trace?trace_id={trace_id}\n") result = await run_with_auto_approval( agent, f"List the files in this allowed directory: {samples_dir}" ) print(result) blocked_result = await run_with_auto_approval( agent, ( f'Create a file at "{target_path}" with the text "hello". ' "If you cannot, explain that write operations are blocked by the tool filter." ), ) print("\nAttempting to write a file (should be blocked):") print(blocked_result)

• gen_trace_id()生成全局唯一追踪 ID，trace(workflow_name=..., trace_id=...)包裹整个工作流以进行端到端追踪；
• 第一轮：要求列出允许目录下的文件——list_directory在白名单内，工具可用，模型应能成功返回sample_files下的books.txt、favorite_songs.txt等文件；
• 第二轮：要求向target_path（即sample_files/test.txt）写入文本——write_file已被过滤，工具不存在，模型应当返回“写入操作被工具过滤器屏蔽”的说明。目标路径特意放在服务器根目录下，确保失败原因确实是过滤而非路径权限。

四、深入底层：create_static_tool_filter 与过滤执行原理

4.1 静态过滤器构造器

create_static_tool_filter定义于 src/agents/mcp/util.py，签名如下：

def create_static_tool_filter( allowed_tool_names: list[str] | None = None, blocked_tool_names: list[str] | None = None, ) -> ToolFilterStatic | None:

其行为：

• 当allowed_tool_names与blocked_tool_names都为None时返回None，表示不过滤；
• 否则构造并返回ToolFilterStatic字典（仅包含显式提供的键）。

ToolFilterStatic是一个 TypedDict（见 util.py）：

class ToolFilterStatic(TypedDict): allowed_tool_names: NotRequired[list[str]] # 白名单：仅这些工具可用 blocked_tool_names: NotRequired[list[str]] # 黑名单：这些工具被过滤掉

4.2 静态过滤的判定顺序

在 src/agents/mcp/server.py 中，_apply_tool_filter会先判断tool_filter的类型：

• 是dict（即ToolFilterStatic）→ 走_apply_static_tool_filter；
• 是可调用对象（ToolFilterCallable）→ 走_apply_dynamic_tool_filter，实现基于RunContextWrapper、Agent 与服务器名的动态过滤，返回True保留、False剔除，且支持同步/异步函数；过滤函数抛异常时该工具会被默认剔除以保证安全。

_apply_static_tool_filter（server.py）的执行顺序是：

1. 若存在allowed_tool_names，先做白名单过滤：filtered_tools = [t for t in filtered_tools if t.name in allowed_names]；
2. 若存在blocked_tool_names，再对剩余集合做黑名单剔除：filtered_tools = [t for t in filtered_tools if t.name not in blocked_names]。

因此当白名单与黑名单同时给出时，先应用白名单、再剔除黑名单中的工具。这与 docs/mcp.md 的说明一致：同时提供allowed_tool_names和blocked_tool_names时，SDK 先应用白名单，再从剩余工具中移除黑名单项。过滤发生在工具列表暴露给模型之前（见 server.py 的get_tools调用链），被过滤的工具对模型完全不可见，因此模型不会尝试调用它——第二轮对话中模型“解释写入被屏蔽”的行为正是这一机制的外在表现。

4.3 静态与动态过滤的取舍

• 静态过滤（本示例采用）：声明式、零开销、可静态分析，适合“工具集固定、安全策略稳定”的场景；
• 动态过滤（ToolFilterCallable，签名见 util.py）：接收ToolFilterContext（含run_context、agent、server_name）和待判定工具，可在每次获取工具列表时按运行上下文做细粒度决策，适合“同一服务器在不同 Agent 或不同会话中暴露不同工具”的场景。

五、require_approval 与 HITL 审批路径

require_approval="always"使服务器上的每次工具调用都进入审批流程，产生interruptions。示例的run_with_auto_approval展示了标准的 HITL 处理模式：

1. 首次Runner.run返回带interruptions的结果；
2. result.to_state()导出可恢复状态；
3. 遍历interruptions，用state.approve(interruption, always_approve=True)批准；
4. 以新状态重新运行，直到无中断。

在生产场景中，将state.approve(...)替换为真实的人工确认（如文件审批、Webhook、聊天确认）即可把示例无缝改造成带人工审批的 MCP 工具调用流水线。仓库中还提供了更完整的 HITL 会话示例（如 examples/memory/file_hitl_example.py、examples/memory/memory_session_hitl_example.py）可供参考。

六、预期输出与结果验证

运行成功时，第一轮应看到模型返回sample_files目录中的文件清单；第二轮打印：

Attempting to write a file (should be blocked): <模型说明：写入操作被工具过滤器屏蔽的文本>

你还可以在 Trace 页面确认两轮对话的工具列表：第一轮可见read_file/list_directory调用，第二轮模型未产生write_file调用——这正是“过滤发生在工具暴露之前”的直接证据。若要进一步验证过滤效果，可修改 main.py 中的allowed_tool_names/blocked_tool_names列表后重跑，例如把list_directory也加入黑名单，观察模型在第一轮是否还能列出目录。

七、小结

通过本示例你可以掌握 openai-agents-python 中 MCP 集成的三个关键能力：

• 接入：用MCPServerStdio+npx快速拉起任意 stdio 型 MCP 服务器（filesystem 只是其中一种）；
• 收敛：用create_static_tool_filter以白名单/黑名单方式把工具面收敛到最小必要集合，从源头杜绝模型调用危险工具（先白名单、后黑名单的判定顺序见 server.py）；
• 管控：用require_approval把工具调用纳入审批策略，配合RunState.approve实现自动或人工的 HITL 审批，让多 Agent 工作流中的外部工具调用始终处于可控范围。

这套“MCP 服务器 + 静态过滤 + 审批策略”的组合，是构建安全、可审计的工具调用链路的通用范式，可直接迁移到数据库访问、代码执行、文件操作等任何基于 MCP 的工具集成场景。

【免费下载链接】openai-agents-pythonA lightweight, powerful framework for multi-agent workflows项目地址: https://gitcode.com/GitHub_Trending/op/openai-agents-python