feat: initial extract of shared anthropic-compat client + translator
Extracted from Plexum-minimax-provider and Plexum-kimi-provider (both
had near-identical copies under internal/anthropic + internal/translate).
Future provider plugins (Qwen, Doubao, …) import these two packages
instead of re-implementing the protocol.
anthropic/ (~220 LOC + 6 tests):
- minimal HTTP+SSE Anthropic Messages client (POST /v1/messages,
stream:true, parses event:/data: SSE)
- Client.UserAgent + Client.ExtraHeaders fields for per-backend
customization (Kimi needs "claude-code/0.1.0"; MiniMax is flexible)
- non-2xx surfaces as Go error; mid-stream errors as final
Event{Type:"error"}
translate/ (~220 LOC):
- CanonicalToAnthropic: TurnRequest → MessagesRequest
- blockToAnthropic: TextBlock / ToolUseBlock / ToolResultBlock /
ThinkingBlock → loose ContentBlock map; preserves signatures
- Translator: per-turn state machine; anthropic.Event stream →
canonical.TurnEvent stream (text + thinking + tool_use deltas;
signature_delta capture; message_delta stop_reason + usage)
Both MiniMax and Kimi now import from here; their internal/* dirs are
gone. Live verified after refactor — both still answer "ready" via
plexum say.
This commit is contained in:
143
anthropic/client_test.go
Normal file
143
anthropic/client_test.go
Normal file
@@ -0,0 +1,143 @@
|
||||
package anthropic
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"strings"
|
||||
"testing"
|
||||
)
|
||||
|
||||
func TestParseSSESingleEvent(t *testing.T) {
|
||||
body := "event: message_start\ndata: {\"type\":\"message_start\",\"message\":{\"id\":\"m1\",\"model\":\"X\"}}\n\n"
|
||||
ch := make(chan Event, 4)
|
||||
if err := parseSSE(context.Background(), strings.NewReader(body), ch); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
close(ch)
|
||||
events := drain(ch)
|
||||
if len(events) != 1 || events[0].Type != "message_start" || events[0].Message.ID != "m1" {
|
||||
t.Errorf("events = %+v", events)
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseSSEMultiLineData(t *testing.T) {
|
||||
// "data:" can be split across multiple lines per spec; joined with '\n'.
|
||||
body := "data: {\"type\":\n" + "data: \"text_delta\",\"delta\":{\"type\":\"text_delta\",\"text\":\"hi\"}}\n\n"
|
||||
ch := make(chan Event, 4)
|
||||
if err := parseSSE(context.Background(), strings.NewReader(body), ch); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
close(ch)
|
||||
events := drain(ch)
|
||||
if len(events) != 1 || events[0].Type != "text_delta" || events[0].Delta.Text != "hi" {
|
||||
t.Errorf("events = %+v", events)
|
||||
}
|
||||
}
|
||||
|
||||
func TestParseSSEFlushTrailingWithoutBlank(t *testing.T) {
|
||||
body := "data: {\"type\":\"message_stop\"}"
|
||||
ch := make(chan Event, 4)
|
||||
if err := parseSSE(context.Background(), strings.NewReader(body), ch); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
close(ch)
|
||||
events := drain(ch)
|
||||
if len(events) != 1 || events[0].Type != "message_stop" {
|
||||
t.Errorf("trailing flush missed: %+v", events)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStreamMessagesSendsRightHeaders(t *testing.T) {
|
||||
var gotAuth, gotVer string
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
gotAuth = r.Header.Get("authorization")
|
||||
gotVer = r.Header.Get("anthropic-version")
|
||||
w.Header().Set("content-type", "text/event-stream")
|
||||
w.Write([]byte("event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n"))
|
||||
}))
|
||||
defer srv.Close()
|
||||
c := New(srv.URL, "fake-key")
|
||||
ch, err := c.StreamMessages(context.Background(), MessagesRequest{
|
||||
Model: "X", MaxTokens: 10,
|
||||
Messages: []Message{{Role: "user", Content: []ContentBlock{{"type": "text", "text": "hi"}}}},
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for range ch {
|
||||
}
|
||||
if gotAuth != "Bearer fake-key" {
|
||||
t.Errorf("auth = %q", gotAuth)
|
||||
}
|
||||
if gotVer != DefaultAPIVersion {
|
||||
t.Errorf("anthropic-version = %q", gotVer)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStreamMessagesNon2xxErrors(t *testing.T) {
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.WriteHeader(http.StatusUnauthorized)
|
||||
w.Write([]byte(`{"error":{"type":"authentication_error","message":"bad key"}}`))
|
||||
}))
|
||||
defer srv.Close()
|
||||
c := New(srv.URL, "bad")
|
||||
_, err := c.StreamMessages(context.Background(), MessagesRequest{Model: "X", MaxTokens: 10})
|
||||
if err == nil || !strings.Contains(err.Error(), "401") {
|
||||
t.Errorf("err = %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestStreamMessagesStreamErrorBecomesEvent(t *testing.T) {
|
||||
// Simulate an SSE stream that errors mid-stream.
|
||||
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
w.Header().Set("content-type", "text/event-stream")
|
||||
flusher, _ := w.(http.Flusher)
|
||||
w.Write([]byte("event: message_start\ndata: {\"type\":\"message_start\"}\n\n"))
|
||||
flusher.Flush()
|
||||
// Hijack-style mid-stream close: not really an error in HTTP terms,
|
||||
// scanner just hits EOF cleanly. So no error event expected.
|
||||
}))
|
||||
defer srv.Close()
|
||||
c := New(srv.URL, "k")
|
||||
ch, err := c.StreamMessages(context.Background(), MessagesRequest{Model: "X", MaxTokens: 10})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
events := drain(ch)
|
||||
if len(events) != 1 || events[0].Type != "message_start" {
|
||||
t.Errorf("events = %+v", events)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMessagesRequestMarshalShape(t *testing.T) {
|
||||
req := MessagesRequest{
|
||||
Model: "MiniMax-M2.7", MaxTokens: 50, Stream: true,
|
||||
Messages: []Message{{Role: "user", Content: []ContentBlock{
|
||||
{"type": "text", "text": "hi"},
|
||||
}}},
|
||||
}
|
||||
raw, _ := json.Marshal(req)
|
||||
// stream:true present
|
||||
if !bytes.Contains(raw, []byte(`"stream":true`)) {
|
||||
t.Errorf("missing stream:true in %s", raw)
|
||||
}
|
||||
if !bytes.Contains(raw, []byte(`"model":"MiniMax-M2.7"`)) {
|
||||
t.Errorf("missing model: %s", raw)
|
||||
}
|
||||
}
|
||||
|
||||
func drain(ch <-chan Event) []Event {
|
||||
var out []Event
|
||||
for ev := range ch {
|
||||
out = append(out, ev)
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// Compile-time check that io.EOF is exported so we don't accidentally
|
||||
// remove the import.
|
||||
var _ = io.EOF
|
||||
Reference in New Issue
Block a user