Claude API × Go言語 本番実装 — Anthropic Go SDK・並行処理・Tool Use・マイクロサービス統合

// tools/engine.go
package tools
 
import (
    "context"
    "encoding/json"
    "fmt"
    "sync"
 
    anthropic "github.com/anthropics/anthropic-sdk-go"
    "golang.org/x/sync/errgroup"
)
 
// ToolFunc はツール実行関数の型です
type ToolFunc func(ctx context.Context, input json.RawMessage) (string, error)
 
// ToolEngine はツールの登録と並行実行を管理します
type ToolEngine struct {
    tools map[string]ToolFunc
    defs  []anthropic.ToolParam
    mu    sync.RWMutex
}
 
func NewToolEngine() *ToolEngine {
    return &ToolEngine{
        tools: make(map[string]ToolFunc),
        defs:  make([]anthropic.ToolParam, 0),
    }
}
 
// Register はツールを登録します
func (e *ToolEngine) Register(name, description string, inputSchema interface{}, fn ToolFunc) {
    e.mu.Lock()
    defer e.mu.Unlock()
 
    schemaBytes, _ := json.Marshal(inputSchema)
    e.tools[name] = fn
    e.defs = append(e.defs, anthropic.ToolParam{
        Name:        anthropic.F(name),
        Description: anthropic.F(description),
        InputSchema: anthropic.F(anthropic.ToolInputSchemaParam{
            Type:       anthropic.F(anthropic.ToolInputSchemaTypeObject),
            Properties: anthropic.Raw[interface{}](schemaBytes),
        }),
    })
}
 
// Definitions はAPI呼び出し用のツール定義リストを返します
func (e *ToolEngine) Definitions() []anthropic.ToolParam {
    e.mu.RLock()
    defer e.mu.RUnlock()
    return e.defs
}
 
// ExecuteParallel は複数のツール呼び出しを並行実行します
func (e *ToolEngine) ExecuteParallel(ctx context.Context, calls []anthropic.ToolUseBlock) ([]anthropic.ToolResultBlockParam, error) {
    e.mu.RLock()
    defer e.mu.RUnlock()
 
    results := make([]anthropic.ToolResultBlockParam, len(calls))
    eg, ctx := errgroup.WithContext(ctx)
 
    for i, call := range calls {
        i, call := i, call // ループ変数のキャプチャ（Go 1.21以前では必須）
        eg.Go(func() error {
            fn, ok := e.tools[call.Name]
            if !ok {
                results[i] = anthropic.NewToolResultBlock(
                    call.ID,
                    fmt.Sprintf("ツール '%s' が見つかりません", call.Name),
                    true,
                )
                return nil // ツール未登録はerrとして伝播させない
            }
 
            output, err := fn(ctx, call.Input)
            if err != nil {
                // ツール実行エラーはClaudeに知らせる（goroutine全体は止めない）
                results[i] = anthropic.NewToolResultBlock(call.ID, err.Error(), true)
                return nil
            }
            results[i] = anthropic.NewToolResultBlock(call.ID, output, false)
            return nil
        })
    }
 
    if err := eg.Wait(); err != nil {
        return nil, err
    }
    return results, nil
}

// agent/loop.go
package agent
 
import (
    "context"
    "fmt"
 
    anthropic "github.com/anthropics/anthropic-sdk-go"
    "your-module/tools"
)
 
// Run はTool Use を含むエージェントの完全なループを実行します
func Run(ctx context.Context, client *anthropic.Client, engine *tools.ToolEngine, userMsg string) (string, error) {
    messages := []anthropic.MessageParam{
        anthropic.NewUserMessage(anthropic.NewTextBlock(userMsg)),
    }
 
    const maxIterations = 10 // 無限ループ防止
    for i := 0; i < maxIterations; i++ {
        resp, err := client.Messages.New(ctx, anthropic.MessageNewParams{
            Model:     anthropic.F(anthropic.ModelClaude_Sonnet_4_6),
            MaxTokens: anthropic.F(int64(4096)),
            Tools:     anthropic.F(engine.Definitions()),
            Messages:  anthropic.F(messages),
        })
        if err != nil {
            return "", fmt.Errorf("反復%d: APIエラー: %w", i+1, err)
        }
 
        // アシスタントメッセージを履歴に追加
        messages = append(messages, anthropic.NewAssistantMessage(resp.Content...))
 
        if resp.StopReason == anthropic.StopReasonEndTurn {
            for _, block := range resp.Content {
                if block.Type == anthropic.ContentBlockTypeText {
                    return block.Text, nil
                }
            }
            return "", nil
        }
 
        if resp.StopReason != anthropic.StopReasonToolUse {
            return "", fmt.Errorf("予期しないstop_reason: %s", resp.StopReason)
        }
 
        // Tool Use ブロックを収集して並行実行
        var toolCalls []anthropic.ToolUseBlock
        for _, block := range resp.Content {
            if block.Type == anthropic.ContentBlockTypeToolUse {
                toolCalls = append(toolCalls, block.AsToolUseBlock())
            }
        }
 
        toolResults, err := engine.ExecuteParallel(ctx, toolCalls)
        if err != nil {
            return "", fmt.Errorf("ツール実行エラー: %w", err)
        }
 
        resultBlocks := make([]anthropic.ContentBlockParamUnion, len(toolResults))
        for j, r := range toolResults {
            resultBlocks[j] = r
        }
        messages = append(messages, anthropic.NewUserMessage(resultBlocks...))
    }
 
    return "", fmt.Errorf("最大反復回数(%d)に達しました", maxIterations)
}

// ratelimit/limiter.go
package ratelimit
 
import (
    "context"
    "fmt"
    "time"
 
    "golang.org/x/time/rate"
)
 
// ClaudeLimiter はClaude API用のレート制限器です
type ClaudeLimiter struct {
    reqLimiter   *rate.Limiter
    tokenLimiter *rate.Limiter
}
 
// NewClaudeLimiter はレート制限器を生成します
// reqPerMin: リクエスト数/分、tokensPerMin: トークン数/分
func NewClaudeLimiter(reqPerMin, tokensPerMin int) *ClaudeLimiter {
    return &ClaudeLimiter{
        // rate.Every で1分間の均等配分を設定
        reqLimiter:   rate.NewLimiter(rate.Every(time.Minute/time.Duration(reqPerMin)), reqPerMin/10),
        tokenLimiter: rate.NewLimiter(rate.Limit(float64(tokensPerMin))/60, tokensPerMin/10),
    }
}
 
// Wait はレート制限を待機します（contextキャンセルに対応）
func (l *ClaudeLimiter) Wait(ctx context.Context, estimatedTokens int) error {
    if err := l.reqLimiter.Wait(ctx); err != nil {
        return fmt.Errorf("リクエスト制限待機中にキャンセル: %w", err)
    }
    if err := l.tokenLimiter.WaitN(ctx, estimatedTokens); err != nil {
        return fmt.Errorf("トークン制限待機中にキャンセル: %w", err)
    }
    return nil
}

// BAD: タイムアウトが短すぎてStreaming中に切れる
func badTimeout(client *anthropic.Client) {
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()
    // Streamingは5秒以上かかることが多い → タイムアウトで失敗する
    stream := client.Messages.NewStreaming(ctx, anthropic.MessageNewParams{
        Model:     anthropic.F(anthropic.ModelClaude_Sonnet_4_6),
        MaxTokens: anthropic.F(int64(2048)),
        Messages:  anthropic.F([]anthropic.MessageParam{}),
    })
    defer stream.Close()
}
 
// GOOD: Streaming全体に余裕のあるタイムアウトを設定する
func goodTimeout(client *anthropic.Client, userMsg string) {
    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
    defer cancel()
 
    stream := client.Messages.NewStreaming(ctx, anthropic.MessageNewParams{
        Model:     anthropic.F(anthropic.ModelClaude_Sonnet_4_6),
        MaxTokens: anthropic.F(int64(2048)),
        Messages: anthropic.F([]anthropic.MessageParam{
            anthropic.NewUserMessage(anthropic.NewTextBlock(userMsg)),
        }),
    })
    defer stream.Close()
    // ...
}

// BAD: 型アサーションの失敗でpanicになるパターン
block := resp.Content[0]
textBlock := block.AsTextBlock() // ToolUseBlockだとpanic
 
// GOOD: 安全な処理
for _, block := range resp.Content {
    switch block.Type {
    case anthropic.ContentBlockTypeText:
        fmt.Println(block.Text)
    case anthropic.ContentBlockTypeToolUse:
        toolBlock := block.AsToolUseBlock()
        fmt.Printf("Tool: %s, ID: %s\n", toolBlock.Name, toolBlock.ID)
    default:
        // 将来の新ブロックタイプへの対応
        fmt.Printf("未知のブロックタイプ: %s\n", block.Type)
    }
}

# nginx.conf — Streamingエンドポイント専用設定
location /api/stream {
    proxy_pass http://backend;
    proxy_read_timeout 300s;
    proxy_send_timeout 300s;
    proxy_buffering off;
    proxy_cache off;
    chunked_transfer_encoding on;
}

// BAD: Go 1.21以前では全goroutineが最後のiとcallを参照してしまう
for i, call := range calls {
    go func() {
        process(i, call) // 全goroutineがループ最終値を参照
    }()
}
 
// GOOD: ループ変数をローカル変数に代入してキャプチャ
for i, call := range calls {
    i, call := i, call // シャドウイングでキャプチャ
    go func() {
        process(i, call) // 各goroutineが正しい値を参照
    }()
}

// main.go — グレースフルシャットダウンの実装
func main() {
    cfg, err := config.Load()
    if err != nil {
        log.Fatalf("設定読み込み失敗: %v", err)
    }
 
    router := setupRouter(cfg)
    srv := &http.Server{
        Addr:         ":8080",
        Handler:      router,
        ReadTimeout:  30 * time.Second,
        WriteTimeout: 10 * time.Minute, // Streaming対応のため長めに設定
        IdleTimeout:  120 * time.Second,
    }
 
    quit := make(chan os.Signal, 1)
    signal.Notify(quit, syscall.SIGINT, syscall.SIGTERM)
 
    go func() {
        if err := srv.ListenAndServe(); err != nil && err != http.ErrServerClosed {
            log.Fatalf("サーバー起動失敗: %v", err)
        }
    }()
 
    log.Printf("サーバー起動: :%d", 8080)
    <-quit
    log.Println("シャットダウン開始（進行中リクエストの完了を待機中）")
 
    // 進行中のStreamingリクエストに最大90秒の余裕を与える
    shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 90*time.Second)
    defer shutdownCancel()
 
    if err := srv.Shutdown(shutdownCtx); err != nil {
        log.Printf("強制シャットダウン: %v", err)
    }
    log.Println("シャットダウン完了")
}

# deployment.yaml
spec:
  terminationGracePeriodSeconds: 120  # shutdownTimeout(90s) + 余裕
  containers:
    - name: claude-service
      livenessProbe:
        httpGet:
          path: /health
          port: 8080
        initialDelaySeconds: 10
        periodSeconds: 30
      readinessProbe:
        httpGet:
          path: /health
          port: 8080
        initialDelaySeconds: 5
        periodSeconds: 10

// claude/interface.go
package claude
 
import (
    "context"
    anthropic "github.com/anthropics/anthropic-sdk-go"
)
 
// MessagesService はClaude Messages APIのインターフェースです
type MessagesService interface {
    New(ctx context.Context, params anthropic.MessageNewParams) (*anthropic.Message, error)
}
 
// MockMessagesService はテスト用のモック実装です
type MockMessagesService struct {
    Responses []string // 順番に返す応答リスト
    Errors    []error
    callCount int
}
 
func (m *MockMessagesService) New(ctx context.Context, params anthropic.MessageNewParams) (*anthropic.Message, error) {
    idx := m.callCount
    m.callCount++
 
    if idx < len(m.Errors) && m.Errors[idx] != nil {
        return nil, m.Errors[idx]
    }
 
    response := "デフォルト応答"
    if idx < len(m.Responses) {
        response = m.Responses[idx]
    }
 
    return &anthropic.Message{
        Content: []anthropic.ContentBlock{
            {Type: anthropic.ContentBlockTypeText, Text: response},
        },
        StopReason: anthropic.StopReasonEndTurn,
    }, nil
}
 
// agent_test.go の例
func TestAgentLoop(t *testing.T) {
    mock := &MockMessagesService{
        Responses: []string{"最初の応答", "最終的な回答"},
    }
 
    result, err := RunWithService(context.Background(), mock, "テスト質問")
    if err != nil {
        t.Fatalf("予期しないエラー: %v", err)
    }
    if result == "" {
        t.Error("空の応答が返された")
    }
}

// claude/stream_leak_test.go
package claude
 
import (
    "context"
    "errors"
    "fmt"
    "net/http"
    "net/http/httptest"
    "testing"
    "time"
 
    "github.com/anthropics/anthropic-sdk-go"
    "github.com/anthropics/anthropic-sdk-go/option"
    "go.uber.org/goleak"
)
 
// パッケージ内の全テスト終了後に、残存goroutineがあれば失敗させます
func TestMain(m *testing.M) {
    goleak.VerifyTestMain(m)
}
 
// 終わらないSSEを返すサーバ。「長い応答の途中で読者が離脱する」状況を再現します
func endlessStreamServer(t *testing.T) *httptest.Server {
    t.Helper()
    return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        w.Header().Set("Content-Type", "text/event-stream")
        flusher, ok := w.(http.Flusher)
        if !ok {
            t.Error("http.Flusher が利用できません")
            return
        }
 
        fmt.Fprint(w, "event: message_start\n"+
            `data: {"type":"message_start","message":{"id":"msg_test","type":"message","role":"assistant","content":[],"model":"test","stop_reason":null,"usage":{"input_tokens":1,"output_tokens":1}}}`+"\n\n")
        flusher.Flush()
 
        for {
            select {
            case <-r.Context().Done(): // クライアント切断で確実に抜ける
                return
            case <-time.After(20 * time.Millisecond):
                fmt.Fprint(w, "event: content_block_delta\n"+
                    `data: {"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"a"}}`+"\n\n")
                flusher.Flush()
            }
        }
    }))
}
 
func TestGoodStreamingStopsOnCancel(t *testing.T) {
    srv := endlessStreamServer(t)
    defer srv.Close()
 
    client := anthropic.NewClient(
        option.WithAPIKey("test-key"),
        option.WithBaseURL(srv.URL),
    )
 
    ctx, cancel := context.WithCancel(context.Background())
    // 容量1: 1トークンだけ読んで放置する＝送信側がブロックする状況を作ります
    output := make(chan string, 1)
 
    done := make(chan error, 1)
    go func() { done <- goodStreaming(ctx, client, "hello", output) }()
 
    <-output // 最初のトークンだけ受け取る
    cancel() // 読者が離脱した想定
 
    select {
    case err := <-done:
        if !errors.Is(err, context.Canceled) {
            t.Fatalf("context.Canceled を期待しましたが %v でした", err)
        }
    case <-time.After(2 * time.Second):
        t.Fatal("キャンセルから2秒経ってもストリーム処理が終了していません")
    }
}

// retry/retry.go
package retry
 
import (
    "context"
    "errors"
    "fmt"
    "math/rand"
    "net/http"
    "strconv"
    "time"
 
    anthropic "github.com/anthropics/anthropic-sdk-go"
)
 
const (
    baseDelay = 500 * time.Millisecond
    maxDelay  = 20 * time.Second
)
 
// retryAfter は「再試行してよいか」と「サーバ指定の待ち時間」を返します
func retryAfter(err error) (bool, time.Duration) {
    var apiErr *anthropic.Error
    if !errors.As(err, &apiErr) {
        return false, 0 // ネットワーク層以外の失敗は握りつぶさない
    }
 
    switch apiErr.StatusCode {
    case http.StatusTooManyRequests, // 429: レート制限
        http.StatusServiceUnavailable, // 503: 一時的な停止
        529:                           // Claude 固有の過負荷コード
    default:
        return false, 0
    }
 
    // Retry-After があれば、こちらの計算より必ず優先します
    if resp := apiErr.Response; resp != nil {
        if v := resp.Header.Get("Retry-After"); v != "" {
            if sec, convErr := strconv.Atoi(v); convErr == nil {
                return true, time.Duration(sec) * time.Second
            }
        }
    }
    return true, 0
}
 
// backoff は full jitter（0〜上限の一様乱数）で待ち時間を決めます
func backoff(attempt int) time.Duration {
    d := baseDelay << uint(attempt)
    if d > maxDelay {
        d = maxDelay
    }
    return time.Duration(rand.Int63n(int64(d)))
}
 
// Do は再試行付きでAPI呼び出しを包みます（Go 1.18以降のジェネリクス）
func Do[T any](ctx context.Context, maxAttempts int, fn func() (T, error)) (T, error) {
    var zero T
    var lastErr error
 
    for attempt := 0; attempt < maxAttempts; attempt++ {
        v, err := fn()
        if err == nil {
            return v, nil
        }
        lastErr = err
 
        ok, serverWait := retryAfter(err)
        if !ok {
            return zero, err // 再試行しても直らない失敗は即座に返す
        }
        if attempt == maxAttempts-1 {
            break
        }
 
        wait := serverWait
        if wait == 0 {
            wait = backoff(attempt)
        }
 
        select {
        case <-time.After(wait):
        case <-ctx.Done():
            return zero, fmt.Errorf("再試行を中断しました: %w", ctx.Err())
        }
    }
    return zero, fmt.Errorf("%d 回試行して失敗しました: %w", maxAttempts, lastErr)
}

msg, err := retry.Do(ctx, 4, func() (*anthropic.Message, error) {
    return client.Messages.New(ctx, params)
})