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 is "enabled" for budget-based thinking (Opus 4.6 and
166 // earlier) or "adaptive" for adaptive-thinking models (Opus 4.7+).
167 Type string `json:"type"`
168 // BudgetTokens is only sent for Type=="enabled". Adaptive models
169 // reject it (400), so it is omitted for them.
170 BudgetTokens int `json:"budget_tokens,omitempty"`
171}
172
173// anthOutputConfig carries the effort knob used by adaptive-thinking
174// models to control reasoning depth ("low"|"medium"|"high"|"xhigh").
175type anthOutputConfig struct {
176 Effort string `json:"effort,omitempty"`
177}
178
179type anthRequest struct {
180 Model string `json:"model"`
181 MaxTokens int `json:"max_tokens"`
182 System []anthSystemBlock `json:"system,omitempty"`
183 Messages []anthMessage `json:"messages"`
184 Tools []anthTool `json:"tools,omitempty"`
185 Temperature *float32 `json:"temperature,omitempty"`
186 Thinking *anthThinking `json:"thinking,omitempty"`
187 OutputConfig *anthOutputConfig `json:"output_config,omitempty"`
188 Stream bool `json:"stream"`
189}
190
191// usesAdaptiveThinking reports whether a model only supports the
192// adaptive thinking mode (Opus 4.7 and later, including Opus 5). These models reject
193// explicit thinking budgets and non-default sampling parameters. The
194// catalog flag is authoritative; the id-substring fallback catches
195// the same family when reached through an Anthropic-Messages-
196// compatible proxy whose catalog row predates the flag.
197func usesAdaptiveThinking(m Model) bool {
198 if m.AdaptiveThinking || m.AdaptiveThinkingCompat {
199 return true
200 }
201 id := strings.ToLower(m.ID)
202 for _, marker := range []string{"opus-4-7", "opus-4.7", "opus-4-8", "opus-4.8", "opus-5", "opus.5", "sonnet-5", "fable-5"} {
203 if strings.Contains(id, marker) {
204 return true
205 }
206 }
207 return false
208}
209
210// ---- request building ----
211
212func (c *anthropicClient) buildRequest(req Request) (*anthRequest, error) {
213 // Look up under the client's actual provider id, which may be "anthropic"
214 // (default) or a third party that speaks the Anthropic Messages API
215 // (kimi, fireworks, minimax, vercel-ai-gateway, ...). Falling back to a
216 // provider-agnostic lookup keeps things working for catalog-less
217 // configurations (e.g. user passes --model on an obscure third party).
218 m, err := FindModel(c.Name(), req.Model)
219 if err != nil {
220 if m2, err2 := FindModel("", req.Model); err2 == nil {
221 m = m2
222 err = nil
223 }
224 }
225 if err != nil {
226 return nil, err
227 }
228 maxTok := req.MaxTokens
229 if maxTok <= 0 {
230 maxTok = m.MaxOutput
231 }
232
233 reasoning := ClampReasoningForModel(m, req.Reasoning)
234 adaptive := usesAdaptiveThinking(m)
235 adaptiveEffort := adaptive && !m.AdaptiveThinkingCompat
236
237 out := &anthRequest{
238 Model: req.Model,
239 MaxTokens: maxTok,
240 Stream: true,
241 }
242 // Anthropic's adaptive-thinking models reject non-default sampling
243 // params. Compatible implementations may use the adaptive thinking type
244 // while continuing to support temperature.
245 if !adaptiveEffort {
246 out.Temperature = req.Temperature
247 }
248
249 stableSystem, dynamicSystem, messages := systemWithDeveloperContext(req)
250
251 // System prompt assembly differs between api-key and OAuth modes.
252 // OAuth requests MUST begin with the Claude Code identity line or
253 // Anthropic rejects them (429 rate_limit_error with zero tokens used).
254 //
255 // Cache budget: anthropic caps cache_control to 4 breakpoints per
256 // request. We spend them on:
257 // 1. claude-code identity (OAuth only; stable forever)
258 // 2. user system prompt (changes per-session at most)
259 // 3. last tool definition (tools change rarely)
260 // 4. last message block (advances every turn)
261 //
262 // The identity line gets its OWN cache_control so the prefix
263 // [identity] is cacheable independently of the user system
264 // prompt. Without that, the cache prefix starts after block 2
265 // and any drift in the user prompt (e.g. the Current date
266 // line flipping at midnight) invalidates everything, including
267 // the 17 identity tokens we have to re-send every request
268 // forever.
269 if c.oauthTok != "" {
270 out.System = []anthSystemBlock{{
271 Type: "text",
272 Text: claudeCodeIdentity,
273 CacheControl: &anthCacheCtrl{Type: "ephemeral"},
274 }}
275 if stableSystem != "" {
276 out.System = append(out.System, anthSystemBlock{
277 Type: "text",
278 Text: stableSystem,
279 CacheControl: &anthCacheCtrl{Type: "ephemeral"},
280 })
281 }
282 } else if stableSystem != "" {
283 out.System = []anthSystemBlock{{
284 Type: "text",
285 Text: stableSystem,
286 CacheControl: &anthCacheCtrl{Type: "ephemeral"},
287 }}
288 }
289 if dynamicSystem != "" {
290 out.System = append(out.System, anthSystemBlock{Type: "text", Text: dynamicSystem})
291 }
292
293 if reasoning != "" && m.Reasoning {
294 if adaptive {
295 // Adaptive-thinking models decide when and how much to think and
296 // reject explicit budgets. Anthropic models also accept the
297 // output_config.effort extension; compatible implementations may not.
298 out.Thinking = &anthThinking{Type: "adaptive"}
299 if adaptiveEffort {
300 if effort := AnthropicAdaptiveEffort(reasoning); effort != "" {
301 out.OutputConfig = &anthOutputConfig{Effort: effort}
302 }
303 }
304 } else {
305 budget := anthropicReasoningBudget(reasoning)
306 if budget > 0 {
307 // Reasoning requires max_tokens > budget. Keep at least a small
308 // answer budget while respecting the model's advertised output cap.
309 const minAnswerTokens = 1024
310 if m.MaxOutput > minAnswerTokens && budget >= m.MaxOutput {
311 budget = m.MaxOutput - minAnswerTokens
312 }
313 out.Thinking = &anthThinking{Type: "enabled", BudgetTokens: budget}
314 if out.MaxTokens <= budget {
315 out.MaxTokens = budget + minAnswerTokens
316 if m.MaxOutput > 0 && out.MaxTokens > m.MaxOutput {
317 out.MaxTokens = m.MaxOutput
318 }
319 }
320 }
321 }
322 }
323
324 for _, t := range activeToolDefinitions(req.Tools, messages) {
325 name := t.Name
326 if c.oauthTok != "" {
327 name = toClaudeCodeToolName(name)
328 }
329 out.Tools = append(out.Tools, anthTool{
330 Name: name,
331 Description: t.Description,
332 InputSchema: t.Schema,
333 })
334 }
335 // Cache the last tool definition (applies cache breakpoint to the whole tools array).
336 if n := len(out.Tools); n > 0 {
337 out.Tools[n-1].CacheControl = &anthCacheCtrl{Type: "ephemeral"}
338 }
339
340 // Convert messages. Anthropic's wire format has only "user" and
341 // "assistant" roles; tool_result blocks live inside user messages.
342 // Some image-capable providers (for example Gemini image generation)
343 // can emit assistant image blocks. Anthropic only accepts image blocks
344 // in user messages, so strip assistant-side images when switching
345 // providers mid-session. The saved-path TextBlock emitted beside the
346 // image keeps the artifact reachable in chat.
347 //
348 // CRITICAL: tool_result blocks go into their OWN new user
349 // message, they are NOT merged into the preceding user message.
350 // Merging would mutate the prior user message's content array
351 // between turn N and turn N+1: turn N caches the prefix ending at
352 // [user: "read sample.ts"], turn N+1 sends
353 // [user: "read sample.ts" + tool_result=...] which is a
354 // different block sequence, busting the cache prefix match.
355 // Anthropic's API happily accepts consecutive user messages, and
356 // emitting them separately keeps each message bit-stable across
357 // turns, so the cache prefix matches for the entire history up
358 // to the newest block.
359 messages = RepairOrphanedToolResults(messages)
360 for _, msg := range messages {
361 renameTools := c.oauthTok != ""
362 switch msg.Role {
363 case RoleUser:
364 out.Messages = append(out.Messages, anthMessage{
365 Role: "user",
366 Content: convertAnthContent(msg.Content, renameTools),
367 })
368 case RoleTool:
369 out.Messages = append(out.Messages, anthMessage{
370 Role: "user",
371 Content: convertAnthContent(msg.Content, renameTools),
372 })
373 case RoleAssistant:
374 out.Messages = append(out.Messages, anthMessage{
375 Role: "assistant",
376 Content: convertAnthContent(filterAnthAssistantContent(msg.Content), renameTools),
377 })
378 }
379 }
380
381 // Tag the LAST user message with cache_control. Spends the 4th
382 // breakpoint. For prefixes under ~1024 tokens (Anthropic's
383 // minimum cacheable block size for Opus), no cache is written.
384 tagLastUserCache(out.Messages)
385
386 return out, nil
387}
388
389// tagLastUserCache marks the last block of the most recent user
390// message. One marker; combined with identity + systemPrompt +
391// last-tool, spends Anthropic's 4-breakpoint budget.
392func tagLastUserCache(msgs []anthMessage) {
393 for i := len(msgs) - 1; i >= 0; i-- {
394 if msgs[i].Role == "user" {
395 markLastBlockEphemeral(msgs[i].Content)
396 return
397 }
398 }
399}
400
401// markLastBlockEphemeral sets CacheControl on the last entry in blocks
402// regardless of whether it's a text, image, tool_use, or tool_result.
403// Each block type carries its own CacheControl pointer so we type-
404// switch + reassign the slice element.
405func markLastBlockEphemeral(blocks []interface{}) {
406 if len(blocks) == 0 {
407 return
408 }
409 i := len(blocks) - 1
410 cc := &anthCacheCtrl{Type: "ephemeral"}
411 switch v := blocks[i].(type) {
412 case anthTextBlock:
413 v.CacheControl = cc
414 blocks[i] = v
415 case anthImageBlock:
416 v.CacheControl = cc
417 blocks[i] = v
418 case anthToolUseBlock:
419 v.CacheControl = cc
420 blocks[i] = v
421 case anthToolResultBlock:
422 v.CacheControl = cc
423 blocks[i] = v
424 }
425}
426
427func anthropicReasoningBudget(level string) int {
428 return ReasoningBudget(level)
429}
430
431func filterAnthAssistantContent(blocks []Content) []Content {
432 filtered := make([]Content, 0, len(blocks))
433 for _, b := range blocks {
434 if _, ok := b.(ImageBlock); ok {
435 continue
436 }
437 filtered = append(filtered, b)
438 }
439 return filtered
440}
441
442func convertAnthContent(blocks []Content, renameTools bool) []interface{} {
443 out := make([]interface{}, 0, len(blocks))
444 for _, b := range blocks {
445 switch v := b.(type) {
446 case TextBlock:
447 if v.Text == "" {
448 continue
449 }
450 out = append(out, anthTextBlock{Type: "text", Text: v.Text})
451 case ImageBlock:
452 data, mime := anthShrinkImageBytesIfTooBig(v.Data, v.MimeType)
453 out = append(out, anthImageBlock{
454 Type: "image",
455 Source: anthImageSource{
456 Type: "base64",
457 MediaType: mime,
458 Data: base64.StdEncoding.EncodeToString(data),
459 },
460 })
461 case ToolCallBlock:
462 args := v.Arguments
463 if len(args) == 0 || !json.Valid(args) {
464 args = json.RawMessage("{}")
465 }
466 name := v.Name
467 if renameTools {
468 name = toClaudeCodeToolName(name)
469 }
470 out = append(out, anthToolUseBlock{
471 Type: "tool_use", ID: v.ID, Name: name, Input: args,
472 })
473 case ToolResultBlock:
474 // Flatten content to a string if all text; else array of blocks.
475 content, _ := anthBuildToolResultContent(v.Content)
476 out = append(out, anthToolResultBlock{
477 Type: "tool_result",
478 ToolUseID: v.CallID,
479 Content: content,
480 IsError: v.IsError,
481 })
482 }
483 }
484 return out
485}
486
487func anthBuildToolResultContent(blocks []Content) (json.RawMessage, error) {
488 onlyText := true
489 var sb strings.Builder
490 for _, b := range blocks {
491 if tb, ok := b.(TextBlock); ok {
492 if sb.Len() > 0 {
493 sb.WriteString("\n")
494 }
495 sb.WriteString(tb.Text)
496 } else {
497 onlyText = false
498 break
499 }
500 }
501 if onlyText {
502 if sb.Len() == 0 {
503 return json.Marshal("")
504 }
505 return json.Marshal(sb.String())
506 }
507 // Array form: text + image blocks.
508 arr := make([]interface{}, 0, len(blocks))
509 for _, b := range blocks {
510 switch v := b.(type) {
511 case TextBlock:
512 arr = append(arr, anthTextBlock{Type: "text", Text: v.Text})
513 case ImageBlock:
514 data, mime := anthShrinkImageBytesIfTooBig(v.Data, v.MimeType)
515 arr = append(arr, anthImageBlock{
516 Type: "image",
517 Source: anthImageSource{
518 Type: "base64",
519 MediaType: mime,
520 Data: base64.StdEncoding.EncodeToString(data),
521 },
522 })
523 }
524 }
525 return json.Marshal(arr)
526}
527
528// ---- streaming ----
529
530func (c *anthropicClient) Stream(ctx context.Context, req Request) (<-chan Event, error) {
531 if err := ValidateFastMode(c.Name(), req.FastMode); err != nil {
532 return nil, err
533 }
534 wire, err := c.buildRequest(req)
535 if err != nil {
536 return nil, err
537 }
538 body, err := json.Marshal(wire)
539 if err != nil {
540 return nil, err
541 }
542
543 // Optional debug dump: when $ZUT_DEBUG_ANTHROPIC is a file path
544 // we append every outgoing request body to it, one JSON object
545 // per line. Useful for diffing turn N vs turn N+1 to understand
546 // why the cache prefix isn't matching.
547 if dump := os.Getenv("ZUT_DEBUG_ANTHROPIC"); dump != "" {
548 if f, derr := os.OpenFile(dump, os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0o600); derr == nil {
549 _, _ = f.Write(body)
550 _, _ = f.Write([]byte{'\n'})
551 _ = f.Close()
552 }
553 }
554
555 newReq := func() (*http.Request, error) {
556 httpReq, err := http.NewRequestWithContext(ctx, "POST", c.baseURL+"/v1/messages", bytes.NewReader(body))
557 if err != nil {
558 return nil, err
559 }
560 httpReq.Header.Set("content-type", "application/json")
561 httpReq.Header.Set("anthropic-version", anthropicAPIVersion)
562 if c.oauthTok != "" {
563 // Claude-Code-shaped request: identical headers and values as the
564 // official CLI. Any drift triggers Anthropic's anti-abuse check and
565 // rate-limits (or outright blocks) the request.
566 httpReq.Header.Set("accept", "application/json")
567 httpReq.Header.Set("authorization", "Bearer "+c.oauthTok)
568 httpReq.Header.Set("anthropic-beta", "claude-code-20250219,oauth-2025-04-20,fine-grained-tool-streaming-2025-05-14")
569 httpReq.Header.Set("anthropic-dangerous-direct-browser-access", "true")
570 httpReq.Header.Set("user-agent", "claude-cli/"+claudeCodeVersion)
571 httpReq.Header.Set("x-app", "cli")
572 // Remove x-api-key entirely by NOT setting it.
573 } else {
574 httpReq.Header.Set("accept", "text/event-stream")
575 httpReq.Header.Set("x-api-key", c.apiKey)
576 }
577 // Extra headers (set by anthropic-messages-compatible third parties
578 // — kimi-coding's X-Msh-*, copilot's Editor-Plugin-Version, etc.).
579 // Applied last so callers can override defaults if they really need to.
580 for k, v := range c.headers {
581 httpReq.Header.Set(k, v)
582 }
583 return httpReq, nil
584 }
585
586 resp, err := doStreamWithRetry(ctx, c.http, newReq, req.Lifecycle)
587 if err != nil {
588 return nil, fmt.Errorf("anthropic: %w", err)
589 }
590 if resp.StatusCode != http.StatusOK {
591 b, _ := io.ReadAll(resp.Body)
592 resp.Body.Close()
593 return nil, fmt.Errorf("anthropic: http %d: %s", resp.StatusCode, strings.TrimSpace(string(b)))
594 }
595
596 out := make(chan Event, 16)
597 go c.runStream(ctx, resp, req, out)
598 return out, nil
599}
600
601func (c *anthropicClient) runStream(ctx context.Context, resp *http.Response, req Request, out chan<- Event) {
602 defer close(out)
603 defer resp.Body.Close()
604
605 // Same lookup-by-actual-provider-id pattern as buildRequest, so cost
606 // calculation works for third-party Anthropic-Messages endpoints.
607 model, _ := FindModel(c.Name(), req.Model)
608 if model.ID == "" {
609 model, _ = FindModel("", req.Model)
610 }
611 out <- EventStart{Model: req.Model, Provider: c.Name()}
612
613 raw := make(chan sseEvent, 16)
614 go readSSE(resp.Body, raw)
615
616 // State for assembling the assistant message. Blocks are indexed
617 // by their `index` field from Anthropic so we can preserve the
618 // interleaved order the model emitted them in (text may come
619 // before OR after tool_use; mixing both happens frequently).
620 type blockEntry struct {
621 kind string // "text" | "tool_use"
622 textBuf strings.Builder
623 toolCall ToolCallBlock
624 toolArgs strings.Builder
625 }
626 var (
627 blocks = map[int]*blockEntry{}
628 blockOrder []int // insertion order of indexes
629 activeIdx = -1
630 usage Usage
631 stop = StopEnd
632 finalErr error
633 )
634 _ = activeIdx // read-only indicator used for legacy parity
635
636 registerBlock := func(idx int, kind string) *blockEntry {
637 if be, ok := blocks[idx]; ok {
638 return be
639 }
640 be := &blockEntry{kind: kind}
641 blocks[idx] = be
642 blockOrder = append(blockOrder, idx)
643 return be
644 }
645
646 assembleMsg := func() Message {
647 content := []Content{}
648 for _, idx := range blockOrder {
649 be := blocks[idx]
650 switch be.kind {
651 case "text":
652 if be.textBuf.Len() > 0 {
653 content = append(content, TextBlock{Text: be.textBuf.String()})
654 }
655 case "tool_use":
656 tc := be.toolCall
657 args := be.toolArgs.String()
658 if args == "" {
659 args = "{}"
660 }
661 tc.Arguments = json.RawMessage(args)
662 content = append(content, tc)
663 }
664 }
665 return Message{Role: RoleAssistant, Content: content, Time: time.Now()}
666 }
667
668 sendDone := func() {
669 usage.CostUSD = ComputeCost(model, usage)
670 out <- EventUsage{Usage: usage}
671 out <- EventDone{Stop: stop, Err: finalErr, Message: assembleMsg()}
672 }
673
674 for {
675 select {
676 case <-ctx.Done():
677 stop = StopAborted
678 finalErr = ctx.Err()
679 sendDone()
680 return
681 case ev, ok := <-raw:
682 if !ok {
683 sendDone()
684 return
685 }
686 if ev.Err != nil {
687 stop = StopError
688 finalErr = fmt.Errorf("read SSE: %w", ev.Err)
689 sendDone()
690 return
691 }
692 // Parse event payload based on event: type.
693 var payload map[string]json.RawMessage
694 if err := json.Unmarshal([]byte(ev.Data), &payload); err != nil {
695 continue
696 }
697 switch ev.Event {
698 case "content_block_start":
699 var idx int
700 if b, ok := payload["index"]; ok {
701 _ = json.Unmarshal(b, &idx)
702 }
703 var block struct {
704 Type string `json:"type"`
705 ID string `json:"id"`
706 Name string `json:"name"`
707 Text string `json:"text"`
708 Input json.RawMessage `json:"input"`
709 }
710 if b, ok := payload["content_block"]; ok {
711 _ = json.Unmarshal(b, &block)
712 }
713 activeIdx = idx
714 switch block.Type {
715 case "tool_use":
716 name := block.Name
717 if c.oauthTok != "" {
718 name = fromClaudeCodeToolName(name, req.Tools)
719 }
720 be := registerBlock(idx, "tool_use")
721 be.toolCall = ToolCallBlock{ID: block.ID, Name: name}
722 out <- EventToolStart{ID: block.ID, Name: name}
723 case "text":
724 registerBlock(idx, "text")
725 case "thinking":
726 // not surfaced
727 }
728 case "content_block_delta":
729 var idx int
730 if b, ok := payload["index"]; ok {
731 _ = json.Unmarshal(b, &idx)
732 }
733 var d struct {
734 Type string `json:"type"`
735 Text string `json:"text"`
736 PartialJSON string `json:"partial_json"`
737 Thinking string `json:"thinking"`
738 }
739 if b, ok := payload["delta"]; ok {
740 _ = json.Unmarshal(b, &d)
741 }
742 switch d.Type {
743 case "text_delta":
744 if be, ok := blocks[idx]; ok && be.kind == "text" {
745 be.textBuf.WriteString(d.Text)
746 }
747 out <- EventTextDelta{Delta: d.Text}
748 case "input_json_delta":
749 if be, ok := blocks[idx]; ok && be.kind == "tool_use" {
750 be.toolArgs.WriteString(d.PartialJSON)
751 out <- EventToolArgs{ID: be.toolCall.ID, Delta: d.PartialJSON}
752 }
753 case "thinking_delta":
754 // Not surfaced in v1.
755 }
756 case "content_block_stop":
757 var idx int
758 if b, ok := payload["index"]; ok {
759 _ = json.Unmarshal(b, &idx)
760 }
761 if be, ok := blocks[idx]; ok && be.kind == "tool_use" {
762 out <- EventToolEnd{ID: be.toolCall.ID}
763 }
764 activeIdx = -1
765 case "message_start":
766 var m struct {
767 Message struct {
768 Usage struct {
769 InputTokens int `json:"input_tokens"`
770 OutputTokens int `json:"output_tokens"`
771 CacheReadInputTokens int `json:"cache_read_input_tokens"`
772 CacheCreationInputTokens int `json:"cache_creation_input_tokens"`
773 } `json:"usage"`
774 } `json:"message"`
775 }
776 _ = json.Unmarshal([]byte(ev.Data), &m)
777 // Anthropic sends cumulative values on message_start and
778 // again on message_delta (refreshed), so assign, don't
779 // accumulate. Accumulating doubles cache_creation_input
780 // which can be 50-70% of cost.
781 usage.InputTokens = m.Message.Usage.InputTokens
782 usage.OutputTokens = m.Message.Usage.OutputTokens
783 usage.CacheReadTokens = m.Message.Usage.CacheReadInputTokens
784 usage.CacheWriteTokens = m.Message.Usage.CacheCreationInputTokens
785 case "message_delta":
786 var m struct {
787 Delta struct {
788 StopReason string `json:"stop_reason"`
789 } `json:"delta"`
790 Usage struct {
791 InputTokens int `json:"input_tokens"`
792 OutputTokens int `json:"output_tokens"`
793 CacheReadInputTokens int `json:"cache_read_input_tokens"`
794 CacheCreationInputTokens int `json:"cache_creation_input_tokens"`
795 } `json:"usage"`
796 }
797 _ = json.Unmarshal([]byte(ev.Data), &m)
798 // Refresh usage from the latest cumulative totals
799 // Anthropic provides. Only apply non-zero fields in case
800 // a given delta only carries output tokens.
801 if m.Usage.InputTokens > 0 {
802 usage.InputTokens = m.Usage.InputTokens
803 }
804 if m.Usage.OutputTokens > 0 {
805 usage.OutputTokens = m.Usage.OutputTokens
806 }
807 if m.Usage.CacheReadInputTokens > 0 {
808 usage.CacheReadTokens = m.Usage.CacheReadInputTokens
809 }
810 if m.Usage.CacheCreationInputTokens > 0 {
811 usage.CacheWriteTokens = m.Usage.CacheCreationInputTokens
812 }
813 switch m.Delta.StopReason {
814 case "end_turn", "stop_sequence":
815 stop = StopEnd
816 case "max_tokens":
817 stop = StopLength
818 case "tool_use":
819 stop = StopToolUse
820 }
821 case "message_stop":
822 sendDone()
823 return
824 case "error":
825 var e struct {
826 Error struct {
827 Type string `json:"type"`
828 Message string `json:"message"`
829 } `json:"error"`
830 }
831 _ = json.Unmarshal([]byte(ev.Data), &e)
832 stop = StopError
833 finalErr = fmt.Errorf("anthropic %s: %s", e.Error.Type, e.Error.Message)
834 sendDone()
835 return
836 }
837 _ = activeIdx
838 }
839 }
840}