Claude Code SDK 프라이빗 배포: HolySheep 게이트웨이 기반 토큰 과금과 감사 실전

저는 최근 사내 결제 마이크로서비스 7개 저장소에 코드 리뷰 자동화 파이프라인을 붙이면서 Claude Code SDK를 프라이빗 배포 형태로 재구성했습니다. 첫 주 만에 엔지니어링 팀이 Anthropic API에서 $4,237을 썼는데 어느 프로젝트의 어떤 PR이 비용을 폭증시켰는지 추적이 전혀 안 됐습니다. 결국 직접 만든 감사 미들웨어를 HolySheep AI 게이트웨이 뒤에 끼워 넣어 모든 호출에 토큰 과금과 감사 로그를 강제하도록 만들었고, 8주간 운영한 결과를 그대로 정리합니다.

5축 실사용 평가 점수

총평: 토큰 과금과 코드 감사라는 두 마리 토끼를 단일 게이트웨이에서 잡을 수 있다는 점이 결정적이었습니다. 8주 운영 후 월 $1,174의 비용을 절감했고, 감사 로그 100% 보존 요건도 충족했습니다.

왜 직접 과금 감사를 구축하지 않았나

• Anthropic 대시보드는 프로젝트별 분할이 안 되어 비용 귀속이 불가능했습니다.
• OpenAI Usage API는 3시간 지연이라 실시간 차단 로직을 짤 수 없었습니다.
• 자체 DB에 토큰 카운터를 적재하려면 토큰화 로직을 직접 검증해야 하는데, Claude·GPT-4.1·Gemini 각각의 토크나이저가 달라 운영 부담이 컸습니다.
• 외부 게이트웨이를 쓰되 무료 크레딧과 로컬 결제가 되는 곳이 필요했고, 후보 비교 끝에 HolySheep AI를 선택했습니다.

HolySheep 게이트웨이 아키텍처 한 장 요약

저희 파이프라인은 다음 흐름으로 동작합니다. Claude Code SDK → 사내 감사 미들웨어 (Express) → https://api.holysheep.ai/v1 → 각 모델 백엔드. 모든 요청과 응답이 HolySheep 게이트웨이를 거치며 다음 메타데이터가 자동으로 부착됩니다: 요청 ID, 프로젝트 ID, 개발자 ID, 입력/출력 토큰, 청구 USD, 모델명, 지연 시간. 콘솔 대시보드에서 이 데이터를 그대로 내려받을 수 있어 별도 ETL 없이 감사 리포트가 완성됩니다.

Python SDK 래퍼 — 토큰 과금 자동 집계

"""claude_code_billing.py
Claude Code SDK + HolySheep 게이트웨이 기반 토큰 과금 래퍼.
모든 호출이 끝나면 ./billing_ledger.jsonl 에 감사 라인을 한 줄씩 append 합니다.
"""
import os
import json
import time
import hashlib
import httpx
from dataclasses import dataclass, asdict
from datetime import datetime, timezone
from typing import Optional

HOLYSHEEP_BASE = "https://api.holysheep.ai/v1"
API_KEY = os.environ["HOLYSHEEP_API_KEY"]  # HolySheep 콘솔에서 발급

모델별 1M 토큰당 가격 (USD) — output 기준
PRICING_PER_MTOK = {
    "claude-sonnet-4.5":      {"input": 3.00,  "output": 15.00},
    "gpt-4.1":                 {"input": 2.50,  "output": 8.00},
    "gemini-2.5-flash":        {"input": 0.30,  "output": 2.50},
    "deepseek-v3.2":           {"input": 0.14,  "output": 0.42},
}

LEDGER_PATH = "./billing_ledger.jsonl"

@dataclass
class BillingEntry:
    request_id: str
    project_id: str
    developer_id: str
    model: str
    input_tokens: int
    output_tokens: int
    cost_usd: float          # USD, 6자리 정밀도
    latency_ms: int
    success: bool
    http_status: int
    timestamp_utc: str

    def to_jsonl(self) -> str:
        return json.dumps(asdict(self), ensure_ascii=False)


def calc_cost(model: str, input_tokens: int, output_tokens: int) -> float:
    rate = PRICING_PER_MTOK.get(model, PRICING_PER_MTOK["claude-sonnet-4.5"])
    in_cost  = (input_tokens  / 1_000_000) * rate["input"]
    out_cost = (output_tokens / 1_000_000) * rate["output"]
    return round(in_cost + out_cost, 6)


def make_request_id(project_id: str, developer_id: str) -> str:
    seed = f"{project_id}:{developer_id}:{datetime.now(timezone.utc).isoformat()}"
    return hashlib.sha256(seed.encode()).hexdigest()[:16]


def invoke(
    *,
    prompt: str,
    project_id: str,
    developer_id: str,
    model: str = "claude-sonnet-4.5",
    system: Optional[str] = None,
) -> tuple[dict, BillingEntry]:
    start = time.perf_counter()
    request_id = make_request_id(project_id, developer_id)

    messages = []
    if system:
        messages.append({"role": "system", "content": system})
    messages.append({"role": "user", "content": prompt})

    payload = {
        "model": model,
        "messages": messages,
        "max_tokens": 1024,
    }

    headers = {
        "Authorization": f"Bearer {API_KEY}",
        "Content-Type": "application/json",
        "X-Project-ID": project_id,
        "X-Developer-ID": developer_id,
        "X-Request-ID": request_id,
    }

    try:
        with httpx.Client(timeout=30.0) as client:
            resp = client.post(
                f"{HOLYSHEEP_BASE}/chat/completions",
                headers=headers,
                json=payload,
            )
        latency_ms = int((time.perf_counter() - start) * 1000)
        data = resp.json()
        usage = data.get("usage", {})
        inp = int(usage.get("prompt_tokens", 0))
        out = int(usage.get("completion_tokens", 0))
        success = resp.status_code == 200
        cost = calc_cost(model, inp, out) if success else 0.0

        entry = BillingEntry(
            request_id=request_id,
            project_id=project_id,
            developer_id=developer_id,
            model=model,
            input_tokens=inp,
            output_tokens=out,
            cost_usd=cost,
            latency_ms=latency_ms,
            success=success,
            http_status=resp.status_code,
            timestamp_utc=datetime.now(timezone.utc).isoformat(),
        )
    except Exception as exc:
        latency_ms = int((time.perf_counter() - start) * 1000)
        entry = BillingEntry(
            request_id=request_id,
            project_id=project_id,
            developer_id=developer_id,
            model=model,
            input_tokens=0,
            output_tokens=0,
            cost_usd=0.0,
            latency_ms=latency_ms,
            success=False,
            http_status=0,
            timestamp_utc=datetime.now(timezone.utc).isoformat(),
        )
        data = {"error": str(exc)}

    with open(LEDGER_PATH, "a", encoding="utf-8") as f:
        f.write(entry.to_jsonl() + "\n")

    return data, entry


if __name__ == "__main__":
    # 실제 운영 시 developer_id 는 SSO 토큰에서 추출
    data, entry = invoke(
        prompt="Review this Python function for race conditions and suggest a fix.",
        project_id="payments-microservice",
        developer_id="dev-2187",
        model="claude-sonnet-4.5",
    )
    print(entry.to_jsonl())
    print("assistant:", data["choices"][0]["message"]["content"])

// auditMiddleware.mjs
// Express 기반 Claude Code SDK 호출 미들웨어.
// 1) 개발자별 일일 한도 검사
// 2) 모든 요청 본문 해시 적재
// 3) HolySheep 게이트웨이로 프록시
import express from "express";
import crypto from "node:crypto";
import fs from "node:fs";
import fetch from "node-fetch";

const HOLYSHEEP_BASE = "https://api.holysheep.ai/v1";
const API_KEY = process.env.HOLYSHEEP_API_KEY;

// 프로젝트별 1일 한도 (USD)
const DAILY_LIMIT_USD = {
    "payments-microservice": 25.0,
    "search-api":            10.0,
    "internal-tools":         5.0,
};

// 메모리 누적 — 운영 시 Redis 권장
const dailySpend = new Map(); // key: ${project}:${YYYY-MM-DD}

function todayKey(projectId) {
    return ${projectId}:${new Date().toISOString().slice(0, 10)};
}

export function auditMiddleware(req, res, next) {
    const projectId    = req.header("X-Project-ID");
    const developerId  = req.header("X-Developer-ID");
    const promptHash   = crypto
        .createHash("sha256")
        .update(JSON.stringify(req.body.messages ?? []))
        .digest("hex")
        .slice(0, 16);

    // 1) 한도 체크
    const spent = dailySpend.get(todayKey(projectId)) ?? 0;
    const limit = DAILY_LIMIT_USD[projectId] ?? 5.0;
    if (spent >= limit) {
        return res.status(429).json({
            error: "DAILY_LIMIT_EXCEEDED",
            project: projectId,
            spent_usd: spent,
            limit_usd: limit,
        });
    }

    // 2) 프록시
    const start = Date.now();
    fetch(${HOLYSHEEP_BASE}/chat/completions, {
        method: "POST",
        headers: {
            "Authorization":  Bearer ${API_KEY},
            "Content-Type":   "application/json",
            "X-Project-ID":   projectId,
            "X-Developer-ID": developerId,
            "X-Prompt-Hash":  promptHash,
        },
        body: JSON.stringify(req.body),
    })
        .then(async (r) => {
            const latency = Date.now() - start;
            const json = await r.json();
            const usage = json.usage ?? {};
            const cost =
                ((usage.prompt_tokens ?? 0) / 1e6) * 3.0 +
                ((usage.completion_tokens ?? 0) / 1e6) * 15.0; // claude-sonnet-4.5 기준

            dailySpend.set(todayKey(projectId), spent + cost);

            // 3) 감사 라인 적재
            fs.appendFileSync(
                "./audit.log",
                JSON.stringify({
                    ts: new Date().toISOString(),
                    project_id: projectId,
                    developer_id: developerId,
                    prompt_hash: promptHash,
                    model: req.body.model,
                    input_tokens: usage.prompt_tokens ?? 0,
                    output_tokens: usage.completion_tokens ?? 0,
                    cost_usd: Number(cost.toFixed(6)),
                    latency_ms: latency,
                    http_status: r.status,
                }) + "\n",
            );

            res.status(r.status).json(json);
        })
        .catch((err) => {
            res.status(502).json({ error: "GATEWAY_UNREACHABLE", detail: String(err) });
        });
}

-- audit.jsonl 과 billing_ledger.jsonl 을 합쳐 프로젝트·개발자별 월간 비용 집계
-- (DuckDB CLI 기준)
COPY (
    SELECT
        project_id,
        developer_id,
        model,
        COUNT(*)                              AS calls,
        SUM(input_tokens)                     AS sum_in,
        SUM(output_tokens)                    AS sum_out,
        ROUND(SUM(cost_usd), 2)               AS cost_usd,
        ROUND(AVG(latency_ms), 0)             AS avg_latency_ms,
        ROUND(100.0 * SUM(CASE WHEN success THEN 1 ELSE 0 END) / COUNT(*), 2) AS success_pct
    FROM read_json_auto('./billing_ledger.jsonl')
    WHERE timestamp_utc LIKE '2025-01%'
    GROUP BY project_id, developer_id, model
    ORDER BY cost_usd DESC
) TO 'monthly_report_2025_01.csv' (HEADER, DELIMITER ',');