1package provider
2
3import (
4 "bytes"
5 "context"
6 "encoding/base64"
7 "encoding/json"
8 "fmt"
9 "io"
10 "net/http"
11 "os"
12 "strings"
13 "time"
14)
15
16const anthropicDefaultBaseURL = "https://api.anthropic.com"
17const anthropicAPIVersion = "2023-06-01"
18
19// Stealth identity used when talking to Anthropic via subscription OAuth.
20// These values mimic the official Claude Code CLI so Anthropic's edge
21// accepts the request; diverging from them causes 429 rate_limit_error
22// or 403 on the very first request.
23const (
24 claudeCodeVersion = "2.1.75"
25 claudeCodeIdentity = "You are Claude Code, Anthropic's official CLI for Claude."
26)
27
28// Claude Code's canonical tool casing. When running under OAuth we must
29// advertise tool names that match this list (case-insensitive lookup),
30// because Anthropic's backend cross-checks them.
31var claudeCodeToolNames = map[string]string{
32 "read": "Read",
33 "write": "Write",
34 "edit": "Edit",
35 "bash": "Bash",
36 "grep": "Grep",
37 "glob": "Glob",
38}
39
40func toClaudeCodeToolName(name string) string {
41 if cc, ok := claudeCodeToolNames[strings.ToLower(name)]; ok {
42 return cc
43 }
44 return name
45}
46
47func fromClaudeCodeToolName(name string, tools []Tool) string {
48 lower := strings.ToLower(name)
49 for _, t := range tools {
50 if strings.ToLower(t.Name) == lower {
51 return t.Name
52 }
53 }
54 return name
55}
56
57// anthropicClient implements Client against the Anthropic Messages API.
58type anthropicClient struct {
59 apiKey string
60 baseURL string
61 oauthTok string // when non-empty, send Bearer auth instead of x-api-key
62 http *http.Client
63
64 // name overrides the default "anthropic" identity. Anthropic-Messages-
65 // compatible third-party endpoints (kimi-coding, fireworks, minimax,
66 // vercel-ai-gateway, etc.) set this so cost lookup / logging / image-
67 // stripping logic can route on a stable provider id.
68 name string
69
70 // headers carries extra request headers (e.g. Kimi Code's X-Msh-*).
71 headers map[string]string
72}
73
74// NewAnthropic creates an Anthropic client using an API key. baseURL may be empty.
75func NewAnthropic(apiKey, baseURL string) Client {
76 if baseURL == "" {
77 baseURL = anthropicDefaultBaseURL
78 }
79 return &anthropicClient{
80 apiKey: apiKey,
81 baseURL: strings.TrimRight(baseURL, "/"),
82 http: &http.Client{Timeout: 0},
83 }
84}
85
86// NewAnthropicOAuth creates an Anthropic client using a subscription OAuth access token.
87func NewAnthropicOAuth(accessToken, baseURL string) Client {
88 if baseURL == "" {
89 baseURL = anthropicDefaultBaseURL
90 }
91 return &anthropicClient{
92 oauthTok: accessToken,
93 baseURL: strings.TrimRight(baseURL, "/"),
94 http: &http.Client{Timeout: 0},
95 }
96}
97
98func (c *anthropicClient) Name() string {
99 if c.name != "" {
100 return c.name
101 }
102 return "anthropic"
103}
104
105// ---- wire types ----
106
107type anthTextBlock struct {
108 Type string `json:"type"` // "text"
109 Text string `json:"text"`
110 CacheControl *anthCacheCtrl `json:"cache_control,omitempty"`
111}
112
113type anthImageSource struct {
114 Type string `json:"type"` // "base64"
115 MediaType string `json:"media_type"`
116 Data string `json:"data"`
117}
118
119type anthImageBlock struct {
120 Type string `json:"type"` // "image"
121 Source anthImageSource `json:"source"`
122 CacheControl *anthCacheCtrl `json:"cache_control,omitempty"`
123}
124
125type anthToolUseBlock struct {
126 Type string `json:"type"` // "tool_use"
127 ID string `json:"id"`
128 Name string `json:"name"`
129 Input json.RawMessage `json:"input"`
130 CacheControl *anthCacheCtrl `json:"cache_control,omitempty"`
131}
132
133type anthToolResultBlock struct {
134 Type string `json:"type"` // "tool_result"
135 ToolUseID string `json:"tool_use_id"`
136 Content json.RawMessage `json:"content"` // string or array of blocks
137 IsError bool `json:"is_error,omitempty"`
138 CacheControl *anthCacheCtrl `json:"cache_control,omitempty"`
139}
140
141type anthCacheCtrl struct {
142 Type string `json:"type"` // "ephemeral"
143 TTL string `json:"ttl,omitempty"`
144}
145
146type anthMessage struct {
147 Role string `json:"role"`
148 Content []interface{} `json:"content"`
149}
150
151type anthSystemBlock struct {
152 Type string `json:"type"` // "text"
153 Text string `json:"text"`
154 CacheControl *anthCacheCtrl `json:"cache_control,omitempty"`
155}
156
157type anthTool struct {
158 Name string `json:"name"`
159 Description string `json:"description,omitempty"`
160 InputSchema json.RawMessage `json:"input_schema"`
161 CacheControl *anthCacheCtrl `json:"cache_control,omitempty"`
162}
163
164type anthThinking struct {
165 Type string `json:"type"` // "enabled"
166 BudgetTokens int `json:"budget_tokens"`
167}
168
169type anthRequest struct {
170 Model string `json:"model"`
171 MaxTokens int `json:"max_tokens"`
172 System []anthSystemBlock `json:"system,omitempty"`
173 Messages []anthMessage `json:"messages"`
174 Tools []anthTool `json:"tools,omitempty"`
175 Temperature *float32 `json:"temperature,omitempty"`
176 Thinking *anthThinking `json:"thinking,omitempty"`
177 Stream bool `json:"stream"`
178}
179
180// ---- request building ----
181
182func (c *anthropicClient) buildRequest(req Request) (*anthRequest, error) {
183 // Look up under the client's actual provider id, which may be "anthropic"
184 // (default) or a third party that speaks the Anthropic Messages API
185 // (kimi, fireworks, minimax, vercel-ai-gateway, ...). Falling back to a
186 // provider-agnostic lookup keeps things working for catalog-less
187 // configurations (e.g. user passes --model on an obscure third party).
188 m, err := FindModel(c.Name(), req.Model)
189 if err != nil {
190 if m2, err2 := FindModel("", req.Model); err2 == nil {
191 m = m2
192 err = nil
193 }
194 }
195 if err != nil {
196 return nil, err
197 }
198 maxTok := req.MaxTokens
199 if maxTok <= 0 {
200 maxTok = m.MaxOutput
201 }
202
203 out := &anthRequest{
204 Model: req.Model,
205 MaxTokens: maxTok,
206 Temperature: req.Temperature,
207 Stream: true,
208 }
209
210 // System prompt assembly differs between api-key and OAuth modes.
211 // OAuth requests MUST begin with the Claude Code identity line or
212 // Anthropic rejects them (429 rate_limit_error with zero tokens used).
213 //
214 // Cache budget: anthropic caps cache_control to 4 breakpoints per
215 // request. We spend them on:
216 // 1. claude-code identity (OAuth only; stable forever)
217 // 2. user system prompt (changes per-session at most)
218 // 3. last tool definition (tools change rarely)
219 // 4. last message block (advances every turn)
220 //
221 // The identity line gets its OWN cache_control so the prefix
222 // [identity] is cacheable independently of the user system
223 // prompt. Without that, the cache prefix starts after block 2
224 // and any drift in the user prompt (e.g. the Current date
225 // line flipping at midnight) invalidates everything, including
226 // the 17 identity tokens we have to re-send every request
227 // forever.
228 if c.oauthTok != "" {
229 out.System = []anthSystemBlock{{
230 Type: "text",
231 Text: claudeCodeIdentity,
232 CacheControl: &anthCacheCtrl{Type: "ephemeral"},
233 }}
234 if req.System != "" {
235 out.System = append(out.System, anthSystemBlock{
236 Type: "text",
237 Text: req.System,
238 CacheControl: &anthCacheCtrl{Type: "ephemeral"},
239 })
240 }
241 } else if req.System != "" {
242 out.System = []anthSystemBlock{{
243 Type: "text",
244 Text: req.System,
245 CacheControl: &anthCacheCtrl{Type: "ephemeral"},
246 }}
247 }
248
249 if req.Reasoning != "" && m.Reasoning {
250 budget := anthropicReasoningBudget(req.Reasoning)
251 if budget > 0 {
252 // Reasoning requires max_tokens > budget. Keep at least a small
253 // answer budget while respecting the model's advertised output cap.
254 const minAnswerTokens = 1024
255 if m.MaxOutput > minAnswerTokens && budget >= m.MaxOutput {
256 budget = m.MaxOutput - minAnswerTokens
257 }
258 out.Thinking = &anthThinking{Type: "enabled", BudgetTokens: budget}
259 if out.MaxTokens <= budget {
260 out.MaxTokens = budget + minAnswerTokens
261 if m.MaxOutput > 0 && out.MaxTokens > m.MaxOutput {
262 out.MaxTokens = m.MaxOutput
263 }
264 }
265 }
266 }
267
268 for _, t := range req.Tools {
269 name := t.Name
270 if c.oauthTok != "" {
271 name = toClaudeCodeToolName(name)
272 }
273 out.Tools = append(out.Tools, anthTool{
274 Name: name,
275 Description: t.Description,
276 InputSchema: t.Schema,
277 })
278 }
279 // Cache the last tool definition (applies cache breakpoint to the whole tools array).
280 if n := len(out.Tools); n > 0 {
281 out.Tools[n-1].CacheControl = &anthCacheCtrl{Type: "ephemeral"}
282 }
283
284 // Convert messages. Anthropic's wire format has only "user" and
285 // "assistant" roles; tool_result blocks live inside user messages.
286 // Some image-capable providers (for example Gemini image generation)
287 // can emit assistant image blocks. Anthropic only accepts image blocks
288 // in user messages, so strip assistant-side images when switching
289 // providers mid-session. The saved-path TextBlock emitted beside the
290 // image keeps the artifact reachable in chat.
291 //
292 // CRITICAL: tool_result blocks go into their OWN new user
293 // message, they are NOT merged into the preceding user message.
294 // Merging would mutate the prior user message's content array
295 // between turn N and turn N+1: turn N caches the prefix ending at
296 // [user: "read sample.ts"], turn N+1 sends
297 // [user: "read sample.ts" + tool_result=...] which is a
298 // different block sequence, busting the cache prefix match.
299 // Anthropic's API happily accepts consecutive user messages, and
300 // emitting them separately keeps each message bit-stable across
301 // turns, so the cache prefix matches for the entire history up
302 // to the newest block.
303 req.Messages = RepairOrphanedToolResults(req.Messages)
304 for _, msg := range req.Messages {
305 renameTools := c.oauthTok != ""
306 switch msg.Role {
307 case RoleUser:
308 out.Messages = append(out.Messages, anthMessage{
309 Role: "user",
310 Content: convertAnthContent(msg.Content, renameTools),
311 })
312 case RoleTool:
313 out.Messages = append(out.Messages, anthMessage{
314 Role: "user",
315 Content: convertAnthContent(msg.Content, renameTools),
316 })
317 case RoleAssistant:
318 out.Messages = append(out.Messages, anthMessage{
319 Role: "assistant",
320 Content: convertAnthContent(filterAnthAssistantContent(msg.Content), renameTools),
321 })
322 }
323 }
324
325 // Tag the LAST user message with cache_control. Spends the 4th
326 // breakpoint. For prefixes under ~1024 tokens (Anthropic's
327 // minimum cacheable block size for Opus), no cache is written.
328 tagLastUserCache(out.Messages)
329
330 return out, nil
331}
332
333// tagLastUserCache marks the last block of the most recent user
334// message. One marker; combined with identity + systemPrompt +
335// last-tool, spends Anthropic's 4-breakpoint budget.
336func tagLastUserCache(msgs []anthMessage) {
337 for i := len(msgs) - 1; i >= 0; i-- {
338 if msgs[i].Role == "user" {
339 markLastBlockEphemeral(msgs[i].Content)
340 return
341 }
342 }
343}
344
345// markLastBlockEphemeral sets CacheControl on the last entry in blocks
346// regardless of whether it's a text, image, tool_use, or tool_result.
347// Each block type carries its own CacheControl pointer so we type-
348// switch + reassign the slice element.
349func markLastBlockEphemeral(blocks []interface{}) {
350 if len(blocks) == 0 {
351 return
352 }
353 i := len(blocks) - 1
354 cc := &anthCacheCtrl{Type: "ephemeral"}
355 switch v := blocks[i].(type) {
356 case anthTextBlock:
357 v.CacheControl = cc
358 blocks[i] = v
359 case anthImageBlock:
360 v.CacheControl = cc
361 blocks[i] = v
362 case anthToolUseBlock:
363 v.CacheControl = cc
364 blocks[i] = v
365 case anthToolResultBlock:
366 v.CacheControl = cc
367 blocks[i] = v
368 }
369}
370
371func anthropicReasoningBudget(level string) int {
372 return ReasoningBudget(level)
373}
374
375func filterAnthAssistantContent(blocks []Content) []Content {
376 filtered := make([]Content, 0, len(blocks))
377 for _, b := range blocks {
378 if _, ok := b.(ImageBlock); ok {
379 continue
380 }
381 filtered = append(filtered, b)
382 }
383 return filtered
384}
385
386func convertAnthContent(blocks []Content, renameTools bool) []interface{} {
387 out := make([]interface{}, 0, len(blocks))
388 for _, b := range blocks {
389 switch v := b.(type) {
390 case TextBlock:
391 if v.Text == "" {
392 continue
393 }
394 out = append(out, anthTextBlock{Type: "text", Text: v.Text})
395 case ImageBlock:
396 data, mime := anthShrinkImageBytesIfTooBig(v.Data, v.MimeType)
397 out = append(out, anthImageBlock{
398 Type: "image",
399 Source: anthImageSource{
400 Type: "base64",
401 MediaType: mime,
402 Data: base64.StdEncoding.EncodeToString(data),
403 },
404 })
405 case ToolCallBlock:
406 args := v.Arguments
407 if len(args) == 0 || !json.Valid(args) {
408 args = json.RawMessage("{}")
409 }
410 name := v.Name
411 if renameTools {
412 name = toClaudeCodeToolName(name)
413 }
414 out = append(out, anthToolUseBlock{
415 Type: "tool_use", ID: v.ID, Name: name, Input: args,
416 })
417 case ToolResultBlock:
418 // Flatten content to a string if all text; else array of blocks.
419 content, _ := anthBuildToolResultContent(v.Content)
420 out = append(out, anthToolResultBlock{
421 Type: "tool_result",
422 ToolUseID: v.CallID,
423 Content: content,
424 IsError: v.IsError,
425 })
426 }
427 }
428 return out
429}
430
431func anthBuildToolResultContent(blocks []Content) (json.RawMessage, error) {
432 onlyText := true
433 var sb strings.Builder
434 for _, b := range blocks {
435 if tb, ok := b.(TextBlock); ok {
436 if sb.Len() > 0 {
437 sb.WriteString("\n")
438 }
439 sb.WriteString(tb.Text)
440 } else {
441 onlyText = false
442 break
443 }
444 }
445 if onlyText {
446 if sb.Len() == 0 {
447 return json.Marshal("")
448 }
449 return json.Marshal(sb.String())
450 }
451 // Array form: text + image blocks.
452 arr := make([]interface{}, 0, len(blocks))
453 for _, b := range blocks {
454 switch v := b.(type) {
455 case TextBlock:
456 arr = append(arr, anthTextBlock{Type: "text", Text: v.Text})
457 case ImageBlock:
458 data, mime := anthShrinkImageBytesIfTooBig(v.Data, v.MimeType)
459 arr = append(arr, anthImageBlock{
460 Type: "image",
461 Source: anthImageSource{
462 Type: "base64",
463 MediaType: mime,
464 Data: base64.StdEncoding.EncodeToString(data),
465 },
466 })
467 }
468 }
469 return json.Marshal(arr)
470}
471
472// ---- streaming ----
473
474func (c *anthropicClient) Stream(ctx context.Context, req Request) (<-chan Event, error) {
475 wire, err := c.buildRequest(req)
476 if err != nil {
477 return nil, err
478 }
479 body, err := json.Marshal(wire)
480 if err != nil {
481 return nil, err
482 }
483
484 // Optional debug dump: when $ZOT_DEBUG_ANTHROPIC is a file path
485 // we append every outgoing request body to it, one JSON object
486 // per line. Useful for diffing turn N vs turn N+1 to understand
487 // why the cache prefix isn't matching.
488 if dump := os.Getenv("ZOT_DEBUG_ANTHROPIC"); dump != "" {
489 if f, derr := os.OpenFile(dump, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600); derr == nil {
490 _, _ = f.Write(body)
491 _, _ = f.Write([]byte{'\n'})
492 _ = f.Close()
493 }
494 }
495
496 newReq := func() (*http.Request, error) {
497 httpReq, err := http.NewRequestWithContext(ctx, "POST", c.baseURL+"/v1/messages", bytes.NewReader(body))
498 if err != nil {
499 return nil, err
500 }
501 httpReq.Header.Set("content-type", "application/json")
502 httpReq.Header.Set("anthropic-version", anthropicAPIVersion)
503 if c.oauthTok != "" {
504 // Claude-Code-shaped request: identical headers and values as the
505 // official CLI. Any drift triggers Anthropic's anti-abuse check and
506 // rate-limits (or outright blocks) the request.
507 httpReq.Header.Set("accept", "application/json")
508 httpReq.Header.Set("authorization", "Bearer "+c.oauthTok)
509 httpReq.Header.Set("anthropic-beta", "claude-code-20250219,oauth-2025-04-20,fine-grained-tool-streaming-2025-05-14")
510 httpReq.Header.Set("anthropic-dangerous-direct-browser-access", "true")
511 httpReq.Header.Set("user-agent", "claude-cli/"+claudeCodeVersion)
512 httpReq.Header.Set("x-app", "cli")
513 // Remove x-api-key entirely by NOT setting it.
514 } else {
515 httpReq.Header.Set("accept", "text/event-stream")
516 httpReq.Header.Set("x-api-key", c.apiKey)
517 }
518 // Extra headers (set by anthropic-messages-compatible third parties
519 // — kimi-coding's X-Msh-*, copilot's Editor-Plugin-Version, etc.).
520 // Applied last so callers can override defaults if they really need to.
521 for k, v := range c.headers {
522 httpReq.Header.Set(k, v)
523 }
524 return httpReq, nil
525 }
526
527 resp, err := doStreamWithRetry(ctx, c.http, newReq)
528 if err != nil {
529 return nil, fmt.Errorf("anthropic: %w", err)
530 }
531 if resp.StatusCode != http.StatusOK {
532 b, _ := io.ReadAll(resp.Body)
533 resp.Body.Close()
534 return nil, fmt.Errorf("anthropic: http %d: %s", resp.StatusCode, strings.TrimSpace(string(b)))
535 }
536
537 out := make(chan Event, 16)
538 go c.runStream(ctx, resp, req, out)
539 return out, nil
540}
541
542func (c *anthropicClient) runStream(ctx context.Context, resp *http.Response, req Request, out chan<- Event) {
543 defer close(out)
544 defer resp.Body.Close()
545
546 // Same lookup-by-actual-provider-id pattern as buildRequest, so cost
547 // calculation works for third-party Anthropic-Messages endpoints.
548 model, _ := FindModel(c.Name(), req.Model)
549 if model.ID == "" {
550 model, _ = FindModel("", req.Model)
551 }
552 out <- EventStart{Model: req.Model, Provider: c.Name()}
553
554 raw := make(chan sseEvent, 16)
555 go readSSE(resp.Body, raw)
556
557 // State for assembling the assistant message. Blocks are indexed
558 // by their `index` field from Anthropic so we can preserve the
559 // interleaved order the model emitted them in (text may come
560 // before OR after tool_use; mixing both happens frequently).
561 type blockEntry struct {
562 kind string // "text" | "tool_use"
563 textBuf strings.Builder
564 toolCall ToolCallBlock
565 toolArgs strings.Builder
566 }
567 var (
568 blocks = map[int]*blockEntry{}
569 blockOrder []int // insertion order of indexes
570 activeIdx = -1
571 usage Usage
572 stop StopReason = StopEnd
573 finalErr error
574 )
575 _ = activeIdx // read-only indicator used for legacy parity
576
577 registerBlock := func(idx int, kind string) *blockEntry {
578 if be, ok := blocks[idx]; ok {
579 return be
580 }
581 be := &blockEntry{kind: kind}
582 blocks[idx] = be
583 blockOrder = append(blockOrder, idx)
584 return be
585 }
586
587 assembleMsg := func() Message {
588 content := []Content{}
589 for _, idx := range blockOrder {
590 be := blocks[idx]
591 switch be.kind {
592 case "text":
593 if be.textBuf.Len() > 0 {
594 content = append(content, TextBlock{Text: be.textBuf.String()})
595 }
596 case "tool_use":
597 tc := be.toolCall
598 args := be.toolArgs.String()
599 if args == "" {
600 args = "{}"
601 }
602 tc.Arguments = json.RawMessage(args)
603 content = append(content, tc)
604 }
605 }
606 return Message{Role: RoleAssistant, Content: content, Time: time.Now()}
607 }
608
609 sendDone := func() {
610 usage.CostUSD = ComputeCost(model, usage)
611 out <- EventUsage{Usage: usage}
612 out <- EventDone{Stop: stop, Err: finalErr, Message: assembleMsg()}
613 }
614
615 for {
616 select {
617 case <-ctx.Done():
618 stop = StopAborted
619 finalErr = ctx.Err()
620 sendDone()
621 return
622 case ev, ok := <-raw:
623 if !ok {
624 sendDone()
625 return
626 }
627 // Parse event payload based on event: type.
628 var payload map[string]json.RawMessage
629 if err := json.Unmarshal([]byte(ev.Data), &payload); err != nil {
630 continue
631 }
632 switch ev.Event {
633 case "content_block_start":
634 var idx int
635 if b, ok := payload["index"]; ok {
636 _ = json.Unmarshal(b, &idx)
637 }
638 var block struct {
639 Type string `json:"type"`
640 ID string `json:"id"`
641 Name string `json:"name"`
642 Text string `json:"text"`
643 Input json.RawMessage `json:"input"`
644 }
645 if b, ok := payload["content_block"]; ok {
646 _ = json.Unmarshal(b, &block)
647 }
648 activeIdx = idx
649 switch block.Type {
650 case "tool_use":
651 name := block.Name
652 if c.oauthTok != "" {
653 name = fromClaudeCodeToolName(name, req.Tools)
654 }
655 be := registerBlock(idx, "tool_use")
656 be.toolCall = ToolCallBlock{ID: block.ID, Name: name}
657 out <- EventToolStart{ID: block.ID, Name: name}
658 case "text":
659 registerBlock(idx, "text")
660 case "thinking":
661 // not surfaced
662 }
663 case "content_block_delta":
664 var idx int
665 if b, ok := payload["index"]; ok {
666 _ = json.Unmarshal(b, &idx)
667 }
668 var d struct {
669 Type string `json:"type"`
670 Text string `json:"text"`
671 PartialJSON string `json:"partial_json"`
672 Thinking string `json:"thinking"`
673 }
674 if b, ok := payload["delta"]; ok {
675 _ = json.Unmarshal(b, &d)
676 }
677 switch d.Type {
678 case "text_delta":
679 if be, ok := blocks[idx]; ok && be.kind == "text" {
680 be.textBuf.WriteString(d.Text)
681 }
682 out <- EventTextDelta{Delta: d.Text}
683 case "input_json_delta":
684 if be, ok := blocks[idx]; ok && be.kind == "tool_use" {
685 be.toolArgs.WriteString(d.PartialJSON)
686 out <- EventToolArgs{ID: be.toolCall.ID, Delta: d.PartialJSON}
687 }
688 case "thinking_delta":
689 // Not surfaced in v1.
690 }
691 case "content_block_stop":
692 var idx int
693 if b, ok := payload["index"]; ok {
694 _ = json.Unmarshal(b, &idx)
695 }
696 if be, ok := blocks[idx]; ok && be.kind == "tool_use" {
697 out <- EventToolEnd{ID: be.toolCall.ID}
698 }
699 activeIdx = -1
700 case "message_start":
701 var m struct {
702 Message struct {
703 Usage struct {
704 InputTokens int `json:"input_tokens"`
705 OutputTokens int `json:"output_tokens"`
706 CacheReadInputTokens int `json:"cache_read_input_tokens"`
707 CacheCreationInputTokens int `json:"cache_creation_input_tokens"`
708 } `json:"usage"`
709 } `json:"message"`
710 }
711 _ = json.Unmarshal([]byte(ev.Data), &m)
712 // Anthropic sends cumulative values on message_start and
713 // again on message_delta (refreshed), so assign, don't
714 // accumulate. Accumulating doubles cache_creation_input
715 // which can be 50-70% of cost.
716 usage.InputTokens = m.Message.Usage.InputTokens
717 usage.OutputTokens = m.Message.Usage.OutputTokens
718 usage.CacheReadTokens = m.Message.Usage.CacheReadInputTokens
719 usage.CacheWriteTokens = m.Message.Usage.CacheCreationInputTokens
720 case "message_delta":
721 var m struct {
722 Delta struct {
723 StopReason string `json:"stop_reason"`
724 } `json:"delta"`
725 Usage struct {
726 InputTokens int `json:"input_tokens"`
727 OutputTokens int `json:"output_tokens"`
728 CacheReadInputTokens int `json:"cache_read_input_tokens"`
729 CacheCreationInputTokens int `json:"cache_creation_input_tokens"`
730 } `json:"usage"`
731 }
732 _ = json.Unmarshal([]byte(ev.Data), &m)
733 // Refresh usage from the latest cumulative totals
734 // Anthropic provides. Only apply non-zero fields in case
735 // a given delta only carries output tokens.
736 if m.Usage.InputTokens > 0 {
737 usage.InputTokens = m.Usage.InputTokens
738 }
739 if m.Usage.OutputTokens > 0 {
740 usage.OutputTokens = m.Usage.OutputTokens
741 }
742 if m.Usage.CacheReadInputTokens > 0 {
743 usage.CacheReadTokens = m.Usage.CacheReadInputTokens
744 }
745 if m.Usage.CacheCreationInputTokens > 0 {
746 usage.CacheWriteTokens = m.Usage.CacheCreationInputTokens
747 }
748 switch m.Delta.StopReason {
749 case "end_turn", "stop_sequence":
750 stop = StopEnd
751 case "max_tokens":
752 stop = StopLength
753 case "tool_use":
754 stop = StopToolUse
755 }
756 case "message_stop":
757 sendDone()
758 return
759 case "error":
760 var e struct {
761 Error struct {
762 Type string `json:"type"`
763 Message string `json:"message"`
764 } `json:"error"`
765 }
766 _ = json.Unmarshal([]byte(ev.Data), &e)
767 stop = StopError
768 finalErr = fmt.Errorf("anthropic %s: %s", e.Error.Type, e.Error.Message)
769 sendDone()
770 return
771 }
772 _ = activeIdx
773 }
774 }
775}