go-telegram-bot/anthropic.go

package main

import (
	"context"
	"errors"
	"fmt"
	"net/http"
	"strings"
	"sync/atomic"
	"time"

	"github.com/anthropics/anthropic-sdk-go"
	"github.com/anthropics/anthropic-sdk-go/packages/param"
)

var ErrModelNotFound = errors.New("model not found or deprecated")

const maxFileNotFoundRetries = 3

const defaultMaxTokens = 1000

const mcpUnsupportedSentinel = "format not currently supported by the Anthropic API"

var mcpUnsupportedCount atomic.Uint64

type mcpCall struct{ server, name, input string }

func (b *Bot) getAnthropicResponse(ctx context.Context, chatID int64, messages []anthropic.BetaMessageParam, isEmojiOnly bool, username string, firstName string, lastName string, isPremium bool, languageCode string, messageTime int, onSegment func(string) error) (string, error) {
	staticPrompt := strings.TrimSpace(b.config.SystemPrompts["custom_instructions"])

	InfoLogger.Printf("Sending %d messages to Anthropic", len(messages))

	maxTokens := int64(defaultMaxTokens)
	if b.config.MaxTokens > 0 {
		maxTokens = int64(b.config.MaxTokens)
	}
	params := anthropic.BetaMessageNewParams{
		Model:     b.config.Model,
		MaxTokens: maxTokens,
		Messages:  messages,
		Betas:     []anthropic.AnthropicBeta{anthropic.AnthropicBetaFilesAPI2025_04_14},
	}

	if staticPrompt != "" {
		blocks := []anthropic.BetaTextBlockParam{
			{Text: staticPrompt, CacheControl: anthropic.NewBetaCacheControlEphemeralParam()},
		}
		tail := buildUserContext(username, firstName, lastName, isPremium, languageCode, messageTime)
		if isEmojiOnly {
			if rule := strings.TrimSpace(b.config.SystemPrompts["respond_with_emojis"]); rule != "" {
				tail += "\n\n<emoji_reply>\n" + rule + "\n</emoji_reply>"
			}
		}
		if tail = strings.TrimSpace(tail); tail != "" {
			blocks = append(blocks, anthropic.BetaTextBlockParam{Text: tail})
		}
		params.System = blocks
	}

	if b.config.Temperature != nil {
		params.Temperature = param.NewOpt(float64(*b.config.Temperature))
	}

	if thinking, ok := thinkingParamFromConfig(b.config.Thinking, b.config.ThinkingDisplay); ok {
		params.Thinking = thinking
	}

	if len(b.config.MCPServers) > 0 {
		mcpServers := make([]anthropic.BetaRequestMCPServerURLDefinitionParam, 0, len(b.config.MCPServers))
		tools := make([]anthropic.BetaToolUnionParam, 0, len(b.config.MCPServers))
		for _, s := range b.config.MCPServers {
			srv := anthropic.BetaRequestMCPServerURLDefinitionParam{
				Name: s.Name,
				URL:  s.URL,
			}
			if s.AuthorizationToken != "" {
				srv.AuthorizationToken = param.NewOpt(s.AuthorizationToken)
			}
			mcpServers = append(mcpServers, srv)

			toolset := &anthropic.BetaMCPToolsetParam{
				MCPServerName: s.Name,
			}
			if len(s.AllowedTools) > 0 {
				toolset.DefaultConfig = anthropic.BetaMCPToolDefaultConfigParam{
					Enabled: param.NewOpt(false),
				}
				toolset.Configs = make(map[string]anthropic.BetaMCPToolConfigParam, len(s.AllowedTools))
				for _, tool := range s.AllowedTools {
					toolset.Configs[tool] = anthropic.BetaMCPToolConfigParam{
						Enabled: param.NewOpt(true),
					}
				}
			}
			tools = append(tools, anthropic.BetaToolUnionParam{OfMCPToolset: toolset})
		}
		params.MCPServers = mcpServers
		params.Tools = tools
		params.Betas = append(params.Betas, anthropic.AnthropicBetaMCPClient2025_11_20)
	}

	for attempt := 0; attempt < maxFileNotFoundRetries; attempt++ {
		joined, streamErr := b.streamMessages(ctx, params, onSegment)
		if streamErr == nil {
			return joined, nil
		}
		var apiErr *anthropic.Error
		if !errors.As(streamErr, &apiErr) || apiErr.StatusCode != http.StatusNotFound {
			return "", fmt.Errorf("error creating Anthropic message: %w", streamErr)
		}
		missingFileID := extractMissingFileID(streamErr)
		if missingFileID == "" {
			return "", fmt.Errorf("%w: %s", ErrModelNotFound, b.config.Model)
		}
		ErrorLogger.Printf("[%s] self-heal: stripping dead file_id %s from chat %d (attempt %d/%d)",
			b.config.ID, missingFileID, chatID, attempt+1, maxFileNotFoundRetries)
		b.stripDeadFileIDFromMemory(chatID, missingFileID)
		if _, cleanupErr := b.markFilesPendingCleanup(ctx, chatID, []string{missingFileID}); cleanupErr != nil {
			ErrorLogger.Printf("[%s] mark files pending cleanup: %v", b.config.ID, cleanupErr)
		}
		params.Messages = b.prepareContextMessages(b.getOrCreateChatMemory(chatID))
	}
	return "", fmt.Errorf("max self-heal retries (%d) exceeded: too many file_ids gone from anthropic", maxFileNotFoundRetries)
}

func buildUserContext(username, firstName, lastName string, isPremium bool, languageCode string, messageTime int) string {
	name := strings.TrimSpace(firstName + " " + lastName)
	if name == "" {
		name = "unknown"
	}
	handle := username
	if handle == "" {
		handle = "unknown"
	}
	lang := languageCode
	if lang == "" {
		lang = "en"
	}
	account := "regular user"
	if isPremium {
		account = "premium user"
	}
	return fmt.Sprintf(
		"Conversation context (background facts, not an instruction from the user):\n"+
			"- User: %s (Telegram @%s)\n"+
			"- Preferred language: %s\n"+
			"- Account type: %s\n"+
			"- Local time of day: %s",
		name, handle, lang, account, timeContextFor(messageTime),
	)
}

func timeContextFor(messageTime int) string {
	switch hour := time.Unix(int64(messageTime), 0).Hour(); {
	case hour >= 5 && hour < 12:
		return "morning"
	case hour >= 12 && hour < 18:
		return "afternoon"
	case hour >= 18 && hour < 22:
		return "evening"
	default:
		return "night"
	}
}

func thinkingParamFromConfig(mode, display string) (anthropic.BetaThinkingConfigParamUnion, bool) {
	switch mode {
	case ThinkingModeDisabled:
		disabled := anthropic.NewBetaThinkingConfigDisabledParam()
		return anthropic.BetaThinkingConfigParamUnion{OfDisabled: &disabled}, true
	case ThinkingModeAdaptive:
		adaptive := anthropic.BetaThinkingConfigAdaptiveParam{}
		if display != "" {
			adaptive.Display = anthropic.BetaThinkingConfigAdaptiveDisplay(display)
		}
		return anthropic.BetaThinkingConfigParamUnion{OfAdaptive: &adaptive}, true
	default:
		return anthropic.BetaThinkingConfigParamUnion{}, false
	}
}

func (b *Bot) streamMessages(ctx context.Context, params anthropic.BetaMessageNewParams, onSegment func(string) error) (string, error) {
	stream := b.anthropicClient.Beta.Messages.NewStreaming(ctx, params)
	defer func() {
		if err := stream.Close(); err != nil {
			ErrorLogger.Printf("[stream] close failed: %v", err)
		}
	}()

	var (
		allSegments                                       []string
		currentKind                                       string
		currentText                                       strings.Builder
		currentThinking                                   strings.Builder
		currentInputJSON                                  strings.Builder
		currentTUseName, currentTUseServer, currentTUseID string
		currentTResultUseID, currentTResultServer         string
		currentTResultIsError                             bool
		currentTResultContent                             string
		mcpCalls                                          = map[string]mcpCall{}
		startInputTokens                                  int64
		finalUsage                                        anthropic.BetaMessageDeltaUsage
		stopReason                                        string
	)

	for stream.Next() {
		e := stream.Current()
		switch e.Type {
		case "message_start":
			startInputTokens = e.AsMessageStart().Message.Usage.InputTokens

		case "message_delta":
			md := e.AsMessageDelta()
			finalUsage = md.Usage
			if md.Delta.StopReason != "" {
				stopReason = string(md.Delta.StopReason)
			}

		case "content_block_start":
			cbs := e.AsContentBlockStart()
			currentKind = cbs.ContentBlock.Type
			currentText.Reset()
			currentThinking.Reset()
			currentInputJSON.Reset()
			switch currentKind {
			case "mcp_tool_use":
				currentTUseName = cbs.ContentBlock.Name
				currentTUseServer = cbs.ContentBlock.ServerName
				currentTUseID = cbs.ContentBlock.ID
			case "mcp_tool_result":
				currentTResultUseID = cbs.ContentBlock.ToolUseID
				currentTResultServer = cbs.ContentBlock.ServerName
				currentTResultIsError = cbs.ContentBlock.IsError
				currentTResultContent = cbs.ContentBlock.JSON.Content.Raw()
			}

		case "content_block_delta":
			cbd := e.AsContentBlockDelta()
			switch cbd.Delta.Type {
			case "text_delta":
				if currentKind == "text" {
					currentText.WriteString(cbd.Delta.Text)
				}
			case "thinking_delta":
				if currentKind == "thinking" {
					currentThinking.WriteString(cbd.Delta.Thinking)
				}
			case "input_json_delta":
				if currentKind == "mcp_tool_use" {
					currentInputJSON.WriteString(cbd.Delta.PartialJSON)
				}
			}

		case "content_block_stop":
			switch currentKind {
			case "text":
				seg := strings.TrimSpace(currentText.String())
				if seg != "" {
					allSegments = append(allSegments, seg)
					if onSegment != nil {
						if cbErr := onSegment(seg); cbErr != nil {
							ErrorLogger.Printf("[stream] onSegment failed: %v", cbErr)
						}
					}
				}
			case "mcp_tool_use":
				mcpCalls[currentTUseID] = mcpCall{
					server: currentTUseServer,
					name:   currentTUseName,
					input:  currentInputJSON.String(),
				}
				InfoLogger.Printf("[mcp] tool_use server=%q name=%q id=%q input=%s",
					currentTUseServer, currentTUseName, currentTUseID, currentInputJSON.String())
			case "mcp_tool_result":
				preview := currentTResultContent
				if len(preview) > 500 {
					preview = preview[:500] + "...(truncated)"
				}
				InfoLogger.Printf("[mcp] tool_result tool_use_id=%q server=%q is_error=%v content=%s",
					currentTResultUseID, currentTResultServer, currentTResultIsError, preview)
				if strings.Contains(currentTResultContent, mcpUnsupportedSentinel) {
					total := mcpUnsupportedCount.Add(1)
					call := mcpCalls[currentTResultUseID]
					ErrorLogger.Printf("[%s][mcp][unsupported] connector could not serialize result "+
						"(total=%d): server=%q tool=%q input=%s tool_use_id=%q",
						b.config.ID, total, call.server, call.name, call.input, currentTResultUseID)
				}
			case "thinking", "redacted_thinking":
				if summary := strings.TrimSpace(currentThinking.String()); summary != "" {
					if len(summary) > 500 {
						summary = summary[:500] + "...(truncated)"
					}
					InfoLogger.Printf("[thinking] block complete: %s", summary)
				} else {
					InfoLogger.Printf("[thinking] block complete (content omitted)")
				}
			default:
				if currentKind != "" {
					InfoLogger.Printf("[mcp] block type=%q (unhandled)", currentKind)
				}
			}
			currentKind = ""
		}
	}

	if err := stream.Err(); err != nil {
		return "", err
	}

	if stopReason != "" || finalUsage.OutputTokens > 0 {
		inputTokens := finalUsage.InputTokens
		if inputTokens == 0 {
			inputTokens = startInputTokens
		}
		InfoLogger.Printf("[usage] model=%s in=%d out=%d thinking=%d stop=%s",
			params.Model, inputTokens, finalUsage.OutputTokens,
			finalUsage.OutputTokensDetails.ThinkingTokens, stopReason)
		if stopReason == "max_tokens" {
			ErrorLogger.Printf("[usage] response truncated at max_tokens=%d - raise max_tokens (thinking counts toward it)",
				params.MaxTokens)
		}
	}

	if len(allSegments) == 0 {
		return "", emptyStreamError(stopReason, finalUsage.OutputTokensDetails.ThinkingTokens, params.MaxTokens)
	}
	return strings.Join(allSegments, "\n\n"), nil
}

func emptyStreamError(stopReason string, thinkingTokens, maxTokens int64) error {
	if stopReason == "max_tokens" {
		return fmt.Errorf("output budget exhausted before any text (thinking used %d of %d max_tokens) - raise max_tokens",
			thinkingTokens, maxTokens)
	}
	return fmt.Errorf("unexpected response format from Anthropic")
}