Agent OS 安全工具开发实战：基于 ATR 构建可复用的 Custom Tools-育师

Agent OS 安全工具开发实战：基于 ATR 构建可复用的 Custom Tools

【免费下载链接】agent-governance-toolkitAI Agent Governance Toolkit — Policy enforcement, zero-trust identity, execution sandboxing, and reliability engineering for autonomous AI agents. Covers 10/10 OWASP Agentic Top 10.项目地址: https://gitcode.com/GitHub_Trending/ag/agent-governance-toolkit

Agent OS（Agent Tool Registry，ATR）为自主智能体提供了一套开箱即用的安全工具集（safe tools），并允许开发者以统一的@tool装饰器编写遵循相同安全原则的自定义工具。本文围绕 custom-tools.md 教程，完整演示如何选用预置的 HTTP、文件、计算、JSON、时间、文本六大安全工具，如何用四种模式创建自定义工具，以及如何将工具注册进 ATR 供智能体发现与调用，并深入源码剖析每一层安全控制（域名白名单、路径沙箱、AST 安全求值、正则保护等）的底层实现。

概述：为什么需要"安全工具"

Agent OS 提供了一批预构建的安全工具（pre-built safe tools），其核心原则是最小权限（principle of least privilege）：能只读就只读、网络操作必须限速、文件访问必须沙箱化、输入必须校验净化、绝不执行 Shell 命令。这批工具位于 atr/tools/safe 包中，包含六个组件：HttpClientTool（带域名白名单的 HTTP 客户端）、FileReaderTool（带路径沙箱的只读文件工具）、JsonParserTool（安全的 JSON/YAML 解析器）、CalculatorTool（不使用 eval() 的安全计算器）、DateTimeTool（时区感知的日期时间工具）以及TextTool（带正则保护的安全文本处理工具）。开发者既可以直接复用它们，也可以遵循同样的安全模式编写自定义工具。

使用预置安全工具

快速开始：一行创建工具包

create_safe_toolkit是安全工具集的工厂函数，位于 toolkit.py，返回一个以工具名为键的字典，并附带register_all注册辅助函数：

from atr.tools.safe import create_safe_toolkit # Create a toolkit with standard tools toolkit = create_safe_toolkit("standard") # Available tools http = toolkit["http"] # HTTP client with rate limiting files = toolkit["files"] # File reader with sandboxing json = toolkit["json"] # JSON/YAML parser calc = toolkit["calculator"] # Safe math operations dt = toolkit["datetime"] # Timezone-aware datetime text = toolkit["text"] # Text processing # Use a tool result = await http.get("https://api.example.com/data")

从源码看，standard预设会为每个工具套用"合理默认值"：HTTP 客户端限速 60 次/分钟、超时 30 秒、最大响应 10 MB；文件读取器默认最大文件 10 MB；JSON 解析器默认 10 MB；计算器精度 15 位；日期时间默认 UTC 时区；文本工具最大长度 1,000,000 字符。所有这些默认值都可通过config字典覆盖。

工具包预设（Toolkit Presets）

工厂函数内置五种预设，分别对应不同的安全暴露面：

# Minimal - no I/O, just processing toolkit = create_safe_toolkit("minimal") # Includes: calculator, datetime, text # Read-only - file access, no network toolkit = create_safe_toolkit("readonly", config={ "sandbox_paths": ["./data"], "allowed_extensions": [".txt", ".json", ".md"] }) # Includes: files, json, calculator, datetime, text # Network - HTTP only, no files toolkit = create_safe_toolkit("network", config={ "allowed_domains": ["api.github.com", "api.openai.com"], "rate_limit": 30 }) # Includes: http # Restricted - all tools with tight limits toolkit = create_safe_toolkit("restricted", config={ "allowed_domains": ["api.internal.com"], "sandbox_paths": ["./safe-data"], "max_file_size": 100_000, "rate_limit": 10 })

对照 toolkit.py 的实现，各预设的精确行为如下：

minimal：仅含计算器、日期时间、文本三个纯处理工具，不含任何 I/O，适合仅做数据加工的场景；precision、timezone、max_text_length可配置。
readonly：文件读取 + JSON 解析 + 计算器 + 日期时间 + 文本，无网络能力；allowed_extensions默认覆盖.txt/.json/.yaml/.yml/.md/.csv/.xml。
network：仅 HTTP 客户端，无文件访问；allowed_domains、blocked_domains、rate_limit（默认 60）、timeout（默认 30）、max_response_size（默认 10 MB）可配置。
restricted：全量工具但收紧限制——HTTP 的allowed_domains与文件的sandbox_paths必须显式指定，限速降到 10 次/分钟，超时 10 秒，响应上限 1 MB，文件大小上限 100 KB，且follow_symlinks=False禁止跟随符号链接；JSON 解析深度限制为 20 层。
standard（默认）：全量工具 + 合理默认值。

toolkit"register_all"会遍历工具实例，将每个带有_tool_metadata属性的可调用方法注册进 ATR 注册表（见 toolkit.py），_preset与_config键则记录了当前预设与配置快照。

单个工具详解

HTTP 客户端：域名白名单与限速

from atr.tools.safe import HttpClientTool http = HttpClientTool( allowed_domains=["api.github.com", "httpbin.org"], rate_limit=30, # requests per minute timeout=10.0, # seconds max_response_size=1_000_000 ) # GET request response = await http.get( "https://api.github.com/users/octocat", headers={"Accept": "application/json"} ) print(response["body"]) # POST request response = await http.post( "https://httpbin.org/post", json_body={"key": "value"} )

其底层实现在 http_client.py，安全控制包含五层：

协议白名单：仅允许http/https，拒绝其他 scheme；
用户信息拦截：拒绝user:pass@host形式的 URL，封堵 SSRF 绕过；
默认黑名单：即使不配置blocked_domains，也会默认屏蔽localhost、127.0.0.1、0.0.0.0、169.254.169.254（AWS 元数据）与metadata.google.internal（GCP 元数据）等地址；
内网地址防护：通过_is_private_domain检查10.x、192.168.x、172.16~31.x、internal、.local等私网模式；
DNS 感知的后缀匹配：白名单匹配采用"精确域名或合法子域名"规则（domain == allowed或domain.endswith("." + allowed)），从源码注释可见这是刻意规避evil-example.com绕过example.com白名单的经典 SSRF/数据外泄漏洞。

限速由RateLimiter滑窗实现（http_client.py），在请求发出前check()并record()；响应体超过max_response_size时会截断并附加... [truncated]标记。该类还提供head方法用于探测资源是否存在。

文件读取器：路径沙箱与目录穿越防护

from atr.tools.safe import FileReaderTool reader = FileReaderTool( sandbox_paths=["./data", "./configs"], allowed_extensions=[".txt", ".json", ".yaml"], max_file_size=1_000_000 ) # Read file result = reader.read_file("./data/config.json") print(result["content"]) # List directory result = reader.list_directory("./data", pattern="*.txt") print(result["files"]) # Check if file exists result = reader.exists("./data/test.txt")

实现位于 file_reader.py，核心是_validate_path：使用Path.resolve()解析路径后，必须能relative_to某个沙箱目录，否则直接抛错；路径字符串中出现..即判定为目录穿越；默认follow_symlinks=False，遇到符号链接直接拒绝。扩展名校验（_validate_extension）实行"黑名单优先、白名单兜底"：默认黑名单涵盖.exe/.dll/.so/.sh/.bash/.py/.pyc/.class/.jar等可执行与脚本文件，若设置了allowed_extensions则只允许列表内的扩展名。

工具提供 5 个方法：read_file（支持max_lines截断）、read_lines（按行号区间读取）、list_directory（支持 glob 模式、递归与隐藏文件开关）、exists、file_info（返回文件元数据）。所有方法都返回包含success、path、size等字段的字典，便于智能体直接消费。

计算器：不使用 eval() 的安全求值

from atr.tools.safe import CalculatorTool calc = CalculatorTool(precision=10) # Evaluate expression (safe - no eval()) result = calc.evaluate("2 + 2 * 3") # 8 # With variables result = calc.evaluate("x * 2 + y", {"x": 5, "y": 10}) # 20 # Math functions result = calc.evaluate("sqrt(16) + sin(0)") # 4.0 # Statistics result = calc.statistics([1, 2, 3, 4, 5]) # {"mean": 3.0, "median": 3.0, "std_dev": 1.58...}

calculator.py 展示了这类工具最关键的实现细节：完全不用eval()/compile()，而是先用ast.parse将表达式解析为 AST，再通过_safe_eval_node递归遍历节点，仅允许白名单内的二元运算符（+ - * / // % ** ^）与一元运算符（正负号），函数调用只允许FUNCTIONS字典中的成员（abs/round/min/max/sum/sqrt/pow/exp/log/sin/cos/tan/...共 24 个），并显式拒绝属性访问（ast.Attribute）。此外还有字符白名单清洗（表达式最长 1000 字符）、max_value=1e308溢出保护、allow_complex复数开关与精度舍入。除evaluate外，还提供add/subtract/multiply/divide/power/sqrt/percentage/statistics等显式算术工具，其中power限制指数绝对值不超过 1000 以防止超大计算。

JSON 解析器：大小、深度与安全 YAML

from atr.tools.safe import JsonParserTool parser = JsonParserTool(max_size=1_000_000, max_depth=50) # Parse JSON result = parser.parse_json('{"key": "value"}') print(result["data"]) # Parse YAML (safe_load) result = parser.parse_yaml("key: value") # Convert to JSON result = parser.to_json({"key": "value"}, indent=2) # Validate schema result = parser.validate_schema( data={"name": "John", "age": 30}, schema={"type": "object", "required": ["name"]} )

json_parser.py 实现三重防护：max_size限制输入字符数防止内存耗尽；_check_depth递归检查嵌套深度与对象键数量（max_depth/max_keys）防止栈溢出；YAML 解析强制使用yaml.safe_load杜绝任意代码执行（parse_yaml的 docstring 明确注明此点）。validate_schema基于jsonschema库校验数据，query支持用点号路径（如users.0.name）安全查询嵌套 JSON。工具还提供to_json/to_yaml序列化方法，输出同样受大小上限约束。

日期时间：时区感知的安全操作

from atr.tools.safe import DateTimeTool dt = DateTimeTool(default_timezone="UTC") # Get current time now = dt.now() print(now["iso"]) # 2024-01-15T10:30:00+00:00 # Parse date result = dt.parse("2024-01-15") # Format date result = dt.format("2024-01-15T10:30:00Z", format="human") # "January 15, 2024 at 10:30 AM" # Add time result = dt.add("2024-01-15T10:30:00Z", days=7, hours=2) # Calculate difference result = dt.diff("2024-01-15", "2024-01-20") # {"days": 5, "total_hours": 120, ...}

datetime_tool.py 内置七种预定义格式（iso、date、time、datetime、human、short、rfc2822），时区解析优先使用标准库zoneinfo.ZoneInfo，兼容+HH:MM偏移量写法，并通过_tz_cache缓存时区对象。add方法对timedelta无法处理的年/月使用calendar.monthrange处理跨月与月末溢出（如 1 月 31 日加一个月自动落到 2 月的有效日期）。除教程演示的方法外，还有convert_timezone（时区转换）与is_before（时间比较）两个方法可用。

文本工具：正则灾难性回溯防护

from atr.tools.safe import TextTool text = TextTool(max_length=100_000) # Split/join result = text.split("hello world", " ") # ["hello", "world"] result = text.join(["a", "b", "c"], "-") # "a-b-c" # Replace result = text.replace("hello world", "world", "universe") # Analyze result = text.analyze("Hello world! How are you?") # {"words": 5, "sentences": 2, "characters": 25, ...} # Safe regex result = text.regex_find(r"\d+", "abc123def456") # {"matches": ["123", "456"]} # Hash result = text.hash("my text", algorithm="sha256")

text_tool.py 的_validate_regex是一大亮点：正则模式长度上限max_regex_length=200，并硬编码拦截(.+)+、(.*)*、(a+)+、(a*)*等典型灾难性回溯（ReDoS）模式，然后才尝试编译。hash方法对md5/sha1会记录弃用警告（对应 CWE-328，弱加密算法）；regex_find的匹配结果上限为max_matches=1000并返回truncated标记。完整方法集还包括regex_replace、trim、change_case、contains、truncate。

创建自定义工具

@tool装饰器由 atr/decorator 导出，它会把普通 Python 函数转换成可被发现、可被 schema 化的 ATR 工具。

基础自定义工具

from atr.decorator import tool @tool( name="my_custom_tool", description="Does something useful", tags=["custom", "safe"] ) def my_tool(input_data: str) -> dict: """ Process input data safely. Args: input_data: Data to process Returns: Processed result """ # Your logic here result = input_data.upper() return { "success": True, "result": result }

注意三个约定：函数必须带类型注解（ATR 据此生成 OpenAI Function Calling / Anthropic Tool Use 兼容的调用 schema，缺少类型注解会抛ValueError，见 test_decorator.py 的test_extract_parameters_without_type_hints_fails）；docstring 会被自动提取为参数说明；返回值建议统一使用含success字段的字典结构。

带输入校验的工具

from atr.decorator import tool from typing import Optional, List @tool( name="data_processor", description="Process data with validation", tags=["data", "safe"] ) def process_data( items: List[str], max_items: int = 100, filter_pattern: Optional[str] = None ) -> dict: """Process a list of items safely.""" # Validate inputs if len(items) > max_items: return { "success": False, "error": f"Too many items: {len(items)}. Max: {max_items}" } # Process result = [] for item in items: if filter_pattern and filter_pattern not in item: continue result.append(item.strip()) return { "success": True, "result": result, "count": len(result) }

输入校验是安全工具的第一道防线：对数量、长度、类型、取值范围的检查应当在处理逻辑之前完成，失败时返回结构化错误而非抛出异常，这样 LLM 智能体可以直接读取错误信息自我纠正。

带状态的工具类

from atr.decorator import tool from typing import Dict, Any class DatabaseTool: """Safe database query tool.""" def __init__( self, connection_string: str, allowed_tables: List[str], max_results: int = 1000 ): self.connection_string = connection_string self.allowed_tables = set(allowed_tables) self.max_results = max_results self._connection = None def _validate_table(self, table: str): """Ensure table is in allowed list.""" if table not in self.allowed_tables: raise ValueError( f"Table '{table}' not allowed. " f"Allowed: {', '.join(self.allowed_tables)}" ) @tool( name="db_select", description="Run a safe SELECT query", tags=["database", "read", "safe"] ) async def select( self, table: str, columns: List[str] = None, where: Dict[str, Any] = None, limit: int = 100 ) -> Dict[str, Any]: """ Run a SELECT query safely. Args: table: Table name (must be in allowed list) columns: Columns to select (default: all) where: WHERE conditions as dict limit: Max rows to return """ # Validate self._validate_table(table) limit = min(limit, self.max_results) # Build query safely (no SQL injection) cols = ", ".join(columns) if columns else "*" query = f"SELECT {cols} FROM {table}" if where: conditions = " AND ".join( f"{k} = ?" for k in where.keys() ) query += f" WHERE {conditions}" query += f" LIMIT {limit}" # Execute (using parameterized query) # ... actual database code ... return { "success": True, "query": query, "rows": [], # results "count": 0 }

@tool可以直接装饰类方法，构造器负责注入配置（连接串、允许的表名白名单、结果上限），业务方法在入口处执行校验——这里用allowed_tables集合做表名白名单，limit = min(limit, self.max_results)做结果上限钳制，where条件全部走参数化查询（?占位符）以杜绝 SQL 注入。注意：ATR 的注册表只保存工具规范（ToolSpec）与元数据，并不执行它们，真正执行由 Agent Runtime（Control Plane）负责，这一设计在 atr/init.py 的 docstring 中有明确说明。

异步工具

from atr.decorator import tool import asyncio @tool( name="async_fetcher", description="Fetch data asynchronously", tags=["async", "network", "safe"] ) async def fetch_multiple(urls: List[str], timeout: float = 10.0) -> dict: """Fetch multiple URLs concurrently.""" import aiohttp async def fetch_one(session, url): try: async with session.get(url, timeout=timeout) as response: return { "url": url, "status": response.status, "success": True } except Exception as e: return { "url": url, "error": str(e), "success": False } async with aiohttp.ClientSession() as session: tasks = [fetch_one(session, url) for url in urls] results = await asyncio.gather(*tasks) return { "success": True, "results": results, "total": len(results), "successful": sum(1 for r in results if r["success"]) }

异步工具天然适合 I/O 密集型场景。ATR 装饰器能自动检测函数是否为 async（async_参数在未指定时自动推断），并支持通过get_tool_handle(name).call_async(...)执行。示例中用asyncio.gather并发抓取多个 URL，并为每个请求单独捕获异常，保证单点失败不影响整体结果。

将工具注册到 ATR

注册流程

from atr import ToolRegistry from atr.tools.safe import create_safe_toolkit # Create registry registry = ToolRegistry() # Register pre-built tools toolkit = create_safe_toolkit("standard") toolkit"register_all" # Register custom tool @tool(name="my_tool", description="My custom tool") def my_tool(x: int) -> int: return x * 2 registry.register(my_tool) # List registered tools for tool in registry.list_tools(): print(f"- {tool.name}: {tool.description}")

除了显式创建ToolRegistry实例，ATR 也提供模块级的全局注册表（atr/__init__.py中的_global_registry），可直接使用atr.register(...)、atr.list_tools(...)、atr.search_tools(query)等函数（atr/init.py）。register装饰器支持的元数据远不止name/description/tags，还包括：

version：语义化版本号（默认"1.0.0"），支持>=1.0.0等版本约束查询；
author与cost：作者信息与执行成本级别（free/low/medium/high）；
side_effects：副作用声明（none/read/write/delete/network/filesystem）；
async_：是否异步（未指定时自动检测）；
permissions：允许调用该工具的 Agent 角色列表；
rate_limit：限速字符串（如"10/minute"、"100/hour"）；
retry_policy：RetryPolicy重试策略（如max_attempts=3, backoff="exponential"）；
health_check、access_policy：健康检查与细粒度访问控制策略；
deprecated/deprecated_message：标记废弃版本与迁移说明。

执行工具与沙箱

注册后可用atr.get_tool(name)获取规范（含 JSON Schema），用atr.get_tool_handle(name)获取带策略（限速、重试、指标、依赖注入）的执行句柄，或用atr.execute_tool(name, args, executor=...)直接执行。execute_tool的默认执行器是LocalExecutor（宿主直接执行，不沙箱），docstring 明确建议：处理不可信代码时应显式传入DockerExecutor获得沙箱执行能力（见 atr/init.py）。

安全最佳实践

1. 输入校验

@tool(name="safe_tool") def safe_tool(data: str, max_length: int = 1000) -> dict: # Always validate inputs if len(data) > max_length: return {"error": f"Input too long: {len(data)} > {max_length}"} # Sanitize data = data.strip() # Process return {"result": process(data)}

对长度、类型、枚举值等输入属性做前置校验，并对输入做清洗（如strip()），这是所有预置工具的通行做法。

2. 输出限制

@tool(name="list_tool") def list_tool(items: list, max_items: int = 100) -> dict: # Limit output size result = items[:max_items] truncated = len(items) > max_items return { "result": result, "truncated": truncated, "total": len(items) }

限制输出规模可防止单次工具调用返回超大结果拖垮 Agent 上下文——预置工具中的TextTool同样在结果中返回truncated标记。

3. 超时保护

import asyncio @tool(name="slow_tool") async def slow_tool(data: str, timeout: float = 30.0) -> dict: try: result = await asyncio.wait_for( slow_operation(data), timeout=timeout ) return {"success": True, "result": result} except asyncio.TimeoutError: return {"success": False, "error": "Operation timed out"}

asyncio.wait_for为可能挂起的操作设置硬超时，与HttpClientTool的timeout参数（基于aiohttp.ClientTimeout）异曲同工。

4. 资源限制

@tool(name="memory_safe_tool") def memory_safe_tool(data: list, max_memory_mb: int = 100) -> dict: import sys # Check memory usage size_bytes = sys.getsizeof(data) size_mb = size_bytes / (1024 * 1024) if size_mb > max_memory_mb: return {"error": f"Data too large: {size_mb:.1f}MB > {max_memory_mb}MB"} return {"result": process(data)}

对于大数据处理工具，用sys.getsizeof做内存占用预检，与JsonParserTool的大小/深度限制、FileReaderTool的文件大小限制共同构成资源防线。

完整源码参考与下一步

本文所述的安全工具包与注册机制均可在仓库中直接查看与验证：

安全工具集包入口：atr/tools/safe/init.py
工具包工厂（五预设 + register_all）：atr/tools/safe/toolkit.py
六个安全工具实现：http_client.py、file_reader.py、calculator.py、json_parser.py、datetime_tool.py、text_tool.py
ATR 全局注册表与register装饰器全参数：atr/init.py
ATR 完整文档：modules/atr/README.md
装饰器测试用例（类型注解强制、元数据提取、不执行函数等）：modules/atr/tests/test_decorator.py
速查表：docs/cheatsheet.md

掌握本文内容后，你可以：用五种预设快速装配安全工具集、按统一模式编写带校验/带状态/异步的自定义工具、通过register的版本、限速、权限、重试等元数据精确控制工具暴露面，并为自己的工具补齐输入校验、输出限制、超时与资源四道安全防线，从而在 Agent OS 中构建既强大又可控的工具生态。

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考

Agent OS 安全工具开发实战：基于 ATR 构建可复用的 Custom Tools