李祯煜1. 从零构建Agent技能工具：完整开发与注册指南
在构建智能代理(Agent)系统时，技能工具(Skill Tools)的创建与注册是核心环节。本文将基于实际工程经验，详细解析如何从零开始设计、实现并注册一个完整的工具系统。不同于简单的API调用，我们将深入探讨工具架构设计、参数校验、错误处理等关键细节，这些都是实际项目中容易踩坑的地方。
1.1 工具系统核心概念解析
在开始编码前，我们需要明确几个核心概念及其相互关系：
-
Tool（工具）：可执行的最小功能单元，包含名称、描述、参数定义和执行逻辑。例如计算器工具、天气查询工具等。每个工具应当保持单一职责原则。
-
Skill（技能）：相关工具的集合，通常对应一个业务领域。例如"文件处理技能"可能包含文件读取、写入、压缩等多个工具。
-
Registry（注册表）：系统的工具管理中心，负责工具的注册、查询和管理。良好的注册表设计能显著降低系统耦合度。
-
Executor（执行器）：负责工具的实际调用，处理参数校验、异常捕获、结果包装等横切关注点。
提示：在实际项目中，建议将工具系统设计为独立模块，与具体的Agent实现解耦。这样同一套工具可以服务于不同类型的Agent。
1.2 工具接口的最小化设计
一个健壮的工具接口应包含以下基本要素：
python复制from abc import ABC, abstractmethod
from typing import Dict, Any, List
class BaseTool(ABC):
@property
@abstractmethod
def name(self) -> str:
"""工具的唯一标识符，建议使用小写加下划线命名"""
pass
@property
@abstractmethod
def description(self) -> str:
"""工具的详细描述，LLM将根据此描述决定是否调用该工具"""
pass
@abstractmethod
def get_parameters(self) -> List[Dict]:
"""返回工具的参数定义，遵循JSON Schema规范"""
pass
@abstractmethod
def run(self, parameters: Dict[str, Any]) -> Dict:
"""
工具的执行入口
返回格式应统一为：{
'success': bool,
'data': Any,
'error': str
}
"""
pass
关键设计考虑：
- 名称规范：工具名称应简洁明确，避免特殊字符，推荐使用
snake_case风格 - 描述质量：描述应当清晰说明工具功能、适用场景和限制条件
- 参数定义：使用标准化的JSON Schema，便于不同系统间的互操作
- 返回结构：统一的结构便于上层处理成功/失败场景
需要模型API调用？ 免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。
2. 计算器工具的实现与注册
2.1 具体工具实现示例
以下是一个完整计算器工具的实现，包含参数校验和错误处理：
python复制import math
from typing import Dict, Any
class CalculatorTool(BaseTool):
@property
def name(self) -> str:
return "advanced_calculator"
@property
def description(self) -> str:
return (
"高级数学计算工具，支持加减乘除、幂运算和开方。"
"表达式格式示例：'(3 + 5) * 2' 或 'sqrt(16)'"
)
def get_parameters(self) -> List[Dict]:
return [
{
"name": "expression",
"type": "string",
"required": True,
"description": "数学表达式，支持+,-,*,/,^,sqrt()",
"examples": ["(3+5)*2", "sqrt(16)"]
}
]
def run(self, parameters: Dict[str, Any]) -> Dict:
try:
expr = parameters.get("expression", "")
if not expr:
return {"success": False, "error": "表达式不能为空"}
# 安全评估表达式
result = self._safe_eval(expr)
return {"success": True, "data": result}
except Exception as e:
return {"success": False, "error": f"计算失败: {str(e)}"}
def _safe_eval(self, expr: str) -> float:
"""安全评估数学表达式"""
allowed_chars = set("0123456789+-*/.()^ ")
if not all(c in allowed_chars for c in expr):
raise ValueError("表达式包含非法字符")
# 替换sqrt为math.sqrt
expr = expr.replace("sqrt", "math.sqrt")
# 替换^为**（幂运算）
expr = expr.replace("^", "**")
# 限制可用的全局变量和函数
safe_globals = {"math": math}
return eval(expr, {"__builtins__": None}, safe_globals)
实现要点：
- 输入验证：检查表达式非空且只包含允许的字符
- 安全评估：限制eval的执行环境，防止代码注入
- 错误处理：捕获所有异常并返回结构化错误信息
- 符号转换：将用户友好的sqrt和^转换为Python语法
2.2 工具注册表的设计与实现
注册表需要管理工具的生命周期并提供查询接口：
python复制from typing import Dict, List
class ToolRegistry:
def __init__(self):
self._tools: Dict[str, BaseTool] = {}
def register(self, tool: BaseTool) -> None:
"""注册工具并进行基本校验"""
if not isinstance(tool, BaseTool):
raise ValueError("必须继承自BaseTool")
if not tool.name or not isinstance(tool.name, str):
raise ValueError("工具名称必须为非空字符串")
if tool.name in self._tools:
raise ValueError(f"工具名称'{tool.name}'已存在")
self._tools[tool.name] = tool
def get(self, name: str) -> BaseTool:
"""根据名称获取工具实例"""
tool = self._tools.get(name)
if not tool:
raise KeyError(f"未找到工具: {name}")
return tool
def list(self) -> List[Dict]:
"""返回所有工具的元信息列表"""
return [
{
"name": tool.name,
"description": tool.description,
"parameters": tool.get_parameters()
}
for tool in self._tools.values()
]
def to_openai_schema(self) -> List[Dict]:
"""转换为OpenAI函数调用兼容的格式"""
return [
{
"name": tool.name,
"description": tool.description,
"parameters": {
"type": "object",
"properties": {
param["name"]: {
k: v for k, v in param.items()
if k != "name"
}
for param in tool.get_parameters()
},
"required": [
param["name"]
for param in tool.get_parameters()
if param.get("required", True)
]
}
}
for tool in self._tools.values()
]
注册表关键功能：
- 名称冲突检测：防止重复注册导致工具覆盖
- 元信息管理：提供工具列表查询功能
- 格式转换：支持转换为不同框架需要的格式
- 类型检查：确保注册的对象符合工具接口规范
3. 工具执行与Agent集成
3.1 工具执行器的实现
执行器负责处理工具调用的完整生命周期：
python复制import logging
from typing import Any, Dict
class ToolExecutor:
def __init__(self, registry: ToolRegistry):
self.registry = registry
self.logger = logging.getLogger(__name__)
def execute(self, tool_name: str, parameters: Dict[str, Any]) -> Dict:
"""
执行工具并返回结构化结果
返回格式：{
'tool': str,
'success': bool,
'data': Any,
'error': str,
'execution_time': float
}
"""
import time
start_time = time.time()
try:
# 获取工具实例
tool = self.registry.get(tool_name)
# 参数预处理
processed_params = self._preprocess_parameters(
tool.get_parameters(),
parameters
)
# 执行工具
result = tool.run(processed_params)
# 记录执行日志
self.logger.info(
f"Tool executed - {tool_name}: "
f"params={parameters}, result={result}"
)
return {
"tool": tool_name,
"success": result["success"],
"data": result.get("data"),
"error": result.get("error", ""),
"execution_time": time.time() - start_time
}
except Exception as e:
self.logger.error(f"Tool execution failed: {str(e)}")
return {
"tool": tool_name,
"success": False,
"error": str(e),
"execution_time": time.time() - start_time
}
def _preprocess_parameters(
self,
schema: List[Dict],
params: Dict[str, Any]
) -> Dict[str, Any]:
"""根据schema预处理参数"""
processed = {}
for param_def in schema:
name = param_def["name"]
required = param_def.get("required", True)
if name not in params:
if required:
raise ValueError(f"缺少必要参数: {name}")
continue
value = params[name]
param_type = param_def.get("type", "string")
# 简单类型转换
try:
if param_type == "number" and isinstance(value, str):
value = float(value)
elif param_type == "integer" and isinstance(value, str):
value = int(value)
elif param_type == "boolean" and isinstance(value, str):
value = value.lower() in ("true", "1", "yes")
except (ValueError, TypeError) as e:
raise ValueError(
f"参数'{name}'类型转换失败: {str(e)}"
) from e
processed[name] = value
return processed
执行器核心功能：
- 参数预处理：根据schema进行类型转换和校验
- 错误隔离：捕获所有异常避免影响Agent主流程
- 性能监控：记录执行时间用于性能分析
- 日志记录：详细记录执行过程便于问题排查
3.2 与Agent系统的集成
将工具系统集成到Agent中的典型流程：
python复制class MyAgent:
def __init__(self, registry: ToolRegistry):
self.registry = registry
self.executor = ToolExecutor(registry)
def generate_response(self, user_input: str) -> str:
# 获取工具schema用于LLM提示
tools_schema = self.registry.to_openai_schema()
# 构造包含工具信息的提示
prompt = self._build_prompt(user_input, tools_schema)
# 调用LLM获取初始响应
llm_response = self._call_llm(prompt)
# 处理可能的工具调用
if self._is_tool_call(llm_response):
tool_name = llm_response["tool_name"]
tool_params = llm_response["parameters"]
# 执行工具
tool_result = self.executor.execute(tool_name, tool_params)
# 将结果反馈给LLM
followup_prompt = self._build_followup_prompt(
user_input,
llm_response,
tool_result
)
# 获取最终响应
final_response = self._call_llm(followup_prompt)
return final_response
return llm_response["text"]
def _build_prompt(self, user_input: str, tools_schema: List) -> str:
"""构造包含工具信息的提示"""
# 实际实现中这里会包含更复杂的提示工程
return f"""
用户问题：{user_input}
可用工具：{tools_schema}
请根据问题决定是否需要调用工具。
"""
def _is_tool_call(self, response: Dict) -> bool:
"""判断LLM响应是否为工具调用"""
return "tool_name" in response and "parameters" in response
集成关键点：
- 提示工程：在提示中清晰描述可用工具及其用法
- 响应解析：正确识别LLM的工具调用意图
- 结果反馈：将工具执行结果有效整合到后续对话中
- 错误恢复：当工具调用失败时提供备选方案
4. 高级主题与最佳实践
4.1 技能包的组织与管理
随着工具数量增长，建议按功能领域组织为技能包：
code复制skills/
├── math/
│ ├── __init__.py # 暴露get_tools()
│ ├── calculator.py
│ └── statistics.py
├── web/
│ ├── __init__.py
│ ├── search.py
│ └── scraper.py
└── file/
├── __init__.py
├── reader.py
└── writer.py
每个技能包的__init__.py提供统一入口：
python复制from .calculator import CalculatorTool
from .statistics import StatsTool
def get_tools() -> List[BaseTool]:
"""返回该技能包提供的所有工具"""
return [
CalculatorTool(),
StatsTool()
]
注册时按需加载技能包：
python复制registry = ToolRegistry()
# 加载数学相关技能
from skills.math import get_tools as get_math_tools
for tool in get_math_tools():
registry.register(tool)
# 加载文件处理技能
from skills.file import get_tools as get_file_tools
for tool in get_file_tools():
registry.register(tool)
4.2 工具测试策略
完善的测试是保证工具可靠性的关键：
python复制import pytest
from skills.math.calculator import CalculatorTool
class TestCalculatorTool:
@pytest.fixture
def tool(self):
return CalculatorTool()
def test_basic_operations(self, tool):
# 测试基本运算
assert tool.run({"expression": "2+3"})["data"] == 5
assert tool.run({"expression": "5-2"})["data"] == 3
assert tool.run({"expression": "2*3"})["data"] == 6
assert tool.run({"expression": "6/2"})["data"] == 3
def test_advanced_operations(self, tool):
# 测试高级函数
assert tool.run({"expression": "sqrt(9)"})["data"] == 3
assert tool.run({"expression": "2^3"})["data"] == 8
def test_error_handling(self, tool):
# 测试错误处理
result = tool.run({"expression": "1/0"})
assert not result["success"]
assert "division" in result["error"].lower()
result = tool.run({"expression": "abc"})
assert not result["success"]
assert "非法字符" in result["error"]
def test_security(self, tool):
# 测试安全限制
result = tool.run({"expression": "__import__('os').system('ls')"})
assert not result["success"]
assert "非法字符" in result["error"]
测试覆盖要点：
- 正常场景：验证核心功能正确性
- 边界条件：测试极端输入和边界值
- 错误处理：验证错误信息的准确性和有用性
- 安全防护：确保不会执行危险操作
4.3 性能优化技巧
在大规模工具系统中，以下优化策略值得考虑：
-
懒加载：延迟加载工具实现，减少启动时间
python复制class LazyTool(BaseTool): def __init__(self, tool_class): self._tool_class = tool_class self._instance = None @property def _tool(self): if self._instance is None: self._instance = self._tool_class() return self._instance def run(self, parameters): return self._tool.run(parameters) -
缓存机制：对计算密集型工具添加结果缓存
python复制class CachedTool(BaseTool): def __init__(self, wrapped_tool, ttl=60): self._wrapped = wrapped_tool self._cache = {} self._ttl = ttl def run(self, parameters): cache_key = self._make_cache_key(parameters) if cache_key in self._cache: cached = self._cache[cache_key] if time.time() - cached["time"] < self._ttl: return cached["result"] result = self._wrapped.run(parameters) self._cache[cache_key] = { "result": result, "time": time.time() } return result -
批量执行：支持并行执行多个工具调用
python复制from concurrent.futures import ThreadPoolExecutor class BatchExecutor: def __init__(self, registry, max_workers=4): self.executor = ToolExecutor(registry) self.pool = ThreadPoolExecutor(max_workers) def execute_batch(self, calls: List[Dict]) -> List[Dict]: futures = [ self.pool.submit( self.executor.execute, call["tool"], call["parameters"] ) for call in calls ] return [f.result() for f in futures] -
流量控制：限制高频工具的调用速率
python复制import time from collections import deque class RateLimitedTool(BaseTool): def __init__(self, wrapped_tool, calls_per_minute=60): self._wrapped = wrapped_tool self._limit = calls_per_minute self._timestamps = deque(maxlen=calls_per_minute) def run(self, parameters): now = time.time() if len(self._timestamps) >= self._limit: oldest = self._timestamps[0] if now - oldest < 60: time.sleep(60 - (now - oldest)) self._timestamps.append(time.time()) return self._wrapped.run(parameters)
4.4 可观测性与监控
生产环境中，完善的监控体系必不可少：
-
日志记录：详细记录工具调用信息
python复制class LoggingTool(BaseTool): def __init__(self, wrapped_tool): self._wrapped = wrapped_tool self.logger = logging.getLogger("tooling") def run(self, parameters): start = time.time() try: result = self._wrapped.run(parameters) self.logger.info( f"Tool {self._wrapped.name} executed - " f"params: {parameters}, " f"result: {result}, " f"duration: {time.time()-start:.3f}s" ) return result except Exception as e: self.logger.error( f"Tool {self._wrapped.name} failed - " f"params: {parameters}, " f"error: {str(e)}" ) raise -
指标收集：跟踪关键性能指标
python复制from prometheus_client import Summary, Counter TOOL_TIME = Summary( 'tool_execution_seconds', 'Time spent processing tool calls', ['tool'] ) TOOL_CALLS = Counter( 'tool_calls_total', 'Total number of tool calls', ['tool', 'status'] ) class InstrumentedTool(BaseTool): def __init__(self, wrapped_tool): self._wrapped = wrapped_tool def run(self, parameters): start = time.time() try: result = self._wrapped.run(parameters) TOOL_TIME.labels( tool=self._wrapped.name ).observe(time.time() - start) TOOL_CALLS.labels( tool=self._wrapped.name, status="success" ).inc() return result except Exception: TOOL_CALLS.labels( tool=self._wrapped.name, status="failure" ).inc() raise -
分布式追踪：集成OpenTelemetry等追踪系统
python复制from opentelemetry import trace tracer = trace.get_tracer("tooling") class TracedTool(BaseTool): def __init__(self, wrapped_tool): self._wrapped = wrapped_tool def run(self, parameters): with tracer.start_as_current_span( f"tool.{self._wrapped.name}" ) as span: span.set_attributes({ "tool.name": self._wrapped.name, "tool.params": str(parameters) }) try: result = self._wrapped.run(parameters) span.set_status( trace.Status(trace.StatusCode.OK) ) return result except Exception as e: span.record_exception(e) span.set_status( trace.Status( trace.StatusCode.ERROR, str(e) ) ) raise
5. 常见问题与解决方案
5.1 工具调用失败处理策略
当工具调用失败时，Agent系统应当具备以下恢复能力：
-
参数修正：分析错误信息并尝试自动修正参数
python复制def try_fix_parameters(tool, error, original_params): error_msg = str(error).lower() if "missing" in error_msg: # 尝试补充缺失参数 for param in tool.get_parameters(): if param["name"] not in original_params: if "default" in param: original_params[param["name"]] = param["default"] if "type" in error_msg: # 尝试类型转换 for param in tool.get_parameters(): if param["name"] in original_params: try: if param["type"] == "number": original_params[param["name"]] = float( original_params[param["name"]] ) elif param["type"] == "integer": original_params[param["name"]] = int( original_params[param["name"]] ) except (ValueError, TypeError): continue return original_params -
备选工具：尝试使用功能相似的其他工具
python复制def find_alternative_tools(registry, failed_tool_name): failed_tool = registry.get(failed_tool_name) alternatives = [] for tool in registry.list(): if tool["name"] != failed_tool_name and ( tool["description"].split()[0] == failed_tool.description.split()[0] ): alternatives.append(tool) return alternatives -
降级处理：返回部分结果或提供手动解决方案
python复制def graceful_degradation(error, context): error_msg = str(error).lower() if "connection" in error_msg: return "系统暂时无法访问所需服务，请稍后再试" elif "invalid" in error_msg: return "输入参数不符合要求，请检查后重试" else: return f"处理请求时遇到问题：{str(error)}"
5.2 工具版本管理策略
随着系统演进，工具接口可能需要变更，良好的版本管理策略至关重要：
-
语义化版本：遵循主版本.次版本.修订号规则
python复制class VersionedTool(BaseTool): def __init__(self): self._version = "1.0.0" @property def version(self) -> str: return self._version -
多版本共存：通过名称区分不同版本
python复制registry.register(CalculatorV1()) # 注册为calculator_v1 registry.register(CalculatorV2()) # 注册为calculator_v2 -
兼容性检查：确保Agent与工具版本兼容
python复制def check_compatibility(agent_version, tool): required = tool.required_agent_version return semver.match(agent_version, f">={required}")
5.3 权限控制实现方案
根据业务需求，可能需要限制某些工具的使用：
-
基于角色的访问控制
python复制class RBACTool(BaseTool): def __init__(self, wrapped_tool, required_roles): self._wrapped = wrapped_tool self._required_roles = required_roles def run(self, parameters, user_roles=None): if user_roles is None: user_roles = [] if not set(self._required_roles).intersection(user_roles): return { "success": False, "error": "权限不足" } return self._wrapped.run(parameters) -
属性基访问控制
python复制class ABACTool(BaseTool): def __init__(self, wrapped_tool, policy): self._wrapped = wrapped_tool self._policy = policy def run(self, parameters, user_attrs=None): if user_attrs is None: user_attrs = {} if not self._policy.evaluate(user_attrs): return { "success": False, "error": "访问被拒绝" } return self._wrapped.run(parameters) -
使用量配额控制
python复制class QuotaTool(BaseTool): def __init__(self, wrapped_tool, quota_manager): self._wrapped = wrapped_tool self._quota = quota_manager def run(self, parameters, user_id=None): if user_id and not self._quota.check(user_id): return { "success": False, "error": "配额已用完" } result = self._wrapped.run(parameters) if user_id and result["success"]: self._quota.consume(user_id) return result
5.4 工具发现与自描述机制
良好的自描述能力可以简化系统集成：
-
OpenAPI兼容描述
python复制def generate_openapi_spec(tool): params = { param["name"]: { "type": param["type"], "description": param.get("description", ""), "required": param.get("required", False) } for param in tool.get_parameters() } return { "operationId": tool.name, "summary": tool.description, "parameters": [ { "name": name, "in": "query", **spec } for name, spec in params.items() ], "responses": { "200": { "description": "成功响应", "content": { "application/json": { "schema": { "type": "object", "properties": { "success": {"type": "boolean"}, "data": {"type": "object"}, "error": {"type": "string"} } } } } } } } -
嵌入式文档支持
python复制class DocumentedTool(BaseTool): def __init__(self, wrapped_tool, examples=None): self._wrapped = wrapped_tool self._examples = examples or [] @property def documentation(self) -> Dict: return { "name": self._wrapped.name, "description": self._wrapped.description, "parameters": self._wrapped.get_parameters(), "examples": self._examples } -
交互式探索接口
python复制class ToolExplorer: def __init__(self, registry): self._registry = registry def list_tools(self) -> List[Dict]: return [ { "name": tool.name, "description": tool.description, "parameters": tool.get_parameters() } for tool in self._registry.list() ] def describe_tool(self, name: str) -> Dict: tool = self._registry.get(name) return { "name": tool.name, "description": tool.description, "parameters": tool.get_parameters(), "examples": getattr(tool, "examples", []) } def try_tool(self, name: str, params: Dict) -> Dict: tool = self._registry.get(name) return tool.run(params)
6. 演进路线与扩展方向
6.1 工具编排与工作流
将多个工具组合成更复杂的工作流：
-
顺序执行模式
python复制def execute_sequence(tools, initial_input): context = initial_input for tool_spec in tools: tool = registry.get(tool_spec["name"]) params = self._resolve_parameters( tool_spec.get("parameters", {}), context ) result = tool.run(params) if not result["success"]: return result context.update(result["data"]) return {"success": True, "data": context} -
条件分支模式
python复制def execute_conditional(tools, context): for condition, tool_spec in tools: if evaluate_condition(condition, context): tool = registry.get(tool_spec["name"]) params = self._resolve_parameters( tool_spec.get("parameters", {}), context ) result = tool.run(params) if not result["success"]: return result context.update(result["data"]) break return {"success": True, "data": context} -
并行执行模式
python复制def execute_parallel(tools, context): with ThreadPoolExecutor() as executor: futures = { tool_spec["name"]: executor.submit( lambda ts: registry.get(ts["name"]).run( self._resolve_parameters( ts.get("parameters", {}), context ) ), tool_spec ) for tool_spec in tools } results = { name: future.result() for name, future in futures.items() } if all(r["success"] for r in results.values()): return { "success": True, "data": { name: r["data"] for name, r in results.items() } } else: return { "success": False, "errors": { name: r["error"] for name, r in results.items() if not r["success"] } }
6.2 工具学习与自适应
使系统能够自动优化工具使用：
-
使用频率统计
python复制class UsageTracker: def __init__(self): self._counts = defaultdict(int) def record_usage(self, tool_name): self._counts[tool_name] += 1 def get_most_used(self, n=5): return sorted( self._counts.items(), key=lambda x: x[1], reverse=True )[:n] -
成功率监控
python复制class SuccessRateMonitor: def __init__(self): self._stats = defaultdict(lambda: {"total": 0, "success": 0}) def record_result(self, tool_name, success): self._stats[tool_name]["total"] += 1 if success: self._stats[tool_name]["success"] += 1 def get_success_rate(self, tool_name): stat = self._stats[tool_name] if stat["total"] == 0: return 0.0 return stat["success"] / stat["total"] -
自动工具推荐
python复制class ToolRecommender: def __init__(self, registry, tracker, monitor): self._registry = registry self._tracker = tracker self._monitor = monitor def recommend(self, query, n=3): # 基于使用频率、成功率和查询相似度综合评分 tools = [ { "tool": tool, "score": self._calculate_score(tool, query) } for tool in self._registry.list() ] return sorted( tools, key=lambda x: x["score"], reverse=True )[:n] def _calculate_score(self, tool, query): usage = self._tracker._counts.get(tool.name, 0) success_rate = self._monitor.get_success_rate(tool.name) similarity = self._calculate_similarity(tool.description, query) return ( 0.3 * math.log(1 + usage) + 0.4 * success_rate + 0.3 * similarity )
6.3 工具市场与共享生态
构建工具共享平台的关键组件：
-
工具打包规范
python复制def create_tool_package(tool, dependencies=None): return { "metadata": { "name": tool.name, "version": getattr(tool, "version", "1.0.0"), "description": tool.description, "author": getattr(tool, "author", "unknown"), "license": getattr(tool, "license", "MIT") }, "parameters": tool.get_parameters(), "code": inspect.getsource(tool.__class__), "dependencies": dependencies or [] } -
依赖管理
python复制class DependencyManager: def __init__(self): self._graph = defaultdict(set) def add_tool(self, tool_name, deps): self._graph[tool_name] = set(deps) def resolve_dependencies(self, tool_name): resolved = set() self._resolve(tool_name, resolved) return resolved def _resolve(self, tool_name, resolved): for dep in self._graph[tool_name]: if dep not in resolved: self._resolve(dep, resolved) resolved.add(tool_name) -
安全扫描
python复制class SecurityScanner: def scan(self, tool_code): issues = [] # 检查危险函数调用 for node in ast.walk(ast.parse(tool_code)): if isinstance(node, ast.Call): if isinstance(node.func, ast.Name): if node.func.id in ("eval", "exec", "open"): issues.append(f"危险函数调用: {node.func.id}") # 检查网络访问 if "import requests" in tool_code: issues.append("工具包含网络访问能力") # 检查文件操作 if "open(" in tool_code: issues.append("工具包含文件操作能力") return issues
6.4 工具可视化开发环境
降低工具开发门槛的可视化方案：
-
参数表单生成
python复制def generate_parameter_form(schema): form = [] for param in schema: field = { "name": param["name"], "label": param.get("label", param["name"]), "type": param["type"], "required": param.get("required", False) } if "description" in param: field["help"] = param["description"] if "default" in param: field["default"] = param["default"] if "enum" in param: field["options"] = param["enum"] form.append(field) return form -
**可视化逻辑编排