"""
DevEnv - Development Environment API for AI Agent
Provides: command execution, file operations, web scraping, health checks
"""
import subprocess
import base64
import os
import json
import tempfile
from flask import Flask, request, jsonify, render_template_string
app = Flask(__name__)
API_TOKEN = os.environ.get("API_TOKEN", "2004")
WORKSPACE = "/home/user/workspace"
def check_auth():
auth = request.headers.get("Authorization", "")
if auth == f"Bearer {API_TOKEN}":
return True
token = request.args.get("token", "")
if token == API_TOKEN:
return True
return False
# ── Dashboard ──
DASHBOARD_HTML = """
🖥️ DevEnv - Agent Development Environment
📊 Status
● Online
Ubuntu 22.04 • Python 3 • Node.js • Git • Chromium
🔌 API Endpoints
- POST
/api/exec — Execute shell commands
- POST
/api/read — Read file contents
- POST
/api/write — Write/create files
- POST
/api/browse — Fetch web page content
- GET
/api/health — Health check
"""
@app.route("/")
def dashboard():
return render_template_string(DASHBOARD_HTML)
@app.route("/api/health")
def health():
return jsonify({"status": "ok", "service": "devenv", "tools": ["bash", "python", "node", "git", "chromium"]})
@app.route("/api/exec", methods=["POST"])
def exec_cmd():
if not check_auth():
return jsonify({"error": "unauthorized"}), 401
data = request.get_json() or {}
cmd = data.get("command", "echo hello")
timeout = min(data.get("timeout", 30), 120)
cwd = data.get("cwd", WORKSPACE)
try:
result = subprocess.run(
cmd, shell=True, capture_output=True, text=True,
timeout=timeout, cwd=cwd,
env={**os.environ, "HOME": "/home/user"}
)
return jsonify({
"stdout": result.stdout[-10000:],
"stderr": result.stderr[-3000:],
"returncode": result.returncode
})
except subprocess.TimeoutExpired:
return jsonify({"error": "timeout", "stdout": "", "stderr": f"Command timed out after {timeout}s"})
except Exception as e:
return jsonify({"error": str(e)})
@app.route("/api/read", methods=["POST"])
def read_file():
if not check_auth():
return jsonify({"error": "unauthorized"}), 401
data = request.get_json() or {}
path = data.get("path", "")
if not path:
return jsonify({"error": "path required"})
try:
with open(path, "r") as f:
content = f.read(100000) # 100KB max
return jsonify({"content": content, "path": path})
except Exception as e:
return jsonify({"error": str(e)})
@app.route("/api/write", methods=["POST"])
def write_file():
if not check_auth():
return jsonify({"error": "unauthorized"}), 401
data = request.get_json() or {}
path = data.get("path", "")
content = data.get("content", "")
if not path:
return jsonify({"error": "path required"})
try:
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
with open(path, "w") as f:
f.write(content)
return jsonify({"status": "ok", "path": path, "bytes": len(content)})
except Exception as e:
return jsonify({"error": str(e)})
@app.route("/api/browse", methods=["POST"])
def browse():
if not check_auth():
return jsonify({"error": "unauthorized"}), 401
data = request.get_json() or {}
url = data.get("url", "")
if not url:
return jsonify({"error": "url required"})
try:
import requests as req
r = req.get(url, timeout=15, headers={"User-Agent": "Mozilla/5.0"})
from bs4 import BeautifulSoup
soup = BeautifulSoup(r.text, "html.parser")
# Remove scripts and styles
for s in soup(["script", "style"]):
s.decompose()
text = soup.get_text(separator="\n", strip=True)
return jsonify({
"url": url,
"status_code": r.status_code,
"title": soup.title.string if soup.title else "",
"text": text[:15000],
"html_length": len(r.text)
})
except Exception as e:
return jsonify({"error": str(e)})
if __name__ == "__main__":
os.makedirs(WORKSPACE, exist_ok=True)
print("🖥️ DevEnv API starting on port 7860...")
app.run(host="0.0.0.0", port=7860)