搭建一个轻量 Agent Harness——让 AI Agent 安全地执行命令、读写文件

Agent 想做什么 → Harness 检查和审批 → 执行 → 返回结果

核心能力：
├─ 🛡️ 沙箱执行：限制 Agent 只能操作指定目录
├─ 📋 命令白名单：只允许安全的命令
└─ ✅ 人工审批：危险操作需要确认

# harness.py
import subprocess, os, shlex
from pathlib import Path

class AgentHarness:
    """Agent 安全执行环境。"""

    ALLOWED_COMMANDS = {"ls", "cat", "head", "tail", "grep", "wc",
                        "find", "echo", "date", "python", "pip", "git"}

    def __init__(self, workspace="./agent_workspace", require_confirm=True):
        self.workspace = Path(workspace).resolve()
        self.workspace.mkdir(parents=True, exist_ok=True)
        self.require_confirm = require_confirm
        self.history = []

    def execute(self, command: str) -> dict:
        """在沙箱中执行命令。"""
        cmd_parts = shlex.split(command)
        if not cmd_parts:
            return {"error": "Empty command"}

        # 安全检查
        base_cmd = cmd_parts[0]
        if base_cmd not in self.ALLOWED_COMMANDS:
            return {"error": f"命令 {base_cmd} 不在白名单中"}

        # 危险操作需要确认
        dangerous = any(k in command for k in ["rm ", "delete", "DROP", ">", ">>"])
        if dangerous and self.require_confirm:
            print(f"\n⚠️ 危险操作：{command}")
            ok = input("确认执行？[y/N] ").strip().lower()
            if ok != "y":
                return {"error": "用户取消"}

        try:
            result = subprocess.run(
                command, shell=True, capture_output=True, text=True,
                timeout=30, cwd=str(self.workspace),
            )
            output = result.stdout.strip()
            if result.returncode != 0:
                output = result.stderr.strip() or output

            self.history.append({"command": command, "output": output[:500]})
            return {"output": output[:2000], "code": result.returncode}
        except subprocess.TimeoutExpired:
            return {"error": "命令执行超时（30s）"}
        except Exception as e:
            return {"error": str(e)}

    def read_file(self, path: str) -> dict:
        """安全读取文件。"""
        full_path = (self.workspace / path).resolve()
        if not str(full_path).startswith(str(self.workspace)):
            return {"error": "不能读取工作目录以外的文件"}

        try:
            content = full_path.read_text(encoding="utf-8", errors="ignore")
            return {"content": content[:5000]}
        except Exception as e:
            return {"error": str(e)}

    def write_file(self, path: str, content: str) -> dict:
        """安全写入文件。"""
        full_path = (self.workspace / path).resolve()
        if not str(full_path).startswith(str(self.workspace)):
            return {"error": "不能写入工作目录以外的文件"}

        if self.require_confirm:
            print(f"\n📝 即将写入：{path}（{len(content)} 字符）")
            ok = input("确认写入？[y/N] ").strip().lower()
            if ok != "y":
                return {"error": "用户取消"}

        try:
            full_path.parent.mkdir(parents=True, exist_ok=True)
            full_path.write_text(content, encoding="utf-8")
            return {"status": "ok", "path": str(full_path.relative_to(self.workspace))}
        except Exception as e:
            return {"error": str(e)}

# agent_with_harness.py
from openai import OpenAI
from harness import AgentHarness
import os, json
from dotenv import load_dotenv
load_dotenv()

client = OpenAI(api_key=os.getenv("DEEPSEEK_API_KEY"), base_url="https://api.deepseek.com/v1")
harness = AgentHarness("./workspace")

TOOLS = [
    {"type": "function", "function": {
        "name": "execute_command", "description": "在沙箱中执行命令。可用命令：ls, cat, grep, find, python, git",
        "parameters": {"type": "object", "properties": {"command": {"type": "string"}}, "required": ["command"]}
    }},
    {"type": "function", "function": {
        "name": "read_file", "description": "读取工作目录中的文件",
        "parameters": {"type": "object", "properties": {"path": {"type": "string"}}, "required": ["path"]}
    }}
]

def run_agent(task):
    messages = [{"role": "system", "content": f"你是一个编程助手。工作目录：{harness.workspace}"}]
    messages.append({"role": "user", "content": task})

    while True:
        resp = client.chat.completions.create(
            model="deepseek-chat", messages=messages, tools=TOOLS, tool_choice="auto"
        )
        msg = resp.choices[0].message

        if not msg.tool_calls:
            return msg.content

        for tc in msg.tool_calls:
            name = tc.function.name
            args = json.loads(tc.function.arguments)

            if name == "execute_command":
                result = harness.execute(args["command"])
            elif name == "read_file":
                result = harness.read_file(args["path"])
            else:
                result = {"error": "Unknown tool"}

            messages.append(msg)
            messages.append({"role": "tool", "tool_call_id": tc.id, "content": json.dumps(result)})

# 使用
result = run_agent("在当前目录创建一个 hello.py，内容为打印 Hello World，然后运行它")
print(result)