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