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("空の応答が返された")
    }
}