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