139 lines
5.3 KiB
Python
Executable File
139 lines
5.3 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
OpenClaw API Server - 纯 Python 实现
|
|
无需外部依赖
|
|
"""
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import http.server
|
|
import socketserver
|
|
from urllib.parse import urlparse, parse_qs
|
|
from datetime import datetime
|
|
|
|
# ============= 配置 =============
|
|
API_KEY = os.getenv("OPENCLAW_API_KEY", "your-api-key-change-me")
|
|
PORT = 8000
|
|
|
|
# ============= 工具函数 =============
|
|
def run_shell(command, timeout=30):
|
|
try:
|
|
result = subprocess.run(command, shell=True, capture_output=True, text=True, timeout=timeout)
|
|
return {"returncode": result.returncode, "stdout": result.stdout, "stderr": result.stderr}
|
|
except Exception as e:
|
|
return {"error": str(e)}
|
|
|
|
def read_file(path):
|
|
try:
|
|
with open(path, 'r', encoding='utf-8') as f:
|
|
return f.read()
|
|
except Exception as e:
|
|
return f"Error: {e}"
|
|
|
|
def write_file(path, content):
|
|
try:
|
|
os.makedirs(os.path.dirname(path), exist_ok=True)
|
|
with open(path, 'w', encoding='utf-8') as f:
|
|
f.write(content)
|
|
return {"status": "success", "path": path}
|
|
except Exception as e:
|
|
return {"status": "error", "message": str(e)}
|
|
|
|
def list_files(path="/root/.openclaw/workspace"):
|
|
try:
|
|
items = []
|
|
for item in os.listdir(path):
|
|
item_path = os.path.join(path, item)
|
|
items.append({"name": item, "type": "directory" if os.path.isdir(item_path) else "file"})
|
|
return {"path": path, "items": items}
|
|
except Exception as e:
|
|
return {"error": str(e)}
|
|
|
|
# ============= API 处理 =============
|
|
class OpenClawHandler(http.server.BaseHTTPRequestHandler):
|
|
def log_message(self, format, *args):
|
|
pass # 禁用日志
|
|
|
|
def send_json(self, data, status=200):
|
|
self.send_response(status)
|
|
self.send_header('Content-Type', 'application/json')
|
|
self.send_header('Access-Control-Allow-Origin', '*')
|
|
self.end_headers()
|
|
self.wfile.write(json.dumps(data, ensure_ascii=False).encode('utf-8'))
|
|
|
|
def do_GET(self):
|
|
parsed = urlparse(self.path)
|
|
path = parsed.path
|
|
query = parse_qs(parsed.query)
|
|
|
|
# 认证
|
|
api_key = self.headers.get('X-API-Key') or self.headers.get('Authorization', '').replace('Bearer ', '')
|
|
if api_key != API_KEY:
|
|
self.send_json({"error": "Invalid API Key"}, status=401)
|
|
return
|
|
|
|
if path == '/' or path == '/health':
|
|
self.send_json({"status": "healthy", "timestamp": datetime.now().isoformat()})
|
|
elif path == '/api/v1/files':
|
|
self.send_json(list_files(query.get('path', ['/root/.openclaw/workspace'])[0]))
|
|
elif path.startswith('/api/v1/memory/'):
|
|
file_path = path.replace('/api/v1/memory/', '')
|
|
self.send_json({"path": file_path, "content": read_file(f"/root/.openclaw/workspace/{file_path}")})
|
|
else:
|
|
self.send_json({"error": "Not found"}, status=404)
|
|
|
|
def do_POST(self):
|
|
parsed = urlparse(self.path)
|
|
path = parsed.path
|
|
|
|
# 认证
|
|
api_key = self.headers.get('X-API-Key') or self.headers.get('Authorization', '').replace('Bearer ', '')
|
|
if api_key != API_KEY:
|
|
self.send_json({"error": "Invalid API Key"}, status=401)
|
|
return
|
|
|
|
length = int(self.headers.get('Content-Length', 0))
|
|
body = self.rfile.read(length).decode('utf-8')
|
|
|
|
try:
|
|
data = json.loads(body) if body else {}
|
|
except:
|
|
data = {}
|
|
|
|
if path == '/api/v1/exec':
|
|
result = run_shell(data.get('command', ''), data.get('timeout', 30))
|
|
self.send_json(result)
|
|
elif path == '/api/v1/git/commit':
|
|
msg = data.get('message', '')
|
|
files = data.get('files', [])
|
|
workspace = '/root/.openclaw/workspace/openclaw-memory'
|
|
for f in files:
|
|
subprocess.run(f"cp {workspace}/../{f} {workspace}/ 2>/dev/null; cd {workspace} && git add {f} 2>/dev/null", shell=True)
|
|
subprocess.run(f"cd {workspace} && git commit -m '{msg}' && git push origin main", shell=True)
|
|
self.send_json({"status": "committed", "files": files, "message": msg})
|
|
elif path == '/api/v1/memory/write':
|
|
result = write_file(data.get('path', ''), data.get('content', ''))
|
|
self.send_json(result)
|
|
else:
|
|
self.send_json({"error": "Not found"}, status=404)
|
|
|
|
# ============= 启动 =============
|
|
if __name__ == '__main__':
|
|
print(f"OpenClaw API Server v1.0")
|
|
print(f"API Key: {API_KEY}")
|
|
print(f"Port: {PORT}")
|
|
print(f"Docs: http://localhost:{PORT}/ (GET)")
|
|
print(f"\nEndpoints:")
|
|
print(f" GET /health - Health check")
|
|
print(f" GET /api/v1/files?path=/path - List files")
|
|
print(f" GET /api/v1/memory/{file} - Read memory file")
|
|
print(f" POST /api/v1/exec - Run command")
|
|
print(f" POST /api/v1/git/commit - Git commit & push")
|
|
print(f" POST /api/v1/memory/write - Write file")
|
|
print(f"\nHeader: X-API-Key: {API_KEY}")
|
|
|
|
with socketserver.TCPServer(("", PORT), OpenClawHandler) as httpd:
|
|
print(f"\nServer running on http://localhost:{PORT}")
|
|
httpd.serve_forever()
|