add

Dockerfile

@@ -0,0 +1,32 @@
+FROM python:3.11-slim
+
+WORKDIR /app
+
+# Install system dependencies
+RUN apt-get update && apt-get install -y \
+    gcc \
+    curl \
+    && rm -rf /var/lib/apt/lists/*
+
+# Copy requirements and install Python dependencies
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy application code
+COPY app.py token_reader.py entrypoint.sh ./
+
+# Make entrypoint executable
+RUN chmod +x entrypoint.sh
+
+# Create .aws directory structure
+RUN mkdir -p /root/.aws/sso/cache
+
+# Expose port
+EXPOSE 8989
+
+# Health check
+HEALTHCHECK --interval=30s --timeout=30s --start-period=5s --retries=3 \
+    CMD curl -f http://localhost:8989/health || exit 1
+
+# Use entrypoint script for smart startup
+ENTRYPOINT ["./entrypoint.sh"]

app.py

@@ -0,0 +1,305 @@
+import os
+import json
+import time
+import uuid
+import httpx
+from fastapi import FastAPI, HTTPException, Request
+from fastapi.responses import StreamingResponse
+from pydantic import BaseModel, Field
+from typing import List, Optional, Dict, Any
+from dotenv import load_dotenv
+
+# Load environment variables
+load_dotenv()
+
+# Initialize FastAPI app
+app = FastAPI(
+    title="Ki2API - Claude Sonnet 4 OpenAI Compatible API",
+    description="Simple Docker-ready OpenAI-compatible API for Claude Sonnet 4",
+    version="1.0.0"
+)
+
+# Configuration
+API_KEY = os.getenv("API_KEY", "ki2api-key-2024")
+KIRO_ACCESS_TOKEN = os.getenv("KIRO_ACCESS_TOKEN")
+KIRO_REFRESH_TOKEN = os.getenv("KIRO_REFRESH_TOKEN")
+KIRO_BASE_URL = "https://codewhisperer.us-east-1.amazonaws.com/generateAssistantResponse"
+PROFILE_ARN = "arn:aws:codewhisperer:us-east-1:699475941385:profile/EHGA3GRVQMUK"
+
+# Model mapping
+MODEL_NAME = "claude-sonnet-4-20250514"
+CODEWHISPERER_MODEL = "CLAUDE_SONNET_4_20250514_V1_0"
+
+# Pydantic models
+class ChatMessage(BaseModel):
+    role: str
+    content: str
+
+class ChatCompletionRequest(BaseModel):
+    model: str
+    messages: List[ChatMessage]
+    temperature: Optional[float] = 0.7
+    max_tokens: Optional[int] = 4000
+    stream: Optional[bool] = False
+
+class ChatCompletionResponse(BaseModel):
+    id: str = Field(default_factory=lambda: f"chatcmpl-{uuid.uuid4()}")
+    object: str = "chat.completion"
+    created: int = Field(default_factory=lambda: int(time.time()))
+    model: str
+    choices: List[Dict[str, Any]]
+    usage: Dict[str, int]
+
+class ChatCompletionStreamResponse(BaseModel):
+    id: str = Field(default_factory=lambda: f"chatcmpl-{uuid.uuid4()}")
+    object: str = "chat.completion.chunk"
+    created: int = Field(default_factory=lambda: int(time.time()))
+    model: str
+    choices: List[Dict[str, Any]]
+
+# Token management
+class TokenManager:
+    def __init__(self):
+        self.access_token = KIRO_ACCESS_TOKEN
+        self.refresh_token = KIRO_REFRESH_TOKEN
+        self.refresh_url = "https://prod.us-east-1.auth.desktop.kiro.dev/refreshToken"
+
+    async def refresh_tokens(self):
+        if not self.refresh_token:
+            return None
+        
+        try:
+            async with httpx.AsyncClient() as client:
+                response = await client.post(
+                    self.refresh_url,
+                    json={"refreshToken": self.refresh_token},
+                    timeout=30
+                )
+                response.raise_for_status()
+                
+                data = response.json()
+                self.access_token = data.get("accessToken")
+                return self.access_token
+        except Exception as e:
+            print(f"Token refresh failed: {e}")
+            return None
+
+    def get_token(self):
+        return self.access_token
+
+token_manager = TokenManager()
+
+# Build CodeWhisperer request
+def build_codewhisperer_request(messages: List[ChatMessage]):
+    conversation_id = str(uuid.uuid4())
+    
+    # Extract system prompt and user messages
+    system_prompt = ""
+    user_messages = []
+    
+    for msg in messages:
+        if msg.role == "system":
+            system_prompt = msg.content
+        else:
+            user_messages.append(msg)
+    
+    if not user_messages:
+        raise HTTPException(status_code=400, detail="No user messages found")
+    
+    # Build history
+    history = []
+    for i in range(0, len(user_messages) - 1, 2):
+        if i + 1 < len(user_messages):
+            history.append({
+                "userInputMessage": {
+                    "content": user_messages[i].content,
+                    "modelId": CODEWHISPERER_MODEL,
+                    "origin": "AI_EDITOR"
+                }
+            })
+            history.append({
+                "assistantResponseMessage": {
+                    "content": user_messages[i + 1].content,
+                    "toolUses": []
+                }
+            })
+    
+    # Build current message
+    current_message = user_messages[-1]
+    content = current_message.content
+    if system_prompt:
+        content = f"{system_prompt}\n\n{content}"
+    
+    return {
+        "profileArn": PROFILE_ARN,
+        "conversationState": {
+            "chatTriggerType": "MANUAL",
+            "conversationId": conversation_id,
+            "currentMessage": {
+                "userInputMessage": {
+                    "content": content,
+                    "modelId": CODEWHISPERER_MODEL,
+                    "origin": "AI_EDITOR",
+                    "userInputMessageContext": {}
+                }
+            },
+            "history": history
+        }
+    }
+
+# Make API call to Kiro/CodeWhisperer
+async def call_kiro_api(messages: List[ChatMessage], stream: bool = False):
+    token = token_manager.get_token()
+    if not token:
+        raise HTTPException(status_code=401, detail="No access token available")
+    
+    request_data = build_codewhisperer_request(messages)
+    
+    headers = {
+        "Authorization": f"Bearer {token}",
+        "Content-Type": "application/json",
+        "Accept": "text/event-stream" if stream else "application/json"
+    }
+    
+    try:
+        async with httpx.AsyncClient() as client:
+            response = await client.post(
+                KIRO_BASE_URL,
+                headers=headers,
+                json=request_data,
+                timeout=120
+            )
+            
+            if response.status_code == 403:
+                # Try to refresh token
+                new_token = await token_manager.refresh_tokens()
+                if new_token:
+                    headers["Authorization"] = f"Bearer {new_token}"
+                    response = await client.post(
+                        KIRO_BASE_URL,
+                        headers=headers,
+                        json=request_data,
+                        timeout=120
+                    )
+            
+            response.raise_for_status()
+            return response
+            
+    except Exception as e:
+        raise HTTPException(status_code=503, detail=f"API call failed: {str(e)}")
+
+# API endpoints
+@app.get("/v1/models")
+async def list_models():
+    return {
+        "object": "list",
+        "data": [
+            {
+                "id": MODEL_NAME,
+                "object": "model",
+                "created": int(time.time()),
+                "owned_by": "ki2api"
+            }
+        ]
+    }
+
+@app.post("/v1/chat/completions")
+async def create_chat_completion(request: ChatCompletionRequest):
+    if request.model != MODEL_NAME:
+        raise HTTPException(status_code=400, detail=f"Only {MODEL_NAME} is supported")
+    
+    if request.stream:
+        return await create_streaming_response(request)
+    else:
+        return await create_non_streaming_response(request)
+
+async def create_non_streaming_response(request: ChatCompletionRequest):
+    response = await call_kiro_api(request.messages, stream=False)
+    
+    # Parse response
+    response_text = response.text
+    
+    return ChatCompletionResponse(
+        model=MODEL_NAME,
+        choices=[{
+            "index": 0,
+            "message": {
+                "role": "assistant",
+                "content": response_text
+            },
+            "finish_reason": "stop"
+        }],
+        usage={
+            "prompt_tokens": 0,
+            "completion_tokens": 0,
+            "total_tokens": 0
+        }
+    )
+
+async def create_streaming_response(request: ChatCompletionRequest):
+    response = await call_kiro_api(request.messages, stream=True)
+    
+    async def generate():
+        # Send initial response
+        initial_chunk = {
+            'id': f'chatcmpl-{uuid.uuid4()}',
+            'object': 'chat.completion.chunk',
+            'created': int(time.time()),
+            'model': MODEL_NAME,
+            'choices': [{
+                'index': 0,
+                'delta': {'role': 'assistant'},
+                'finish_reason': None
+            }]
+        }
+        yield f"data: {json.dumps(initial_chunk)}\n\n"
+        
+        # Read response and stream content
+        content = ""
+        async for line in response.aiter_lines():
+            if line.startswith('data: '):
+                try:
+                    data = json.loads(line[6:])
+                    if 'content' in data:
+                        content += data['content']
+                        chunk = {
+                            'id': f'chatcmpl-{uuid.uuid4()}',
+                            'object': 'chat.completion.chunk',
+                            'created': int(time.time()),
+                            'model': MODEL_NAME,
+                            'choices': [{
+                                'index': 0,
+                                'delta': {'content': data['content']},
+                                'finish_reason': None
+                            }]
+                        }
+                        yield f"data: {json.dumps(chunk)}\n\n"
+                except:
+                    continue
+        
+        # Send final response
+        final_chunk = {
+            'id': f'chatcmpl-{uuid.uuid4()}',
+            'object': 'chat.completion.chunk',
+            'created': int(time.time()),
+            'model': MODEL_NAME,
+            'choices': [{
+                'index': 0,
+                'delta': {},
+                'finish_reason': 'stop'
+            }]
+        }
+        yield f"data: {json.dumps(final_chunk)}\n\n"
+        
+        yield "data: [DONE]\n\n"
+    
+    return StreamingResponse(generate(), media_type="text/event-stream")
+
+# Health check
+@app.get("/health")
+async def health_check():
+    return {"status": "ok", "service": "ki2api"}
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8989)

docker-compose.yml

@@ -0,0 +1,19 @@
+version: '3.8'
+
+services:
+  ki2api:
+    build: .
+    ports:
+      - "8989:8989"
+    environment:
+      - API_KEY=ki2api-key-2024
+    volumes:
+      # 自动挂载token目录，实现零配置
+      - ~/.aws/sso/cache:/root/.aws/sso/cache:ro
+    restart: unless-stopped
+    healthcheck:
+      test: ["CMD", "curl", "-f", "http://localhost:8989/health"]
+      interval: 30s
+      timeout: 10s
+      retries: 3
+      start_period: 40s

entrypoint.sh

@@ -0,0 +1,41 @@
+#!/bin/bash
+
+# 智能Docker入口脚本
+# 自动读取token并启动服务
+
+echo "🚀 Ki2API 启动中..."
+
+# 检查是否存在token文件
+TOKEN_FILE="/root/.aws/sso/cache/kiro-auth-token.json"
+
+if [ -f "$TOKEN_FILE" ]; then
+    echo "📁 发现token文件，正在读取..."
+    
+    # 运行token读取脚本
+    python token_reader.py
+    
+    if [ $? -eq 0 ]; then
+        echo "✅ Token配置完成"
+    else
+        echo "⚠️  Token读取失败，继续启动（需要手动配置token）"
+    fi
+else
+    echo "⚠️  未找到token文件: $TOKEN_FILE"
+    echo "请确保已登录Kiro，或手动设置环境变量"
+fi
+
+# 检查环境变量
+if [ -z "$KIRO_ACCESS_TOKEN" ] || [ -z "$KIRO_REFRESH_TOKEN" ]; then
+    echo "⚠️  环境变量未设置，尝试从.env文件加载..."
+    if [ -f ".env" ]; then
+        export $(cat .env | xargs)
+        echo "✅ 已从.env文件加载token"
+    else
+        echo "❌ 未找到token配置，服务可能无法正常工作"
+        echo "请设置 KIRO_ACCESS_TOKEN 和 KIRO_REFRESH_TOKEN 环境变量"
+    fi
+fi
+
+# 启动应用
+echo "🎯 启动FastAPI服务..."
+exec python app.py

requirements.txt

@@ -0,0 +1,5 @@
+fastapi==0.104.1
+uvicorn[standard]==0.24.0
+httpx==0.25.2
+python-dotenv==1.0.0
+pydantic==2.5.0

token_reader.py

@@ -0,0 +1,77 @@
+#!/usr/bin/env python3
+"""
+自动读取Kiro token的脚本
+在Docker容器启动时自动读取宿主机的token文件
+"""
+
+import os
+import json
+import sys
+from pathlib import Path
+
+def get_token_file_path():
+    """获取token文件路径"""
+    home = Path.home()
+    return home / ".aws" / "sso" / "cache" / "kiro-auth-token.json"
+
+def read_tokens():
+    """读取token文件"""
+    token_file = get_token_file_path()
+    
+    if not token_file.exists():
+        print(f"❌ Token文件不存在: {token_file}")
+        print("请确保已登录Kiro，或手动创建token文件")
+        return None, None
+    
+    try:
+        with open(token_file, 'r', encoding='utf-8') as f:
+            data = json.load(f)
+        
+        access_token = data.get('accessToken')
+        refresh_token = data.get('refreshToken')
+        
+        if not access_token or not refresh_token:
+            print("❌ Token文件格式错误，缺少accessToken或refreshToken")
+            return None, None
+            
+        return access_token, refresh_token
+        
+    except json.JSONDecodeError:
+        print("❌ Token文件JSON格式错误")
+        return None, None
+    except Exception as e:
+        print(f"❌ 读取token文件失败: {e}")
+        return None, None
+
+def create_env_file(access_token, refresh_token):
+    """创建.env文件"""
+    env_content = f"""# Kiro Token配置
+# 自动生成于 {os.path.basename(__file__)}
+KIRO_ACCESS_TOKEN={access_token}
+KIRO_REFRESH_TOKEN={refresh_token}
+"""
+    
+    with open('.env', 'w', encoding='utf-8') as f:
+        f.write(env_content)
+    
+    print("✅ .env文件已创建/更新")
+
+def main():
+    """主函数"""
+    print("🔍 正在读取Kiro token...")
+    
+    access_token, refresh_token = read_tokens()
+    
+    if access_token and refresh_token:
+        create_env_file(access_token, refresh_token)
+        print("✅ Token读取成功，服务即将启动...")
+        return 0
+    else:
+        print("❌ 无法获取token，请检查：")
+        print("1. 是否已登录Kiro (https://kiro.dev)")
+        print("2. token文件是否存在: ~/.aws/sso/cache/kiro-auth-token.json")
+        print("3. 或手动创建.env文件并设置token")
+        return 1
+
+if __name__ == "__main__":
+    sys.exit(main())