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

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

Agent 不只是调用 LLM，还需要执行命令、读写文件、调 API。但让 AI 直接操作你的电脑是有风险的。Agent Harness 就是解决这个问题的——给 Agent 一个受限的"沙箱"。

Harness 做什么

Agent 想做什么 → Harness 检查和审批 → 执行 → 返回结果 核心能力： ├─ 🛡️ 沙箱执行：限制 Agent 只能操作指定目录 ├─ 📋 命令白名单：只允许安全的命令 └─ ✅ 人工审批：危险操作需要确认

核心代码

# harness.pyimportsubprocess,os,shlexfrompathlibimportPathclassAgentHarness:"""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=[]defexecute(self,command:str)->dict:"""在沙箱中执行命令。"""cmd_parts=shlex.split(command)ifnotcmd_parts:return{"error":"Empty command"}# 安全检查base_cmd=cmd_parts[0]ifbase_cmdnotinself.ALLOWED_COMMANDS:return{"error":f"命令{base_cmd}不在白名单中"}# 危险操作需要确认dangerous=any(kincommandforkin["rm ","delete","DROP",">",">>"])ifdangerousandself.require_confirm:print(f"\n⚠️ 危险操作：{command}")ok=input("确认执行？[y/N] ").strip().lower()ifok!="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()ifresult.returncode!=0:output=result.stderr.strip()oroutput self.history.append({"command":command,"output":output[:500]})return{"output":output[:2000],"code":result.returncode}exceptsubprocess.TimeoutExpired:return{"error":"命令执行超时（30s）"}exceptExceptionase:return{"error":str(e)}defread_file(self,path:str)->dict:"""安全读取文件。"""full_path=(self.workspace/path).resolve()ifnotstr(full_path).startswith(str(self.workspace)):return{"error":"不能读取工作目录以外的文件"}try:content=full_path.read_text(encoding="utf-8",errors="ignore")return{"content":content[:5000]}exceptExceptionase:return{"error":str(e)}defwrite_file(self,path:str,content:str)->dict:"""安全写入文件。"""full_path=(self.workspace/path).resolve()ifnotstr(full_path).startswith(str(self.workspace)):return{"error":"不能写入工作目录以外的文件"}ifself.require_confirm:print(f"\n📝 即将写入：{path}（{len(content)}字符）")ok=input("确认写入？[y/N] ").strip().lower()ifok!="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))}exceptExceptionase:return{"error":str(e)}

Agent + Harness 组合使用

# agent_with_harness.pyfromopenaiimportOpenAIfromharnessimportAgentHarnessimportos,jsonfromdotenvimportload_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"]}}}]defrun_agent(task):messages=[{"role":"system","content":f"你是一个编程助手。工作目录：{harness.workspace}"}]messages.append({"role":"user","content":task})whileTrue:resp=client.chat.completions.create(model="deepseek-chat",messages=messages,tools=TOOLS,tool_choice="auto")msg=resp.choices[0].messageifnotmsg.tool_calls:returnmsg.contentfortcinmsg.tool_calls:name=tc.function.name args=json.loads(tc.function.arguments)ifname=="execute_command":result=harness.execute(args["command"])elifname=="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)