diff --git a/README.md b/README.md index 28e0eae..7144312 100644 --- a/README.md +++ b/README.md @@ -4,10 +4,13 @@ Dynagate is a lightweight, high-performance LLM gateway written in Go that acts ## Features -1. **OpenAI-compatible endpoints**: +1. **Multi-API Protocol Endpoints (Stateless)**: - `/v1/models` (GET): Dynamically lists unique active model IDs. - `/v1/chat/completions` (POST): Proxies non-streaming and streaming (`text/event-stream`) completions. + - `/v1/responses` (POST): Stateless-style OpenAI Responses API endpoint (automatically routed to upstream chat completions endpoints). + - `/v1/messages` (POST): Anthropic-compatible Messages API endpoint (automatically routed to upstream chat completions endpoints). - `/v1/images/generations` (POST): Proxies image generation requests. + > **Note**: Dynagate operates in a completely stateless manner across all API formats (`/v1/chat/completions`, `/v1/responses`, `/v1/messages`). Dynagate does not store or persist server-side conversation state. Clients must include the full conversation history in each request for multi-turn dialogues. 2. **Zero-downtime live-reloading**: - Watches `models.csv` (overridable via command-line flags) continuously using a background thread and automatically reloads configuration updates without dropping active connections. @@ -112,6 +115,15 @@ The gateway maps columns dynamically by looking at the header row. If no header - **`"-blank-"`**: The gateway will attach exactly `Authorization: Bearer` without any key appended. - **Any other string**: The gateway will attach `Authorization: Bearer `. +### Stateless Operation & Multi-Turn Conversations + +Dynagate is designed as a lightweight, zero-state proxy gateway. All supported request formats (`/v1/chat/completions`, `/v1/responses`, `/v1/messages`) are processed statelessly without storing session context or conversation state on disk or in memory. + +To maintain context in multi-turn conversations, clients **must include the entire conversation history** in every request payload: +- **OpenAI Chat Completions (`/v1/chat/completions`)**: Pass the full array of system, user, and assistant messages in `messages`. +- **OpenAI Responses API (`/v1/responses`)**: Pass previous conversation turns in `input` (or `messages`) along with `instructions`. +- **Anthropic Messages API (`/v1/messages`)**: Pass the full history of user and assistant messages in `messages` along with `system`. + ### Manual testing with curl #### 1. Model listing (authenticated) @@ -142,6 +154,30 @@ curl -i -X POST http://localhost:8080/v1/chat/completions \ }' ``` +#### 4. OpenAI-compatible Responses API (`/v1/responses`) +```bash +curl -i -X POST http://localhost:8080/v1/responses \ + -H "Authorization: Bearer my-secure-gateway-token" \ + -H "Content-Type: application/json" \ + -d '{ + "model": "gpt-4o", + "instructions": "Be concise", + "input": "Explain relativity in one sentence" + }' +``` + +#### 5. Anthropic-compatible Messages API (`/v1/messages`) +```bash +curl -i -X POST http://localhost:8080/v1/messages \ + -H "Authorization: Bearer my-secure-gateway-token" \ + -H "Content-Type: application/json" \ + -d '{ + "model": "claude-3-5-sonnet", + "system": "You are a helpful assistant", + "messages": [{"role": "user", "content": "Hello world"}] + }' +``` + ## Troubleshooting ### Dynamic loading failures diff --git a/gateway_test.go b/gateway_test.go index e57af3f..3a0c2b6 100644 --- a/gateway_test.go +++ b/gateway_test.go @@ -988,6 +988,285 @@ func TestFibonacciBackoff(t *testing.T) { } } +func TestAnthropicMessagesEndpoint(t *testing.T) { + var receivedOpenAIBody map[string]any + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/v1/chat/completions" { + t.Errorf("Expected request to be routed to /v1/chat/completions, got %s", r.URL.Path) + } + + var body map[string]any + _ = json.NewDecoder(r.Body).Decode(&body) + receivedOpenAIBody = body + + isStream, _ := body["stream"].(bool) + w.Header().Set("Content-Type", "application/json") + if isStream { + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + flusher, _ := w.(http.Flusher) + _, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"Hello \"}}]}\n\n")) + flusher.Flush() + _, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"Anthropic!\"},\"finish_reason\":\"stop\"}]}\n\n")) + flusher.Flush() + _, _ = w.Write([]byte("data: [DONE]\n\n")) + flusher.Flush() + return + } + + w.WriteHeader(http.StatusOK) + resp := map[string]any{ + "id": "chatcmpl-test", + "model": "claude-model", + "choices": []any{ + map[string]any{ + "index": 0, + "message": map[string]any{ + "role": "assistant", + "content": "Hello from Anthropic proxy", + }, + "finish_reason": "stop", + }, + }, + "usage": map[string]any{ + "prompt_tokens": 10, + "completion_tokens": 15, + "total_tokens": 25, + }, + } + _ = json.NewEncoder(w).Encode(resp) + })) + defer server.Close() + + configs := []ModelConfig{ + {Model: "claude-model", Key: "test-key", Endpoint: server.URL}, + } + cm := &ConfigManager{ + configs: configs, + uniqueModels: []string{"claude-model"}, + } + handler := handleMessages(cm, nil, 0) + + // 1. Non-streaming test + t.Run("Non-streaming Anthropic message", func(t *testing.T) { + reqObj := map[string]any{ + "model": "claude-model", + "system": "You are a helpful bot", + "messages": []any{ + map[string]any{ + "role": "user", + "content": "Hello", + }, + }, + "max_tokens": 100, + } + reqBytes, _ := json.Marshal(reqObj) + req := httptest.NewRequest("POST", "/v1/messages", bytes.NewReader(reqBytes)) + req.Header.Set("Content-Type", "application/json") + w := httptest.NewRecorder() + + handler.ServeHTTP(w, req) + + if w.Result().StatusCode != http.StatusOK { + t.Fatalf("Expected 200 OK, got %d", w.Result().StatusCode) + } + + var anthropicResp map[string]any + _ = json.NewDecoder(w.Body).Decode(&anthropicResp) + + if anthropicResp["type"] != "message" { + t.Errorf("Expected response type 'message', got %v", anthropicResp["type"]) + } + if anthropicResp["role"] != "assistant" { + t.Errorf("Expected role 'assistant', got %v", anthropicResp["role"]) + } + content, ok := anthropicResp["content"].([]any) + if !ok || len(content) == 0 { + t.Fatalf("Expected non-empty content array in Anthropic response, got %+v", anthropicResp) + } + textVal := content[0].(map[string]any)["text"].(string) + if textVal != "Hello from Anthropic proxy" { + t.Errorf("Unexpected text in response: %s", textVal) + } + + // Verify system message was prepended into OpenAI messages format + openAIMsgs, ok := receivedOpenAIBody["messages"].([]any) + if !ok || len(openAIMsgs) != 2 { + t.Fatalf("Expected 2 messages (system + user) in OpenAI body, got %+v", receivedOpenAIBody) + } + if openAIMsgs[0].(map[string]any)["role"] != "system" { + t.Errorf("Expected first message to be system prompt") + } + }) + + // 2. Streaming test + t.Run("Streaming Anthropic message", func(t *testing.T) { + reqObj := map[string]any{ + "model": "claude-model", + "stream": true, + "messages": []any{ + map[string]any{ + "role": "user", + "content": "Stream me", + }, + }, + } + reqBytes, _ := json.Marshal(reqObj) + req := httptest.NewRequest("POST", "/v1/messages", bytes.NewReader(reqBytes)) + req.Header.Set("Content-Type", "application/json") + w := httptest.NewRecorder() + + handler.ServeHTTP(w, req) + + if w.Result().StatusCode != http.StatusOK { + t.Fatalf("Expected 200 OK, got %d", w.Result().StatusCode) + } + + bodyStr := w.Body.String() + if !strings.Contains(bodyStr, "event: message_start") || !strings.Contains(bodyStr, "event: content_block_delta") || !strings.Contains(bodyStr, "event: message_stop") { + t.Errorf("Expected Anthropic SSE stream events in response, got: %s", bodyStr) + } + }) +} + +func TestOpenAIResponsesEndpoint(t *testing.T) { + var receivedOpenAIBody map[string]any + + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/v1/chat/completions" { + t.Errorf("Expected request to be routed to /v1/chat/completions, got %s", r.URL.Path) + } + + var body map[string]any + _ = json.NewDecoder(r.Body).Decode(&body) + receivedOpenAIBody = body + + isStream, _ := body["stream"].(bool) + w.Header().Set("Content-Type", "application/json") + if isStream { + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + flusher, _ := w.(http.Flusher) + _, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"Response \"}}]}\n\n")) + flusher.Flush() + _, _ = w.Write([]byte("data: {\"choices\":[{\"delta\":{\"content\":\"Chunk\"},\"finish_reason\":\"stop\"}]}\n\n")) + flusher.Flush() + _, _ = w.Write([]byte("data: [DONE]\n\n")) + flusher.Flush() + return + } + + w.WriteHeader(http.StatusOK) + resp := map[string]any{ + "id": "chatcmpl-resp-test", + "model": "gpt-5-model", + "choices": []any{ + map[string]any{ + "index": 0, + "message": map[string]any{ + "role": "assistant", + "content": "Hello from Responses proxy", + }, + "finish_reason": "stop", + }, + }, + "usage": map[string]any{ + "prompt_tokens": 12, + "completion_tokens": 18, + "total_tokens": 30, + }, + } + _ = json.NewEncoder(w).Encode(resp) + })) + defer server.Close() + + configs := []ModelConfig{ + {Model: "gpt-5-model", Key: "test-key", Endpoint: server.URL}, + } + cm := &ConfigManager{ + configs: configs, + uniqueModels: []string{"gpt-5-model"}, + } + handler := handleResponses(cm, nil, 0) + + // 1. Non-streaming test + t.Run("Non-streaming OpenAI response", func(t *testing.T) { + reqObj := map[string]any{ + "model": "gpt-5-model", + "instructions": "Be accurate", + "input": "What is 2+2?", + } + reqBytes, _ := json.Marshal(reqObj) + req := httptest.NewRequest("POST", "/v1/responses", bytes.NewReader(reqBytes)) + req.Header.Set("Content-Type", "application/json") + w := httptest.NewRecorder() + + handler.ServeHTTP(w, req) + + if w.Result().StatusCode != http.StatusOK { + t.Fatalf("Expected 200 OK, got %d", w.Result().StatusCode) + } + + var respObj map[string]any + _ = json.NewDecoder(w.Body).Decode(&respObj) + + if respObj["object"] != "response" { + t.Errorf("Expected object 'response', got %v", respObj["object"]) + } + if respObj["status"] != "completed" { + t.Errorf("Expected status 'completed', got %v", respObj["status"]) + } + output, ok := respObj["output"].([]any) + if !ok || len(output) == 0 { + t.Fatalf("Expected output array in Responses API response, got %+v", respObj) + } + firstOutput := output[0].(map[string]any) + contentSlice, ok := firstOutput["content"].([]any) + if !ok || len(contentSlice) == 0 { + t.Fatalf("Expected output message content slice, got %+v", firstOutput) + } + textVal := contentSlice[0].(map[string]any)["text"].(string) + if textVal != "Hello from Responses proxy" { + t.Errorf("Unexpected text in response output: %s", textVal) + } + + // Verify instructions and input were parsed into OpenAI messages format + openAIMsgs, ok := receivedOpenAIBody["messages"].([]any) + if !ok || len(openAIMsgs) != 2 { + t.Fatalf("Expected 2 messages in OpenAI body, got %+v", receivedOpenAIBody) + } + if openAIMsgs[0].(map[string]any)["role"] != "system" { + t.Errorf("Expected system role for instructions") + } + }) + + // 2. Streaming test + t.Run("Streaming OpenAI response", func(t *testing.T) { + reqObj := map[string]any{ + "model": "gpt-5-model", + "stream": true, + "input": "Stream test", + } + reqBytes, _ := json.Marshal(reqObj) + req := httptest.NewRequest("POST", "/v1/responses", bytes.NewReader(reqBytes)) + req.Header.Set("Content-Type", "application/json") + w := httptest.NewRecorder() + + handler.ServeHTTP(w, req) + + if w.Result().StatusCode != http.StatusOK { + t.Fatalf("Expected 200 OK, got %d", w.Result().StatusCode) + } + + bodyStr := w.Body.String() + if !strings.Contains(bodyStr, "event: response.created") || !strings.Contains(bodyStr, "event: response.output_text.delta") || !strings.Contains(bodyStr, "event: response.completed") { + t.Errorf("Expected OpenAI Responses SSE events in response stream, got: %s", bodyStr) + } + }) +} + + diff --git a/handler.go b/handler.go index 58921dc..555197a 100644 --- a/handler.go +++ b/handler.go @@ -3,6 +3,7 @@ package main import ( + "bufio" "bytes" "encoding/json" "fmt" @@ -853,4 +854,1066 @@ func fibonacci(n int) int64 { return b } +func handleMessages(cm *ConfigManager, expectedTokens []string, retryBaseDelay ...time.Duration) http.HandlerFunc { + baseDelay := 100 * time.Millisecond + if len(retryBaseDelay) > 0 { + baseDelay = retryBaseDelay[0] + } + + return func(w http.ResponseWriter, r *http.Request) { + if !checkAuth(expectedTokens, r) { + sendUnauthorized(w) + return + } + + if r.Method != http.MethodPost { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusMethodNotAllowed) + _ = json.NewEncoder(w).Encode(map[string]any{ + "error": map[string]any{ + "message": "Method not allowed", + "type": "invalid_request_error", + }, + }) + return + } + + bodyBytes, err := io.ReadAll(r.Body) + if err != nil { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + _ = json.NewEncoder(w).Encode(map[string]any{ + "error": map[string]any{ + "message": "Failed to read request body", + "type": "invalid_request_error", + }, + }) + return + } + + var bodyMap map[string]any + if err := json.Unmarshal(bodyBytes, &bodyMap); err != nil { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + _ = json.NewEncoder(w).Encode(map[string]any{ + "error": map[string]any{ + "message": "Invalid JSON in request body", + "type": "invalid_request_error", + }, + }) + return + } + + openAIBodyMap := convertAnthropicToOpenAI(bodyMap) + + var requestedModel string + if m, ok := openAIBodyMap["model"]; ok { + if s, ok := m.(string); ok { + requestedModel = s + } + } + + configs := cm.GetConfigs() + uniqueModels := cm.GetUniqueModels() + + if len(configs) == 0 { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusServiceUnavailable) + _ = json.NewEncoder(w).Encode(map[string]any{ + "error": map[string]any{ + "message": "No model configurations loaded", + "type": "gateway_error", + }, + }) + return + } + + trialConfigs := getTrialConfigs(configs, uniqueModels, requestedModel) + if len(trialConfigs) == 0 { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusServiceUnavailable) + _ = json.NewEncoder(w).Encode(map[string]any{ + "error": map[string]any{ + "message": "No valid trial configuration candidates", + "type": "gateway_error", + }, + }) + return + } + + var isStream bool + if s, ok := openAIBodyMap["stream"]; ok { + if b, ok := s.(bool); ok { + isStream = b + } + } + + log.Printf("Received Anthropic messages request for model %q (stream=%t). Found %d config trials.", requestedModel, isStream, len(trialConfigs)) + + for i, trial := range trialConfigs { + if i > 0 && baseDelay > 0 { + delay := time.Duration(fibonacci(i)) * baseDelay + log.Printf("Trial %d/%d: Fibonacci backoff delay of %v before retry...", i+1, len(trialConfigs), delay) + select { + case <-r.Context().Done(): + log.Printf("Request context cancelled during retry delay before trial %d", i+1) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(499) + _ = json.NewEncoder(w).Encode(map[string]any{ + "error": map[string]any{ + "message": "Client closed request", + "type": "gateway_error", + }, + }) + return + case <-time.After(delay): + } + } + + log.Printf("Trial %d/%d: model=%s endpoint=%s key_len=%d", i+1, len(trialConfigs), trial.Model, trial.Endpoint, len(trial.Key)) + + openAIBodyMap["model"] = trial.Model + modifiedBody, err := json.Marshal(openAIBodyMap) + if err != nil { + log.Printf("Trial %d: Failed to marshal body for %s: %v", i+1, trial.Model, err) + continue + } + + targetURL := buildURL(trial.Endpoint, "/v1/chat/completions") + + outReq, err := http.NewRequestWithContext(r.Context(), "POST", targetURL, bytes.NewReader(modifiedBody)) + if err != nil { + log.Printf("Trial %d: Failed to create outgoing request to %s: %v", i+1, targetURL, err) + continue + } + + for k, vv := range r.Header { + kLower := strings.ToLower(k) + if kLower == "authorization" || kLower == "host" || kLower == "content-length" { + continue + } + for _, v := range vv { + outReq.Header.Add(k, v) + } + } + if trial.Key == "-blank-" { + outReq.Header.Set("Authorization", "Bearer") + } else if trial.Key != "" && trial.Key != "-" { + outReq.Header.Set("Authorization", "Bearer "+trial.Key) + } + outReq.Header.Set("Content-Type", "application/json") + + if trial.Extra != "" { + var extraHeaders map[string]any + if err := json.Unmarshal([]byte(trial.Extra), &extraHeaders); err != nil { + log.Printf("Trial %d: Failed to parse extra headers JSON: %v", i+1, err) + } else { + for hk, hv := range extraHeaders { + var valStr string + switch v := hv.(type) { + case string: + valStr = v + default: + valStr = fmt.Sprintf("%v", v) + } + outReq.Header.Set(hk, valStr) + } + } + } + + resp, err := httpClient.Do(outReq) + if err != nil { + log.Printf("Trial %d: Request to %s failed: %v", i+1, targetURL, err) + continue + } + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + errBody, _ := io.ReadAll(io.LimitReader(resp.Body, 1024)) + resp.Body.Close() + log.Printf("Trial %d: Request to %s returned error status %d: %s", i+1, targetURL, resp.StatusCode, strings.TrimSpace(string(errBody))) + continue + } + + log.Printf("Trial %d: Connection established with status %d. Proxying response.", i+1, resp.StatusCode) + + if !isStream { + respBody, readErr := io.ReadAll(resp.Body) + resp.Body.Close() + if readErr != nil { + log.Printf("Trial %d: Failed to read non-streaming response body: %v", i+1, readErr) + continue + } + + var openAIResp map[string]any + if err := json.Unmarshal(respBody, &openAIResp); err != nil { + log.Printf("Trial %d: Failed to unmarshal upstream response JSON: %v", i+1, err) + continue + } + + anthropicResp := convertOpenAIToAnthropicResponse(openAIResp, requestedModel) + + for k, vv := range resp.Header { + kLower := strings.ToLower(k) + if kLower == "content-length" || kLower == "content-type" { + continue + } + for _, v := range vv { + w.Header().Add(k, v) + } + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(resp.StatusCode) + _ = json.NewEncoder(w).Encode(anthropicResp) + return + } + + flusher, ok := w.(http.Flusher) + if !ok { + log.Printf("Trial %d: Flusher not supported on current ResponseWriter", i+1) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusInternalServerError) + _ = json.NewEncoder(w).Encode(map[string]any{ + "error": map[string]any{ + "message": "Response flusher not supported", + "type": "gateway_error", + }, + }) + return + } + + proxyAnthropicStream(w, resp.Body, requestedModel, flusher) + return + } + + log.Printf("All trials failed. Returning Bad Gateway.") + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadGateway) + _ = json.NewEncoder(w).Encode(map[string]any{ + "error": map[string]any{ + "message": "All configured models and keys failed to respond.", + "type": "gateway_error", + "param": nil, + "code": "all_endpoints_failed", + }, + }) + } +} + +func handleResponses(cm *ConfigManager, expectedTokens []string, retryBaseDelay ...time.Duration) http.HandlerFunc { + baseDelay := 100 * time.Millisecond + if len(retryBaseDelay) > 0 { + baseDelay = retryBaseDelay[0] + } + + return func(w http.ResponseWriter, r *http.Request) { + if !checkAuth(expectedTokens, r) { + sendUnauthorized(w) + return + } + + if r.Method != http.MethodPost { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusMethodNotAllowed) + _ = json.NewEncoder(w).Encode(map[string]any{ + "error": map[string]any{ + "message": "Method not allowed", + "type": "invalid_request_error", + }, + }) + return + } + + bodyBytes, err := io.ReadAll(r.Body) + if err != nil { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + _ = json.NewEncoder(w).Encode(map[string]any{ + "error": map[string]any{ + "message": "Failed to read request body", + "type": "invalid_request_error", + }, + }) + return + } + + var bodyMap map[string]any + if err := json.Unmarshal(bodyBytes, &bodyMap); err != nil { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadRequest) + _ = json.NewEncoder(w).Encode(map[string]any{ + "error": map[string]any{ + "message": "Invalid JSON in request body", + "type": "invalid_request_error", + }, + }) + return + } + + openAIBodyMap := convertResponsesToOpenAI(bodyMap) + + var requestedModel string + if m, ok := openAIBodyMap["model"]; ok { + if s, ok := m.(string); ok { + requestedModel = s + } + } + + configs := cm.GetConfigs() + uniqueModels := cm.GetUniqueModels() + + if len(configs) == 0 { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusServiceUnavailable) + _ = json.NewEncoder(w).Encode(map[string]any{ + "error": map[string]any{ + "message": "No model configurations loaded", + "type": "gateway_error", + }, + }) + return + } + + trialConfigs := getTrialConfigs(configs, uniqueModels, requestedModel) + if len(trialConfigs) == 0 { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusServiceUnavailable) + _ = json.NewEncoder(w).Encode(map[string]any{ + "error": map[string]any{ + "message": "No valid trial configuration candidates", + "type": "gateway_error", + }, + }) + return + } + + var isStream bool + if s, ok := openAIBodyMap["stream"]; ok { + if b, ok := s.(bool); ok { + isStream = b + } + } + + log.Printf("Received OpenAI Responses request for model %q (stream=%t). Found %d config trials.", requestedModel, isStream, len(trialConfigs)) + + for i, trial := range trialConfigs { + if i > 0 && baseDelay > 0 { + delay := time.Duration(fibonacci(i)) * baseDelay + log.Printf("Trial %d/%d: Fibonacci backoff delay of %v before retry...", i+1, len(trialConfigs), delay) + select { + case <-r.Context().Done(): + log.Printf("Request context cancelled during retry delay before trial %d", i+1) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(499) + _ = json.NewEncoder(w).Encode(map[string]any{ + "error": map[string]any{ + "message": "Client closed request", + "type": "gateway_error", + }, + }) + return + case <-time.After(delay): + } + } + + log.Printf("Trial %d/%d: model=%s endpoint=%s key_len=%d", i+1, len(trialConfigs), trial.Model, trial.Endpoint, len(trial.Key)) + + openAIBodyMap["model"] = trial.Model + modifiedBody, err := json.Marshal(openAIBodyMap) + if err != nil { + log.Printf("Trial %d: Failed to marshal body for %s: %v", i+1, trial.Model, err) + continue + } + + targetURL := buildURL(trial.Endpoint, "/v1/chat/completions") + + outReq, err := http.NewRequestWithContext(r.Context(), "POST", targetURL, bytes.NewReader(modifiedBody)) + if err != nil { + log.Printf("Trial %d: Failed to create outgoing request to %s: %v", i+1, targetURL, err) + continue + } + + for k, vv := range r.Header { + kLower := strings.ToLower(k) + if kLower == "authorization" || kLower == "host" || kLower == "content-length" { + continue + } + for _, v := range vv { + outReq.Header.Add(k, v) + } + } + if trial.Key == "-blank-" { + outReq.Header.Set("Authorization", "Bearer") + } else if trial.Key != "" && trial.Key != "-" { + outReq.Header.Set("Authorization", "Bearer "+trial.Key) + } + outReq.Header.Set("Content-Type", "application/json") + + if trial.Extra != "" { + var extraHeaders map[string]any + if err := json.Unmarshal([]byte(trial.Extra), &extraHeaders); err != nil { + log.Printf("Trial %d: Failed to parse extra headers JSON: %v", i+1, err) + } else { + for hk, hv := range extraHeaders { + var valStr string + switch v := hv.(type) { + case string: + valStr = v + default: + valStr = fmt.Sprintf("%v", v) + } + outReq.Header.Set(hk, valStr) + } + } + } + + resp, err := httpClient.Do(outReq) + if err != nil { + log.Printf("Trial %d: Request to %s failed: %v", i+1, targetURL, err) + continue + } + + if resp.StatusCode < 200 || resp.StatusCode >= 300 { + errBody, _ := io.ReadAll(io.LimitReader(resp.Body, 1024)) + resp.Body.Close() + log.Printf("Trial %d: Request to %s returned error status %d: %s", i+1, targetURL, resp.StatusCode, strings.TrimSpace(string(errBody))) + continue + } + + log.Printf("Trial %d: Connection established with status %d. Proxying response.", i+1, resp.StatusCode) + + if !isStream { + respBody, readErr := io.ReadAll(resp.Body) + resp.Body.Close() + if readErr != nil { + log.Printf("Trial %d: Failed to read non-streaming response body: %v", i+1, readErr) + continue + } + + var openAIResp map[string]any + if err := json.Unmarshal(respBody, &openAIResp); err != nil { + log.Printf("Trial %d: Failed to unmarshal upstream response JSON: %v", i+1, err) + continue + } + + responsesResp := convertOpenAIToResponsesResponse(openAIResp, requestedModel) + + for k, vv := range resp.Header { + kLower := strings.ToLower(k) + if kLower == "content-length" || kLower == "content-type" { + continue + } + for _, v := range vv { + w.Header().Add(k, v) + } + } + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(resp.StatusCode) + _ = json.NewEncoder(w).Encode(responsesResp) + return + } + + flusher, ok := w.(http.Flusher) + if !ok { + log.Printf("Trial %d: Flusher not supported on current ResponseWriter", i+1) + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusInternalServerError) + _ = json.NewEncoder(w).Encode(map[string]any{ + "error": map[string]any{ + "message": "Response flusher not supported", + "type": "gateway_error", + }, + }) + return + } + + proxyResponsesStream(w, resp.Body, requestedModel, flusher) + return + } + + log.Printf("All trials failed. Returning Bad Gateway.") + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(http.StatusBadGateway) + _ = json.NewEncoder(w).Encode(map[string]any{ + "error": map[string]any{ + "message": "All configured models and keys failed to respond.", + "type": "gateway_error", + "param": nil, + "code": "all_endpoints_failed", + }, + }) + } +} + +func convertAnthropicToOpenAI(bodyMap map[string]any) map[string]any { + openAIBody := make(map[string]any) + + if m, ok := bodyMap["model"]; ok { + openAIBody["model"] = m + } + if s, ok := bodyMap["stream"].(bool); ok { + openAIBody["stream"] = s + } + if t, ok := bodyMap["temperature"]; ok { + openAIBody["temperature"] = t + } + if p, ok := bodyMap["top_p"]; ok { + openAIBody["top_p"] = p + } + if mt, ok := bodyMap["max_tokens"]; ok { + openAIBody["max_tokens"] = mt + } + if stops, ok := bodyMap["stop_sequences"]; ok { + openAIBody["stop"] = stops + } + + var openAIMessages []any + + if sysVal, ok := bodyMap["system"]; ok && sysVal != nil { + sysContent := normalizeContent(sysVal) + if sysContent != nil { + openAIMessages = append(openAIMessages, map[string]any{ + "role": "system", + "content": sysContent, + }) + } + } + + if msgs, ok := bodyMap["messages"].([]any); ok { + for _, item := range msgs { + if msgMap, ok := item.(map[string]any); ok { + role, _ := msgMap["role"].(string) + if role == "" { + role = "user" + } + cnt := msgMap["content"] + normCnt := convertAnthropicContentToOpenAI(cnt) + + openAIMessages = append(openAIMessages, map[string]any{ + "role": role, + "content": normCnt, + }) + } + } + } + + openAIBody["messages"] = openAIMessages + autodetectAndNormalizeMessages(openAIBody) + return openAIBody +} + +func convertAnthropicContentToOpenAI(cntAny any) any { + if slice, ok := cntAny.([]any); ok { + var newSlice []any + for _, item := range slice { + if m, ok := item.(map[string]any); ok { + if t, ok := m["type"].(string); ok && t == "image" { + if src, ok := m["source"].(map[string]any); ok { + mediaType, _ := src["media_type"].(string) + if mediaType == "" { + mediaType = "image/png" + } + b64Data, _ := src["data"].(string) + dataURL := fmt.Sprintf("data:%s;base64,%s", mediaType, b64Data) + newSlice = append(newSlice, map[string]any{ + "type": "image_url", + "image_url": map[string]any{ + "url": dataURL, + }, + }) + continue + } + } + } + newSlice = append(newSlice, item) + } + return normalizeContent(newSlice) + } + return normalizeContent(cntAny) +} + +func convertOpenAIToAnthropicResponse(openAIResp map[string]any, requestedModel string) map[string]any { + id, _ := openAIResp["id"].(string) + if id == "" { + id = fmt.Sprintf("msg_%d", time.Now().UnixNano()) + } else if !strings.HasPrefix(id, "msg_") { + id = "msg_" + id + } + + model, _ := openAIResp["model"].(string) + if model == "" { + model = requestedModel + } + + var textContent string + var finishReason string + + if choices, ok := openAIResp["choices"].([]any); ok && len(choices) > 0 { + if choice, ok := choices[0].(map[string]any); ok { + if fr, ok := choice["finish_reason"].(string); ok { + finishReason = fr + } + if msg, ok := choice["message"].(map[string]any); ok { + if cnt, ok := msg["content"].(string); ok { + textContent = cnt + } else if cntNorm := normalizeContent(msg["content"]); cntNorm != nil { + if s, ok := cntNorm.(string); ok { + textContent = s + } + } + } + } + } + + stopReason := "end_turn" + switch finishReason { + case "length": + stopReason = "max_tokens" + case "tool_calls", "function_call": + stopReason = "tool_use" + case "stop": + stopReason = "end_turn" + } + + inputTokens := 0 + outputTokens := 0 + if usage, ok := openAIResp["usage"].(map[string]any); ok { + if pt, ok := usage["prompt_tokens"].(float64); ok { + inputTokens = int(pt) + } + if ct, ok := usage["completion_tokens"].(float64); ok { + outputTokens = int(ct) + } + } + + return map[string]any{ + "id": id, + "type": "message", + "role": "assistant", + "model": model, + "content": []any{ + map[string]any{ + "type": "text", + "text": textContent, + }, + }, + "stop_reason": stopReason, + "stop_sequence": nil, + "usage": map[string]any{ + "input_tokens": inputTokens, + "output_tokens": outputTokens, + }, + } +} + +func proxyAnthropicStream(w http.ResponseWriter, respBody io.ReadCloser, requestedModel string, flusher http.Flusher) { + defer respBody.Close() + + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + w.WriteHeader(http.StatusOK) + flusher.Flush() + + reader := bufio.NewReader(respBody) + + msgID := fmt.Sprintf("msg_%d", time.Now().UnixNano()) + + msgStartObj := map[string]any{ + "type": "message_start", + "message": map[string]any{ + "id": msgID, + "type": "message", + "role": "assistant", + "model": requestedModel, + "content": []any{}, + "stop_reason": nil, + "stop_sequence": nil, + "usage": map[string]any{ + "input_tokens": 0, + "output_tokens": 0, + }, + }, + } + msgStartBytes, _ := json.Marshal(msgStartObj) + _, _ = fmt.Fprintf(w, "event: message_start\ndata: %s\n\n", msgStartBytes) + + blockStartObj := map[string]any{ + "type": "content_block_start", + "index": 0, + "content_block": map[string]any{ + "type": "text", + "text": "", + }, + } + blockStartBytes, _ := json.Marshal(blockStartObj) + _, _ = fmt.Fprintf(w, "event: content_block_start\ndata: %s\n\n", blockStartBytes) + flusher.Flush() + + var finalStopReason = "end_turn" + + for { + lineBytes, err := reader.ReadBytes('\n') + if len(lineBytes) > 0 { + line := strings.TrimSpace(string(lineBytes)) + if strings.HasPrefix(line, "data: ") { + dataStr := strings.TrimPrefix(line, "data: ") + dataStr = strings.TrimSpace(dataStr) + if dataStr == "[DONE]" { + break + } + var chunkMap map[string]any + if json.Unmarshal([]byte(dataStr), &chunkMap) == nil { + if choices, ok := chunkMap["choices"].([]any); ok && len(choices) > 0 { + if choice, ok := choices[0].(map[string]any); ok { + if fr, ok := choice["finish_reason"].(string); ok && fr != "" { + switch fr { + case "length": + finalStopReason = "max_tokens" + case "tool_calls", "function_call": + finalStopReason = "tool_use" + case "stop": + finalStopReason = "end_turn" + } + } + if delta, ok := choice["delta"].(map[string]any); ok { + if contentStr, ok := delta["content"].(string); ok && contentStr != "" { + deltaObj := map[string]any{ + "type": "content_block_delta", + "index": 0, + "delta": map[string]any{ + "type": "text_delta", + "text": contentStr, + }, + } + deltaBytes, _ := json.Marshal(deltaObj) + _, _ = fmt.Fprintf(w, "event: content_block_delta\ndata: %s\n\n", deltaBytes) + flusher.Flush() + } + } + } + } + } + } + } + if err != nil { + break + } + } + + blockStopObj := map[string]any{ + "type": "content_block_stop", + "index": 0, + } + blockStopBytes, _ := json.Marshal(blockStopObj) + _, _ = fmt.Fprintf(w, "event: content_block_stop\ndata: %s\n\n", blockStopBytes) + + msgDeltaObj := map[string]any{ + "type": "message_delta", + "delta": map[string]any{ + "stop_reason": finalStopReason, + "stop_sequence": nil, + }, + "usage": map[string]any{ + "output_tokens": 0, + }, + } + msgDeltaBytes, _ := json.Marshal(msgDeltaObj) + _, _ = fmt.Fprintf(w, "event: message_delta\ndata: %s\n\n", msgDeltaBytes) + + msgStopObj := map[string]any{ + "type": "message_stop", + } + msgStopBytes, _ := json.Marshal(msgStopObj) + _, _ = fmt.Fprintf(w, "event: message_stop\ndata: %s\n\n", msgStopBytes) + flusher.Flush() +} + +func convertResponsesToOpenAI(bodyMap map[string]any) map[string]any { + openAIBody := make(map[string]any) + + if m, ok := bodyMap["model"]; ok { + openAIBody["model"] = m + } + if s, ok := bodyMap["stream"].(bool); ok { + openAIBody["stream"] = s + } + if t, ok := bodyMap["temperature"]; ok { + openAIBody["temperature"] = t + } + if p, ok := bodyMap["top_p"]; ok { + openAIBody["top_p"] = p + } + if mt, ok := bodyMap["max_output_tokens"]; ok { + openAIBody["max_tokens"] = mt + } else if mt, ok := bodyMap["max_tokens"]; ok { + openAIBody["max_tokens"] = mt + } + + var openAIMessages []any + + if instVal, ok := bodyMap["instructions"]; ok && instVal != nil { + sysContent := normalizeContent(instVal) + if sysContent != nil { + openAIMessages = append(openAIMessages, map[string]any{ + "role": "system", + "content": sysContent, + }) + } + } + + if inputVal, ok := bodyMap["input"]; ok && inputVal != nil { + switch inp := inputVal.(type) { + case string: + openAIMessages = append(openAIMessages, map[string]any{ + "role": "user", + "content": inp, + }) + case []any: + for _, elem := range inp { + switch item := elem.(type) { + case string: + openAIMessages = append(openAIMessages, map[string]any{ + "role": "user", + "content": item, + }) + case map[string]any: + role, _ := item["role"].(string) + if role == "" { + role = "user" + } + cnt := item["content"] + if cnt == nil { + if txt, ok := item["text"].(string); ok { + cnt = txt + } + } + openAIMessages = append(openAIMessages, map[string]any{ + "role": role, + "content": normalizeContent(cnt), + }) + } + } + case map[string]any: + role, _ := inp["role"].(string) + if role == "" { + role = "user" + } + cnt := inp["content"] + if cnt == nil { + if txt, ok := inp["text"].(string); ok { + cnt = txt + } + } + openAIMessages = append(openAIMessages, map[string]any{ + "role": role, + "content": normalizeContent(cnt), + }) + } + } else if msgs, ok := bodyMap["messages"].([]any); ok { + for _, item := range msgs { + if msgMap, ok := item.(map[string]any); ok { + role, _ := msgMap["role"].(string) + if role == "" { + role = "user" + } + openAIMessages = append(openAIMessages, map[string]any{ + "role": role, + "content": normalizeContent(msgMap["content"]), + }) + } + } + } + + openAIBody["messages"] = openAIMessages + autodetectAndNormalizeMessages(openAIBody) + return openAIBody +} + +func convertOpenAIToResponsesResponse(openAIResp map[string]any, requestedModel string) map[string]any { + id, _ := openAIResp["id"].(string) + if id == "" { + id = fmt.Sprintf("resp_%d", time.Now().UnixNano()) + } else if !strings.HasPrefix(id, "resp_") { + id = "resp_" + id + } + + model, _ := openAIResp["model"].(string) + if model == "" { + model = requestedModel + } + + var textContent string + if choices, ok := openAIResp["choices"].([]any); ok && len(choices) > 0 { + if choice, ok := choices[0].(map[string]any); ok { + if msg, ok := choice["message"].(map[string]any); ok { + if cnt, ok := msg["content"].(string); ok { + textContent = cnt + } else if cntNorm := normalizeContent(msg["content"]); cntNorm != nil { + if s, ok := cntNorm.(string); ok { + textContent = s + } + } + } + } + } + + created := time.Now().Unix() + if c, ok := openAIResp["created"].(float64); ok { + created = int64(c) + } + + promptTokens := 0 + completionTokens := 0 + totalTokens := 0 + if usage, ok := openAIResp["usage"].(map[string]any); ok { + if pt, ok := usage["prompt_tokens"].(float64); ok { + promptTokens = int(pt) + } + if ct, ok := usage["completion_tokens"].(float64); ok { + completionTokens = int(ct) + } + if tt, ok := usage["total_tokens"].(float64); ok { + totalTokens = int(tt) + } + } + + msgID := fmt.Sprintf("msg_%d", time.Now().UnixNano()) + + return map[string]any{ + "id": id, + "object": "response", + "created_at": created, + "status": "completed", + "model": model, + "output": []any{ + map[string]any{ + "id": msgID, + "type": "message", + "status": "completed", + "role": "assistant", + "content": []any{ + map[string]any{ + "type": "output_text", + "text": textContent, + "annotations": []any{}, + "logprobs": []any{}, + }, + }, + }, + }, + "usage": map[string]any{ + "prompt_tokens": promptTokens, + "completion_tokens": completionTokens, + "total_tokens": totalTokens, + }, + } +} + +func proxyResponsesStream(w http.ResponseWriter, respBody io.ReadCloser, requestedModel string, flusher http.Flusher) { + defer respBody.Close() + + w.Header().Set("Content-Type", "text/event-stream") + w.Header().Set("Cache-Control", "no-cache") + w.Header().Set("Connection", "keep-alive") + w.WriteHeader(http.StatusOK) + flusher.Flush() + + reader := bufio.NewReader(respBody) + + respID := fmt.Sprintf("resp_%d", time.Now().UnixNano()) + msgID := fmt.Sprintf("msg_%d", time.Now().UnixNano()) + + createdObj := map[string]any{ + "type": "response.created", + "response": map[string]any{ + "id": respID, + "object": "response", + "status": "in_progress", + "model": requestedModel, + }, + } + createdBytes, _ := json.Marshal(createdObj) + _, _ = fmt.Fprintf(w, "event: response.created\ndata: %s\n\n", createdBytes) + + partAddedObj := map[string]any{ + "type": "response.content_part.added", + "part": map[string]any{ + "type": "output_text", + "text": "", + }, + } + partAddedBytes, _ := json.Marshal(partAddedObj) + _, _ = fmt.Fprintf(w, "event: response.content_part.added\ndata: %s\n\n", partAddedBytes) + flusher.Flush() + + var fullTextBuf strings.Builder + + for { + lineBytes, err := reader.ReadBytes('\n') + if len(lineBytes) > 0 { + line := strings.TrimSpace(string(lineBytes)) + if strings.HasPrefix(line, "data: ") { + dataStr := strings.TrimPrefix(line, "data: ") + dataStr = strings.TrimSpace(dataStr) + if dataStr == "[DONE]" { + break + } + var chunkMap map[string]any + if json.Unmarshal([]byte(dataStr), &chunkMap) == nil { + if choices, ok := chunkMap["choices"].([]any); ok && len(choices) > 0 { + if choice, ok := choices[0].(map[string]any); ok { + if delta, ok := choice["delta"].(map[string]any); ok { + if contentStr, ok := delta["content"].(string); ok && contentStr != "" { + fullTextBuf.WriteString(contentStr) + deltaObj := map[string]any{ + "type": "response.output_text.delta", + "delta": contentStr, + } + deltaBytes, _ := json.Marshal(deltaObj) + _, _ = fmt.Fprintf(w, "event: response.output_text.delta\ndata: %s\n\n", deltaBytes) + flusher.Flush() + } + } + } + } + } + } + } + if err != nil { + break + } + } + + completedObj := map[string]any{ + "type": "response.completed", + "response": map[string]any{ + "id": respID, + "object": "response", + "status": "completed", + "model": requestedModel, + "created_at": time.Now().Unix(), + "output": []any{ + map[string]any{ + "id": msgID, + "type": "message", + "status": "completed", + "role": "assistant", + "content": []any{ + map[string]any{ + "type": "output_text", + "text": fullTextBuf.String(), + "annotations": []any{}, + "logprobs": []any{}, + }, + }, + }, + }, + }, + } + completedBytes, _ := json.Marshal(completedObj) + _, _ = fmt.Fprintf(w, "event: response.completed\ndata: %s\n\n", completedBytes) + flusher.Flush() +} + + diff --git a/main.go b/main.go index 8bc0ebd..5f36ccd 100644 --- a/main.go +++ b/main.go @@ -77,8 +77,15 @@ func main() { // Register routes mux := http.NewServeMux() mux.HandleFunc("/v1/models", handleModels(cm, authTokens)) + mux.HandleFunc("/models", handleModels(cm, authTokens)) mux.HandleFunc("/v1/chat/completions", handleChatCompletions(cm, authTokens, retryDelay)) + mux.HandleFunc("/chat/completions", handleChatCompletions(cm, authTokens, retryDelay)) mux.HandleFunc("/v1/images/generations", handleImageGenerations(cm, authTokens, retryDelay)) + mux.HandleFunc("/images/generations", handleImageGenerations(cm, authTokens, retryDelay)) + mux.HandleFunc("/v1/messages", handleMessages(cm, authTokens, retryDelay)) + mux.HandleFunc("/messages", handleMessages(cm, authTokens, retryDelay)) + mux.HandleFunc("/v1/responses", handleResponses(cm, authTokens, retryDelay)) + mux.HandleFunc("/responses", handleResponses(cm, authTokens, retryDelay)) server := &http.Server{ Addr: fmt.Sprintf(":%d", *port), diff --git a/models.csv b/models.csv index 0e4de1b..2e2a8b8 100644 --- a/models.csv +++ b/models.csv @@ -7,6 +7,7 @@ inclusionai/ling-3.0-flash:free,-,https://api.kilo.ai/api/openrouter, kilo-auto/free,-,https://api.kilo.ai/api/openrouter, laguna-s-2.1-free,-,https://opencode.ai/zen, ling-3.0-flash-free,-,https://opencode.ai/zen, +longcat-2.0-free,-,https://opencode.ai/zen, Meta-Llama-3_3-70B-Instruct,-,https://oai.endpoints.kepler.ai.cloud.ovh.net, mimo-v2.5-free,-,https://opencode.ai/zen, Mistral-7B-Instruct-v0.3,-,https://oai.endpoints.kepler.ai.cloud.ovh.net,