server-configs/openclaw_openai_api.py

#!/usr/bin/env python3
"""
OpenClaw OpenAI Compatible API Server
让笔记软件像调用 OpenAI 一样调用 OpenClaw
"""

import json
import os
import http.server
import socketserver
import uuid
import time
from urllib.parse import urlparse
from datetime import datetime

# ============= 配置 =============
API_KEY = os.getenv("OPENCLAW_API_KEY", "oc_gFRPndPnFxa3COTX0tAdh2dnN92UmR8U")
PORT = 8080
AGENT_URL = os.getenv("OPENCLAW_AGENT_URL", "http://localhost:11434")

# ============= 工具函数 =============
def generate_id():
    return f"chatcmpl-{uuid.uuid4().hex[:24]}"

def current_timestamp():
    return int(time.time())

# ============= API 处理 =============
class OpenAIHandler(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.send_header('Access-Control-Allow-Methods', 'GET, POST, OPTIONS')
        self.send_header('Access-Control-Allow-Headers', 'Authorization, Content-Type')
        self.end_headers()
        self.wfile.write(json.dumps(data, ensure_ascii=False).encode('utf-8'))
    
    def do_OPTIONS(self):
        self.send_response(200)
        self.end_headers()
    
    def do_GET(self):
        parsed = urlparse(self.path)
        
        # 认证
        auth = self.headers.get('Authorization', '')
        if not auth.startswith(f'Bearer {API_KEY}') and auth != API_KEY:
            self.send_json({"error": {"message": "Invalid API Key", "type": "authentication_error", "code": "invalid_api_key"}}, status=401)
            return
        
        if parsed.path == '/health' or parsed.path == '/':
            self.send_json({
                "status": "healthy",
                "timestamp": datetime.now().isoformat(),
                "version": "1.0.0"
            })
        elif parsed.path == '/v1/models':
            self.send_json({
                "object": "list",
                "data": [{
                    "id": "openclaw",
                    "object": "model",
                    "created": current_timestamp(),
                    "owned_by": "openclaw"
                }]
            })
        else:
            self.send_json({"error": "Not found"}, status=404)
    
    def do_POST(self):
        parsed = urlparse(self.path)
        
        # 认证
        auth = self.headers.get('Authorization', '')
        if not auth.startswith(f'Bearer {API_KEY}') and auth != API_KEY:
            self.send_json({"error": {"message": "Invalid API Key", "type": "authentication_error", "code": "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:
            self.send_json({"error": "Invalid JSON"}, status=400)
            return
        
        # Chat Completions API
        if parsed.path == '/v1/chat/completions':
            model = data.get('model', 'openclaw')
            messages = data.get('messages', [])
            stream = data.get('stream', False)
            user_message = ""
            
            for msg in messages:
                if msg.get('role') == 'user':
                    user_message = msg.get('content', '')
                    break
            
            if not user_message:
                self.send_json({"error": {"message": "No user message", "type": "invalid_request_error"}}, status=400)
                return
            
            # 这里可以调用实际的 agent
            # 目前返回占位回复
            response_content = f"收到消息: {user_message}\n\n(API 已收到，正在处理...)"
            
            if stream:
                # 流式响应
                self.send_response(200)
                self.send_header('Content-Type', 'text/event-stream')
                self.send_header('Cache-Control', 'no-cache')
                self.send_header('Connection', 'keep-alive')
                self.end_headers()
                
                chunks = response_content.split(' ')
                for i, chunk in enumerate(chunks):
                    chunk_data = {
                        "id": generate_id(),
                        "object": "chat.completion.chunk",
                        "created": current_timestamp(),
                        "model": model,
                        "choices": [{
                            "index": 0,
                            "delta": {"content": chunk + " "},
                            "finish_reason": None if i < len(chunks) - 1 else "stop"
                        }]
                    }
                    self.wfile.write(f"data: {json.dumps(chunk_data, ensure_ascii=False)}\n\n".encode('utf-8'))
                self.wfile.write(b"data: [DONE]\n\n")
            else:
                # 非流式响应
                response = {
                    "id": generate_id(),
                    "object": "chat.completion",
                    "created": current_timestamp(),
                    "model": model,
                    "choices": [{
                        "index": 0,
                        "message": {
                            "role": "assistant",
                            "content": response_content
                        },
                        "finish_reason": "stop"
                    }],
                    "usage": {
                        "prompt_tokens": len(user_message.split()),
                        "completion_tokens": len(response_content.split()),
                        "total_tokens": len(user_message.split()) + len(response_content.split())
                    },
                    "system_fingerprint": "fp_openclaw_1"
                }
                self.send_json(response)
        
        # Completions API (非聊天)
        elif parsed.path == '/v1/completions':
            prompt = data.get('prompt', '')
            response = {
                "id": generate_id(),
                "object": "text_completion",
                "created": current_timestamp(),
                "model": data.get('model', 'openclaw-text'),
                "choices": [{
                    "text": f"响应: {prompt}",
                    "index": 0,
                    "finish_reason": "stop"
                }],
                "usage": {
                    "prompt_tokens": 10,
                    "completion_tokens": 20,
                    "total_tokens": 30
                }
            }
            self.send_json(response)
        
        else:
            self.send_json({"error": "Not found"}, status=404)

# ============= 启动 =============
if __name__ == '__main__':
    print(f"OpenClaw OpenAI Compatible API v1.0")
    print(f"API Key: {API_KEY}")
    print(f"Port: {PORT}")
    print(f"\nEndpoints:")
    print(f"  GET  /health              - Health check")
    print(f"  GET  /v1/models           - List models")
    print(f"  POST /v1/chat/completions - Chat completion")
    print(f"  POST /v1/completions      - Text completion")
    print(f"\n兼容 OpenAI SDK 用法:")
    print(f'  client = OpenAI(api_key="your-key", base_url="http://localhost:{PORT}/v1")')
    print(f'  response = client.chat.completions.create(model="openclaw", messages=[{{"role":"user","content":"hi"}}])')
    print(f"\nCurl 示例:")
    print(f'  curl -X POST http://localhost:{PORT}/v1/chat/completions \\')
    print(f'    -H "Authorization: Bearer {API_KEY}" \\')
    print(f'    -H "Content-Type: application/json" \\')
    print(f'    -d \'{{"model":"openclaw","messages":[{{"role":"user","content":"你好"}}]}}\'')
    
    with socketserver.TCPServer(("", PORT), OpenAIHandler) as httpd:
        print(f"\nServer running on http://localhost:{PORT}")
        httpd.serve_forever()