Custom Code Tool - 개발자 툴킷
이 샘플 코드는 Aidputils를 사용하여 사용자 정의 코드 도구를 테스트하는 방법을 보여줍니다.
Developer Toolkit 예제는 utils/ 디렉토리에서 다중 도구 패키지 및 helper 모듈 사용을 보여줍니다. 이 패키지는 bash 명령 실행기, 파일 작업 도구 및 Python 코드 실행기라는 세 가지 도구를 등록하고 공유 도우미 함수를 사용하여 출력 자르기 및 경로 제재를 수행합니다.
주:
Developer Toolkit은 설명 예제입니다. Bash 명령 실행 및 Python 코드 실행은 상당한 보안 영향을 미칩니다. 프로덕션 환경에서 AI 컴퓨팅을 제한하고, 작업을 샌드박스화하고, 도구가 실행할 명령 및 코드 패턴에 대해 엄격한 허용 목록을 적용합니다.패키지 레이아웃
advanced_tool.zip
├── tool_implementation.py
├── tool_config.json
├── requirements.txt # stdlib only
└── utils/
├── __init__.py
└── text_utils.py # truncate_output, sanitize_path
도구_구현.py
import subprocess
import os
from aidputils.agents.tools.custom_tools.base import CustomToolBase
from .utils.text_utils import truncate_output, sanitize_path
def _get_cfg(conf, key, default):
"""Read a config value from either the outer dict or the
nested user conf. Coerces numeric settings to int to avoid
type mismatches when values are rendered as strings by the
template substitution layer."""
inner = conf.get("conf") if isinstance(conf, dict) else None
if isinstance(inner, dict) and key in inner:
value = inner[key]
elif isinstance(conf, dict) and key in conf:
value = conf[key]
else:
value = default
if isinstance(default, int) and not isinstance(value, bool):
try:
return int(value)
except (TypeError, ValueError):
return default
return value
@BaseTool.register
class BashTool(CustomToolBase):
"""Execute bash commands and return output."""
@classmethod
def _execute_tool(cls, conf, runtime_params, **context_vars):
command = runtime_params.get("command", "")
timeout = _get_cfg(conf, "timeout", 30)
max_lines = _get_cfg(conf, "max_output_lines", 200)
try:
result = subprocess.run(
["bash", "-c", command],
capture_output=True, text=True, timeout=timeout
)
except subprocess.TimeoutExpired:
# Surface the timeout as a tool failure rather than
# returning {"error": ...}, which would be treated as
# a successful response.
raise RuntimeError(f"Command timed out after {timeout}s")
output = result.stdout or ""
if result.stderr:
output += "\n[stderr]\n" + result.stderr
return {"output": truncate_output(output, max_lines)}
@BaseTool.register
class FileTool(CustomToolBase):
"""Read, write, or list files in the workspace."""
@classmethod
def _execute_tool(cls, conf, runtime_params, **context_vars):
operation = runtime_params.get("operation", "")
path = runtime_params.get("path", "")
content = runtime_params.get("content", "")
base_dir = _get_cfg(conf, "base_dir", "/workspace")
max_size = _get_cfg(conf, "max_file_size_kb", 1024) * 1024
safe_path = sanitize_path(base_dir, path)
if safe_path is None:
raise ValueError("Invalid path: path traversal detected")
if operation == "read":
with open(safe_path, "r") as f:
return {"output": f.read()}
if operation == "write":
parent = os.path.dirname(safe_path)
if parent:
os.makedirs(parent, exist_ok=True)
with open(safe_path, "w") as f:
f.write(content)
return {"output": f"Written {len(content)} chars to {path}"}
if operation == "list":
target = safe_path if os.path.isdir(safe_path) else os.path.dirname(safe_path)
return {"output": "\n".join(sorted(os.listdir(target)))}
raise ValueError(f"Unknown operation: {operation}. Use read/write/list")
@BaseTool.register
class PythonTool(CustomToolBase):
"""Execute Python code in an isolated subprocess."""
@classmethod
def _execute_tool(cls, conf, runtime_params, **context_vars):
code = runtime_params.get("code", "")
timeout = _get_cfg(conf, "timeout", 60)
max_lines = _get_cfg(conf, "max_output_lines", 500)
try:
result = subprocess.run(
["python3", "-c", code],
capture_output=True, text=True, timeout=timeout
)
except subprocess.TimeoutExpired:
raise RuntimeError(f"Execution timed out after {timeout}s")
output = result.stdout or ""
if result.stderr:
output += "\n[stderr]\n" + result.stderr
return {"output": truncate_output(output, max_lines)}
도구_config.json
{
"displayName": "Developer Toolkit",
"description": "A collection of tools for bash commands, file operations, and Python execution",
"tools": [
{
"toolClassName": "BashTool",
"displayName": "Bash Tool",
"description": "Executes a bash command and returns stdout/stderr output",
"version": "1.0.0",
"schema": [
{
"name": "command",
"type": "string",
"description": "The bash command to execute"
}
],
"conf": {
"timeout": 30,
"max_output_lines": 200
}
},
{
"toolClassName": "FileTool",
"displayName": "File Tool",
"description": "Read, write, or list files in the workspace",
"version": "1.0.0",
"schema": [
{"name": "operation", "type": "string",
"description": "Operation to perform: read, write, or list"},
{"name": "path", "type": "string",
"description": "File or directory path"},
{"name": "content", "type": "string",
"description": "Content to write (for write operation)"}
],
"conf": {
"base_dir": "/workspace",
"max_file_size_kb": 1024
}
},
{
"toolClassName": "PythonTool",
"displayName": "Python Tool",
"description": "Executes Python code in an isolated subprocess and returns the output",
"version": "1.0.0",
"schema": [
{"name": "code", "type": "string",
"description": "The Python code to execute"}
],
"conf": {
"timeout": 60,
"max_output_lines": 500
}
}
]
}유틸리티/텍스트_utils.py
def truncate_output(text, max_lines=200):
if not text:
return ""
try:
max_lines = int(max_lines)
except (TypeError, ValueError):
max_lines = 200
lines = text.strip().split("\n")
if len(lines) > max_lines:
lines = lines[:max_lines] + [f"... ({len(lines) - max_lines} lines truncated)"]
return "\n".join(lines)
def sanitize_path(base_dir, relative_path):
import os
if not relative_path:
return base_dir
full = os.path.normpath(os.path.join(base_dir, relative_path))
if not full.startswith(os.path.normpath(base_dir)):
return None
return full
uts/__init__.py
# Empty file. Required for Python to treat utils/ as a package.요구 사항.txt
# stdlib onlyZIP을 업로드한 후 Package(패키지) 탭에 검색된 세 가지 도구가 표시되고 각 도구를 사용 또는 사용 안함으로 설정할 수 있습니다. 매개변수 탭에는 BashTool, FileTool 및 PythonTool 간을 전환하고 오른쪽에 도구별 구성(timeout, max_output_lines, base_dir, max_file_size_kb)을 노출하는 Tool Class 드롭다운이 표시됩니다.