homelab/review-bot/anthropic.go

package main

import (
	"context"
	"errors"
	"fmt"
	"net/http"
	"slices"
	"strings"

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

// mcpServer is the name the Gitea MCP server has inside a request.
const mcpServer = "gitea"

// maxContinuations bounds how often a paused turn (the server-side tool loop reached its limit)
// is resumed.
const maxContinuations = 30

// fallbackModels serve a refused request with Anthropic's server-side fallback model.
var fallbackModels = []string{"claude-fable-5-1", "claude-opus-5-5", "claude-opus-5", "claude-sonnet-5-5"}

// Anthropic runs the review on the Messages API. Anthropic connects to the Gitea MCP server
// itself, with a read-only token, and only the listed read tools are enabled.
type Anthropic struct {
	client   anthropic.Client
	mcpURL   string
	mcpToken string
	tools    []string
	effort   string
}

func newAnthropic(cfg Config) *Anthropic {
	opts := []option.RequestOption{option.WithAPIKey(cfg.AnthropicKey), option.WithMaxRetries(0)}
	if cfg.AnthropicURL != "" {
		opts = append(opts, option.WithBaseURL(cfg.AnthropicURL))
	}
	return &Anthropic{client: anthropic.NewClient(opts...), mcpURL: cfg.MCPURL, mcpToken: cfg.MCPToken,
		tools: cfg.MCPTools, effort: cfg.AnthropicEffort}
}

// Prepare has nothing to set up: the prompt and tools travel with every request.
func (a *Anthropic) Prepare(context.Context, []string) error { return nil }

func (a *Anthropic) params(model, task string) anthropic.BetaMessageNewParams {
	configs := map[string]anthropic.BetaMCPToolConfigParam{}
	for _, t := range a.tools {
		configs[t] = anthropic.BetaMCPToolConfigParam{Enabled: param.NewOpt(true)}
	}
	p := anthropic.BetaMessageNewParams{
		Model:     model,
		MaxTokens: 64000,
		System:    []anthropic.BetaTextBlockParam{{Text: reviewPrompt, CacheControl: anthropic.NewBetaCacheControlEphemeralParam()}},
		Messages:  []anthropic.BetaMessageParam{anthropic.NewBetaUserMessage(anthropic.NewBetaTextBlock(task))},
		MCPServers: []anthropic.BetaRequestMCPServerURLDefinitionParam{{
			Name: mcpServer, URL: a.mcpURL, AuthorizationToken: param.NewOpt(a.mcpToken),
		}},
		Tools: []anthropic.BetaToolUnionParam{{OfMCPToolset: &anthropic.BetaMCPToolsetParam{
			MCPServerName: mcpServer,
			DefaultConfig: anthropic.BetaMCPToolDefaultConfigParam{Enabled: param.NewOpt(false)},
			Configs:       configs,
		}}},
		OutputConfig: anthropic.BetaOutputConfigParam{Effort: anthropic.BetaOutputConfigEffort(a.effort)},
		Betas:        []anthropic.AnthropicBeta{anthropic.AnthropicBetaMCPClient2025_11_20},
	}
	if slices.Contains(fallbackModels, model) {
		p.Fallbacks = anthropic.BetaFallbacksParamUnion{OfDefault: constant.ValueOf[constant.Default]()}
		p.Betas = append(p.Betas, anthropic.AnthropicBetaServerSideFallback2026_07_01)
	}
	return p
}

// Ask streams the review, resuming paused turns, and takes the text after the last tool call.
func (a *Anthropic) Ask(ctx context.Context, model, task string) (Answer, error) {
	p := a.params(model, task)
	var ans Answer
	var texts []string
	for turn := 0; ; turn++ {
		stream := a.client.Beta.Messages.NewStreaming(ctx, p)
		msg := anthropic.BetaMessage{}
		for stream.Next() {
			if err := msg.Accumulate(stream.Current()); err != nil {
				_ = stream.Close()
				return Answer{}, &streamError{Provider: "anthropic", Err: err}
			}
		}
		if err := stream.Err(); err != nil {
			return Answer{}, anthropicError(err)
		}
		for _, block := range msg.Content {
			switch b := block.AsAny().(type) {
			case anthropic.BetaMCPToolUseBlock:
				ans.ToolCalls++
				if !slices.Contains(a.tools, b.Name) {
					ans.Violations = append(ans.Violations, b.Name)
				}
				texts = nil
			case anthropic.BetaMCPToolResultBlock:
				texts = nil
			case anthropic.BetaTextBlock:
				texts = append(texts, b.Text)
			}
		}
		switch msg.StopReason {
		case anthropic.BetaStopReasonPauseTurn:
			if turn >= maxContinuations {
				return Answer{}, fmt.Errorf("anthropic: the turn was still paused after %d continuations", maxContinuations)
			}
			p.Messages = append(p.Messages, msg.ToParam())
			continue
		case anthropic.BetaStopReasonRefusal:
			return Answer{}, fmt.Errorf("anthropic refused the review (%s): %s", msg.StopDetails.Category, msg.StopDetails.Explanation)
		case anthropic.BetaStopReasonMaxTokens:
			return Answer{}, errors.New("anthropic: the answer hit max_tokens")
		}
		ans.Text = strings.Join(texts, "")
		return ans, nil
	}
}

// anthropicError makes an API error an apiError, so the retry policy sees its status, and any
// other failure of the stream a streamError.
func anthropicError(err error) error {
	var api *anthropic.Error
	if !errors.As(err, &api) {
		return &streamError{Provider: "anthropic", Err: err}
	}
	resp := api.Response
	if resp == nil {
		resp = &http.Response{StatusCode: api.StatusCode, Header: http.Header{}}
	}
	return newAPIError("anthropic", http.MethodPost, "/v1/messages", resp, []byte(api.RawJSON()))
}