89 lines
2.9 KiB
Python
89 lines
2.9 KiB
Python
import subprocess
|
|
import time
|
|
import os
|
|
from pathlib import Path
|
|
from typing import Dict, Any
|
|
|
|
import yaml
|
|
|
|
from .models import RunRequest, RunResult
|
|
|
|
|
|
def load_config(config_path: str | Path = None) -> Dict[str, Any]:
|
|
"""加载配置，默认使用 AGENT_CONFIG 或 ./config.yaml。"""
|
|
if config_path is None:
|
|
env_path = os.environ.get("AGENT_CONFIG")
|
|
config_path = Path(env_path) if env_path else Path(__file__).resolve().parent.parent / "config.yaml"
|
|
config_path = Path(config_path)
|
|
with open(config_path, "r", encoding="utf-8") as f:
|
|
return yaml.safe_load(f) or {}
|
|
|
|
|
|
def resolve_tool(config: Dict[str, Any], tool_name: str) -> Path:
|
|
tools = config.get("tools", {})
|
|
if tool_name not in tools:
|
|
raise ValueError(f"未知工具: {tool_name}")
|
|
tool_path = Path(tools[tool_name]["path"])
|
|
if not tool_path.exists():
|
|
raise FileNotFoundError(f"工具不存在: {tool_path}")
|
|
return tool_path
|
|
|
|
|
|
def run_tool(config: Dict[str, Any], req: RunRequest) -> RunResult:
|
|
start = time.time()
|
|
defaults = config.get("run_defaults", {})
|
|
|
|
tool_path = resolve_tool(config, req.tool)
|
|
workdir = Path(req.workdir or config.get("base_workdir", ".")).resolve()
|
|
workdir.mkdir(parents=True, exist_ok=True)
|
|
|
|
# 处理参数与环境
|
|
args = [str(tool_path)] + req.args
|
|
env = os.environ.copy()
|
|
if config.get("env"):
|
|
env.update({k: str(v) for k, v in config["env"].items()})
|
|
if req.env:
|
|
env.update({k: str(v) for k, v in req.env.items()})
|
|
|
|
timeout = req.timeout or defaults.get("timeout")
|
|
capture_output = req.capture_output if req.capture_output is not None else defaults.get("capture_output", True)
|
|
strip_output = req.strip_output if req.strip_output is not None else defaults.get("strip_output", True)
|
|
|
|
try:
|
|
completed = subprocess.run(
|
|
args,
|
|
cwd=str(workdir),
|
|
env=env,
|
|
timeout=timeout,
|
|
capture_output=capture_output,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
stdout = completed.stdout if capture_output else None
|
|
stderr = completed.stderr if capture_output else None
|
|
if strip_output:
|
|
stdout = stdout.strip() if stdout is not None else None
|
|
stderr = stderr.strip() if stderr is not None else None
|
|
|
|
elapsed = time.time() - start
|
|
return RunResult(
|
|
success=completed.returncode == 0,
|
|
return_code=completed.returncode,
|
|
stdout=stdout,
|
|
stderr=stderr,
|
|
elapsed=elapsed,
|
|
tool_path=str(tool_path),
|
|
workdir=str(workdir),
|
|
)
|
|
except Exception as exc:
|
|
elapsed = time.time() - start
|
|
return RunResult(
|
|
success=False,
|
|
return_code=None,
|
|
stdout=None,
|
|
stderr=str(exc),
|
|
elapsed=elapsed,
|
|
tool_path=str(tool_path),
|
|
workdir=str(workdir),
|
|
)
|