凌晨 3시, 본인은 수백 개의 문서를 분석하는 배치 작업을 실행했습니다. 예상 처리 시간은 30분 이상. 그런데 갑자기 504 Gateway Timeout 오류가 발생하며 연결이 끊겼습니다. 다시 시도할까? 아니면 결과를 받을 방법이 없을까? 이 충격적인 경험이 본인을 Claude API 백그라운드 태스크와 Webhook 콜백 시스템으로 이끌었습니다.

본 글에서는 HolySheep AI 게이트웨이를 통해 Claude API의 긴 작업을 안정적으로 처리하는 방법을 실전 경험 바탕으로 설명드리겠습니다.

문제 상황: 왜 긴 작업 처리가 중요한가?

Claude API를 사용할 때 아래와 같은 상황이 빈번합니다:

이 작업들은 HTTP 요청 타임아웃 (30초~120초)을 초과할 수 있습니다. 표준 동기 호출로는 이러한 긴 작업을 완료할 수 없으며, 백그라운드 처리와 비동기 콜백 메커니즘이 필수적입니다.

Claude API 백그라운드 태스크 이해

Anthropic의 Claude API는 긴 작업 완료를 위해 두 가지 메커니즘을 제공합니다:

1. 오래 실행되는 요청의 처리 원리

기본적으로 Claude API 요청은 완료될 때까지 연결을 유지합니다. 그러나:

2. HolySheep AI를 통한 최적화

지금 가입하여 HolySheep AI를 사용하면:

실전 구현: Python으로 백그라운드 태스크 처리

기본 설정

"""
Claude API 백그라운드 태스크 처리 - HolySheep AI 게이트웨이
저자: HolySheep AI 기술 블로그
"""

import requests
import time
import json
import hmac
import hashlib
from typing import Optional, Dict, Any
from dataclasses import dataclass
from enum import Enum

class TaskStatus(Enum):
    PENDING = "pending"
    PROCESSING = "processing"
    COMPLETED = "completed"
    FAILED = "failed"
    CANCELLED = "cancelled"

@dataclass
class ClaudeConfig:
    """Claude API 설정"""
    api_key: str = "YOUR_HOLYSHEEP_API_KEY"  # HolySheep AI 키로 교체
    base_url: str = "https://api.holysheep.ai/v1"
    model: str = "claude-sonnet-4-20250514"
    max_tokens: int = 8192
    timeout: int = 300  # 5분 타임아웃
    
    # Webhook 설정 (선택사항)
    webhook_url: Optional[str] = None
    webhook_secret: Optional[str] = None

class ClaudeLongTaskHandler:
    """
    Claude API 긴 작업 처리 핸들러
    - 재시도 로직 자동화
    - Webhook 콜백 지원
    - 진행 상황 추적
    """
    
    def __init__(self, config: ClaudeConfig):
        self.config = config
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {config.api_key}",
            "Content-Type": "application/json",
            "Anthropic-Version": "2023-06-01"
        })
    
    def _verify_webhook_signature(self, payload: bytes, signature: str) -> bool:
        """Webhook 서명 검증"""
        if not self.config.webhook_secret:
            return True
        
        expected = hmac.new(
            self.config.webhook_secret.encode(),
            payload,
            hashlib.sha256
        ).hexdigest()
        
        return hmac.compare_digest(f"sha256={expected}", signature)
    
    def _handle_webhook(self, webhook_data: Dict[str, Any]) -> Optional[Dict[str, Any]]:
        """Webhook 콜백 처리"""
        payload = json.dumps(webhook_data).encode()
        # 실제 환경에서는 HTTP 요청으로 받은 서명 사용
        # signature = request.headers.get("X-Webhook-Signature")
        
        result = webhook_data.get("result", {})
        task_id = webhook_data.get("task_id")
        status = webhook_data.get("status")
        
        print(f"[Webhook] Task {task_id}: {status}")
        
        if status == "completed":
            return result.get("content")
        elif status == "failed":
            error = result.get("error", {})
            print(f"[Webhook Error] {error.get('type')}: {error.get('message')}")
            return None
        
        return None
    
    def process_long_task_sync(
        self,
        prompt: str,
        max_retries: int = 3,
        retry_delay: int = 5
    ) -> Optional[Dict[str, Any]]:
        """
        긴 작업 동기 처리 (재시도 로직 포함)
        - HolySheep AI의 안정적 연결 활용
        """
        
        payload = {
            "model": self.config.model,
            "max_tokens": self.config.max_tokens,
            "messages": [{"role": "user", "content": prompt}]
        }
        
        # Webhook URL 추가 (선택사항)
        if self.config.webhook_url:
            payload["metadata"] = {
                "webhook_url": self.config.webhook_url
            }
        
        last_error = None
        
        for attempt in range(max_retries):
            try:
                print(f"[Attempt {attempt + 1}/{max_retries}] Sending request...")
                
                response = self.session.post(
                    f"{self.config.base_url}/messages",
                    json=payload,
                    timeout=self.config.timeout
                )
                
                if response.status_code == 200:
                    result = response.json()
                    print(f"[Success] Response received in {response.elapsed.total_seconds():.2f}s")
                    return result
                
                elif response.status_code == 429:
                    # Rate limit - 지수 백오프
                    wait_time = 2 ** attempt + 1
                    print(f"[Rate Limited] Waiting {wait_time}s before retry...")
                    time.sleep(wait_time)
                    
                elif response.status_code == 500:
                    # 서버 오류 - 재시도
                    last_error = f"Server Error: {response.text}"
                    print(f"[Server Error] Retrying in {retry_delay}s...")
                    time.sleep(retry_delay)
                    
                else:
                    last_error = f"HTTP {response.status_code}: {response.text}"
                    print(f"[Error] {last_error}")
                    
            except requests.exceptions.Timeout:
                last_error = "Request Timeout"
                print(f"[Timeout] Attempt {attempt + 1} failed, retrying...")
                time.sleep(retry_delay)
                
            except requests.exceptions.ConnectionError as e:
                last_error = f"Connection Error: {str(e)}"
                print(f"[Connection Error] {last_error}, retrying...")
                time.sleep(retry_delay)
                
            except requests.exceptions.RequestException as e:
                last_error = f"Request Exception: {str(e)}"
                print(f"[Exception] {last_error}")
                break
        
        print(f"[Failed] All {max_retries} attempts exhausted")
        print(f"[Last Error] {last_error}")
        return None


사용 예제

if __name__ == "__main__":
    config = ClaudeConfig(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        webhook_url="https://your-server.com/webhook/claude-result"
    )
    
    handler = ClaudeLongTaskHandler(config)
    
    # 긴 프롬프트 예시
    long_prompt = """
    다음 100개의 문서를 분석하고 핵심 포인트를 요약해주세요:
    [문서 내용 1...]
    [문서 내용 2...]
    ...
    """
    
    result = handler.process_long_task_sync(long_prompt)
    
    if result:
        print(f"작업 완료: {len(result.get('content', []))} 개의 컨텐츠 블록")

Webhook 서버 구현

"""
Claude API Webhook 콜백 서버 - FastAPI 기반
HolySheep AI에서 긴 작업 완료 알림 수신
"""

from fastapi import FastAPI, Request, HTTPException, Header
from pydantic import BaseModel
from typing import Optional, List, Dict, Any
import hmac
import hashlib
import json
import asyncio
from datetime import datetime

app = FastAPI(title="Claude Webhook Server")


설정

WEBHOOK_SECRET = "your-webhook-secret-key"  # HolySheep AI Dashboard에서 확인
PROCESSING_QUEUE = asyncio.Queue()

class ClaudeMessageContent(BaseModel):
    """Claude 메시지 컨텐츠 구조"""
    type: str
    text: Optional[str] = None
    id: Optional[str] = None

class ClaudeResponse(BaseModel):
    """Claude API 응답 구조"""
    id: str
    type: str
    role: str
    content: List[ClaudeMessageContent]
    model: str
    stop_reason: Optional[str] = None
    stop_sequence: Optional[str] = None
    usage: Optional[Dict[str, int]] = None

class WebhookPayload(BaseModel):
    """Webhook 페이로드 구조"""
    task_id: str
    status: str  # "completed", "failed", "processing"
    result: Optional[Dict[str, Any]] = None
    error: Optional[Dict[str, str]] = None
    completed_at: Optional[str] = None

class ClaudeWebhookHandler:
    """Claude Webhook 처리 로직"""
    
    def __init__(self, secret: str):
        self.secret = secret
    
    def verify_signature(self, payload: bytes, signature: str) -> bool:
        """HMAC-SHA256 서명 검증"""
        expected = hmac.new(
            self.secret.encode(),
            payload,
            hashlib.sha256
        ).hexdigest()
        
        expected_signature = f"sha256={expected}"
        
        return hmac.compare_digest(expected_signature, signature)
    
    async def save_result(self, result: ClaudeResponse, task_id: str):
        """결과를 데이터베이스 또는 파일로 저장"""
        # 실제 환경에서는 DB 저장 로직 구현
        filename = f"results/{task_id}_{datetime.now().timestamp()}.json"
        
        with open(filename, "w", encoding="utf-8") as f:
            json.dump(result.model_dump(), f, ensure_ascii=False, indent=2)
        
        print(f"[Saved] Result to {filename}")
    
    async def process_completed_task(self, payload: WebhookPayload):
        """완료된 작업 후처리"""
        if payload.status != "completed":
            return
        
        result = payload.result
        if not result:
            print(f"[Warning] Task {payload.task_id} completed but no result")
            return
        
        try:
            response = ClaudeResponse(**result)
            
            # 토큰 사용량 로깅 (비용 추적에 활용)
            if response.usage:
                input_tokens = response.usage.get("input_tokens", 0)
                output_tokens = response.usage.get("output_tokens", 0)
                
                # HolySheep AI 가격 계산
                # Claude Sonnet 4.5: $15/MTok (입력), $15/MTok (출력)
                input_cost = (input_tokens / 1_000_000) * 15  # USD
                output_cost = (output_tokens / 1_000_000) * 15  # USD
                total_cost = input_cost + output_cost
                
                print(f"""
[Cost Analysis] Task {payload.task_id}
├─ Input Tokens:  {input_tokens:,}
├─ Output Tokens: {output_tokens:,}
├─ Input Cost:    ${input_cost:.4f}
├─ Output Cost:   ${output_cost:.4f}
└─ Total Cost:    ${total_cost:.4f}
                """)
            
            await self.save_result(response, payload.task_id)
            
        except Exception as e:
            print(f"[Error] Processing task {payload.task_id}: {e}")


Webhook 핸들러 인스턴스

handler = ClaudeWebhookHandler(WEBHOOK_SECRET)

@app.post("/webhook/claude-result")
async def receive_claude_webhook(
    request: Request,
    x_webhook_signature: Optional[str] = Header(None)
):
    """
    HolySheep AI에서 Claude 작업 완료 시 콜백 수신
    """
    body = await request.body()
    
    # 서명 검증 (production에서는 필수)
    if x_webhook_signature:
        if not handler.verify_signature(body, x_webhook_signature):
            raise HTTPException(status_code=401, detail="Invalid signature")
    
    # JSON 파싱
    try:
        data = await request.json()
    except json.JSONDecodeError:
        raise HTTPException(status_code=400, detail="Invalid JSON")
    
    # Webhook 페이로드 검증
    payload = WebhookPayload(**data)
    
    print(f"[Webhook Received] Task: {payload.task_id}, Status: {payload.status}")
    
    # 작업 상태별 처리
    if payload.status == "completed":
        await handler.process_completed_task(payload)
        return {"status": "success", "message": "Result processed"}
    
    elif payload.status == "failed":
        print(f"[Failed Task] {payload.task_id}: {payload.error}")
        return {"status": "acknowledged", "message": "Failure logged"}
    
    elif payload.status == "processing":
        print(f"[Processing] Task {payload.task_id} is being processed")
        return {"status": "acknowledged", "message": "Processing started"}
    
    return {"status": "ok"}

@app.get("/health")
async def health_check():
    """헬스 체크 엔드포인트"""
    return {"status": "healthy", "timestamp": datetime.now().isoformat()}


실행: uvicorn main:app --host 0.0.0.0 --port 8000

배치 처리 구현: 다중 문서 분석

"""
Claude API 배치 처리 시스템 - 대규모 문서 분석
HolySheep AI 게이트웨이 활용 (비용 최적화)
"""

import asyncio
import aiohttp
import json
import time
from typing import List, Dict, Any, Optional
from dataclasses import dataclass
from concurrent.futures import ThreadPoolExecutor
import queue
import threading

@dataclass
class Document:
    """분석할 문서"""
    id: str
    content: str
    category: Optional[str] = None

@dataclass
class BatchResult:
    """배치 처리 결과"""
    document_id: str
    status: str
    response: Optional[Dict[str, Any]] = None
    error: Optional[str] = None
    processing_time: float = 0.0
    tokens_used: int = 0

class ClaudeBatchProcessor:
    """
    Claude API 배치 처리 시스템
    - 동시 요청 관리
    - Rate limit 자동 처리
    - 비용 추적
    """
    
    def __init__(
        self,
        api_key: str,
        base_url: str = "https://api.holysheep.ai/v1",
        model: str = "claude-sonnet-4-20250514",
        max_concurrent: int = 5,
        requests_per_minute: int = 50
    ):
        self.api_key = api_key
        self.base_url = base_url
        self.model = model
        self.max_concurrent = max_concurrent
        self.rpm_limit = requests_per_minute
        
        # Rate limiting
        self.request_times: List[float] = []
        self.rate_limit_lock = threading.Lock()
        
        # 통계
        self.stats = {
            "total_requests": 0,
            "successful": 0,
            "failed": 0,
            "total_input_tokens": 0,
            "total_output_tokens": 0,
            "total_cost_usd": 0.0
        }
        
        # HolySheep AI 가격표 (2024년 기준)
        # Claude Sonnet 4.5: $15/MTok 입력, $15/MTok 출력
        self.price_per_mtok_input = 15.0
        self.price_per_mtok_output = 15.0
    
    def _check_rate_limit(self) -> float:
        """Rate limit 확인 및 대기 시간 반환"""
        current_time = time.time()
        
        with self.rate_limit_lock:
            # 1분 이내의 요청만 필터링
            self.request_times = [
                t for t in self.request_times 
                if current_time - t < 60
            ]
            
            if len(self.request_times) >= self.rpm_limit:
                # 가장 오래된 요청 후 1분 대기
                oldest = min(self.request_times)
                wait_time = 60 - (current_time - oldest) + 0.1
                if wait_time > 0:
                    print(f"[Rate Limit] Waiting {wait_time:.2f}s...")
                    time.sleep(wait_time)
                    self.request_times = [
                        t for t in self.request_times 
                        if time.time() - t < 60
                    ]
            
            self.request_times.append(time.time())
        
        return 0.0
    
    def _calculate_cost(self, input_tokens: int, output_tokens: int) -> float:
        """토큰 사용량 기반 비용 계산"""
        input_cost = (input_tokens / 1_000_000) * self.price_per_mtok_input
        output_cost = (output_tokens / 1_000_000) * self.price_per_mtok_output
        return input_cost + output_cost
    
    async def _process_single_document(
        self,
        session: aiohttp.ClientSession,
        document: Document
    ) -> BatchResult:
        """단일 문서 처리"""
        start_time = time.time()
        
        # Rate limit 체크
        self._check_rate_limit()
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
            "Anthropic-Version": "2023-06-01"
        }
        
        payload = {
            "model": self.model,
            "max_tokens": 4096,
            "messages": [{
                "role": "user",
                "content": f"""다음 문서를 분석하고 핵심 내용을 요약해주세요:

문서 ID: {document.id}
카테고리: {document.category or '미분류'}

내용:
{document.content}

응답 형식:
{{
    "summary": "핵심 요약 (100자 이내)",
    "key_points": ["포인트 1", "포인트 2", "포인트 3"],
    "sentiment": "positive/negative/neutral",
    "action_items": ["실행 항목 1", "실행 항목 2"]
}}
"""
            }]
        }
        
        try:
            async with session.post(
                f"{self.base_url}/messages",
                json=payload,
                timeout=aiohttp.ClientTimeout(total=120)
            ) as response:
                
                processing_time = time.time() - start_time
                self.stats["total_requests"] += 1
                
                if response.status == 200:
                    result = await response.json()
                    self.stats["successful"] += 1
                    
                    # 토큰 사용량 추적
                    if "usage" in result:
                        input_tokens = result["usage"].get("input_tokens", 0)
                        output_tokens = result["usage"].get("output_tokens", 0)
                        
                        self.stats["total_input_tokens"] += input_tokens
                        self.stats["total_output_tokens"] += output_tokens
                        
                        cost = self._calculate_cost(input_tokens, output_tokens)
                        self.stats["total_cost_usd"] += cost
                        
                        return BatchResult(
                            document_id=document.id,
                            status="success",
                            response=result,
                            processing_time=processing_time,
                            tokens_used=input_tokens + output_tokens
                        )
                    
                    return BatchResult(
                        document_id=document.id,
                        status="success",
                        response=result,
                        processing_time=processing_time
                    )
                
                elif response.status == 429:
                    self.stats["failed"] += 1
                    return BatchResult(
                        document_id=document.id,
                        status="rate_limited",
                        error="Rate limit exceeded",
                        processing_time=processing_time
                    )
                
                else:
                    error_text = await response.text()
                    self.stats["failed"] += 1
                    return BatchResult(
                        document_id=document.id,
                        status="error",
                        error=f"HTTP {response.status}: {error_text}",
                        processing_time=processing_time
                    )
        
        except asyncio.TimeoutError:
            self.stats["failed"] += 1
            return BatchResult(
                document_id=document.id,
                status="timeout",
                error="Request timeout after 120s",
                processing_time=time.time() - start_time
            )
        
        except Exception as e:
            self.stats["failed"] += 1
            return BatchResult(
                document_id=document.id,
                status="error",
                error=str(e),
                processing_time=time.time() - start_time
            )
    
    async def process_batch(
        self,
        documents: List[Document],
        progress_callback=None
    ) -> List[BatchResult]:
        """배치 문서 처리 (병렬 처리)"""
        
        connector = aiohttp.TCPConnector(limit=self.max_concurrent)
        
        async with aiohttp.ClientSession(connector=connector) as session:
            tasks = []
            
            for i, doc in enumerate(documents):
                task = self._process_single_document(session, doc)
                tasks.append(task)
                
                # 진행 상황 콜백
                if progress_callback:
                    progress_callback(i + 1, len(documents))
            
            # 병렬 실행
            results = await asyncio.gather(*tasks)
        
        return results
    
    def print_statistics(self):
        """처리 통계 출력"""
        print(f"""
╔══════════════════════════════════════════════════════════╗
║              Claude 배치 처리 결과 요약                   ║
╠══════════════════════════════════════════════════════════╣
║  총 요청 수:       {self.stats['total_requests']:>6}                          ║
║  성공:             {self.stats['successful']:>6} ({self.stats['successful']/max(self.stats['total_requests'],1)*100:>5.1f}%)              ║
║  실패:             {self.stats['failed']:>6} ({self.stats['failed']/max(self.stats['total_requests'],1)*100:>5.1f}%)              ║
╠══════════════════════════════════════════════════════════╣
║  총 입력 토큰:     {self.stats['total_input_tokens']:>15,}              ║
║  총 출력 토큰:     {self.stats['total_output_tokens']:>15,}              ║
║  예상 비용 (USD):  ${self.stats['total_cost_usd']:>15.4f}              ║
╚══════════════════════════════════════════════════════════╝
        """)

async def main():
    """메인 실행 함수"""
    
    processor = ClaudeBatchProcessor(
        api_key="YOUR_HOLYSHEEP_API_KEY",
        max_concurrent=3,  # 동시 3개 요청
        requests_per_minute=50
    )
    
    # 테스트 문서 생성
    test_documents = [
        Document(
            id=f"doc_{i}",
            content=f"이것은 테스트 문서 {i}의 내용입니다. " * 50,
            category="test"
        )
        for i in range(10)
    ]
    
    def progress(current: int, total: int):
        print(f"Progress: {current}/{total} ({current/total*100:.1f}%)")
    
    print(f"총 {len(test_documents)}개 문서 처리 시작...")
    
    start_time = time.time()
    results = await processor.process_batch(
        test_documents,
        progress_callback=progress
    )
    total_time = time.time() - start_time
    
    # 결과 출력
    processor.print_statistics()
    
    print(f"\n총 처리 시간: {total_time:.2f}초")
    print(f"평균 처리 시간: {total_time/len(results):.2f}초/문서")
    
    # 성공한 결과 저장
    successful_results = [r for r in results if r.status == "success"]
    
    with open("batch_results.json", "w", encoding="utf-8") as f:
        json.dump([
            {
                "document_id": r.document_id,
                "status": r.status,
                "processing_time": r.processing_time,
                "tokens_used": r.tokens_used
            }
            for r in results
        ], f, ensure_ascii=False, indent=2)
    
    print(f"\n결과가 batch_results.json에 저장되었습니다.")

if __name__ == "__main__":
    asyncio.run(main())

HolySheep AI 활용: 추가 최적화 팁

1. 모델 선택 가이드

모델가격 ($/MTok)적합한 작업추천 상황
Claude Sonnet 4$15일반 분석, 요약가성비 최적
Claude Opus 4$75복잡한推理, 긴 컨텍스트정확도 우선
Claude Haiku$1.50빠른 분류, 간단한 작업대량 처리

2. 비용 최적화 전략

자주 발생하는 오류와 해결

오류 1: ConnectionError: timeout

# ❌ 잘못된 접근 - 타임아웃 기본값 사용
response = requests.post(url, json=payload)  # 기본 30초 타임아웃


✅ 해결 방법 1: 명시적 타임아웃 설정

from requests.exceptions import ReadTimeout, ConnectTimeout

try:
    response = requests.post(
        url,
        json=payload,
        timeout=(10, 120),  # (연결타임아웃, 읽기타임아웃)
        headers={"Connection": "keep-alive"}
    )
except (ConnectTimeout, ReadTimeout) as e:
    print(f"타임아웃 발생: {e}")
    # 재시도 로직 실행
    time.sleep(5)
    response = requests.post(url, json=payload, timeout=(30, 180))


✅ 해결 방법 2: aiohttp로 비동기 타임아웃

import aiohttp

async def fetch_with_timeout():
    timeout = aiohttp.ClientTimeout(total=300)  # 5분
    async with aiohttp.ClientSession(timeout=timeout) as session:
        async with session.post(url, json=payload) as response:
            return await response.json()

오류 2: 401 Unauthorized

# ❌ 잘못된 접근 - 환경변수 직접 참조
headers = {"Authorization": f"Bearer {os.getenv('KEY')}"}


✅ 해결 방법 1: 명시적 API 키 설정 및 검증

API_KEY = os.getenv("HOLYSHEEP_API_KEY") or "YOUR_HOLYSHEEP_API_KEY"

if not API_KEY or API_KEY == "YOUR_HOLYSHEEP_API_KEY":
    raise ValueError("""
    HolySheep AI API 키가 설정되지 않았습니다.
    1. https://www.holysheep.ai/register 에서 가입
    2. Dashboard에서 API 키 생성
    3. 환경변수 HOLYSHEEP_API_KEY 설정
    """)

headers = {
    "Authorization": f"Bearer {API_KEY}",
    "Content-Type": "application/json"
}


✅ 해결 방법 2: 키 유효성 검증 함수

def validate_api_key(key: str) -> bool:
    """API 키 형식 검증"""
    if not key:
        return False
    if len(key) < 20:
        return False
    # HolySheep AI 키는 sk-hs- 접두사
    if not key.startswith("sk-hs-"):
        print("경고: HolySheep AI 키 형식이 아닙니다")
    return True

if not validate_api_key(API_KEY):
    raise ValueError("유효하지 않은 API 키입니다")

오류 3: RateLimitError: 429 Too Many Requests

# ❌ 잘못된 접근 - 재시도 없음
response = requests.post(url, json=payload)
if response.status_code == 429:
    print("Rate limit 도달")  # 아무 처리 없음


✅ 해결 방법: 지수 백오프 재시도 로직

import time
from datetime import datetime, timedelta

def exponential_backoff_with_jitter():
    """지수 백오프 + 지터 (Jitter) for Rate Limit 처리"""
    max_retries = 5
    base_delay = 1
    
    for attempt in range(max_retries):
        # HolySheep AI는 Retry-After 헤더를 반환할 수 있음
        retry_after = response.headers.get("Retry-After")
        
        if retry_after:
            wait_time = int(retry_after)
        else:
            # 지수 백오프: 1s, 2s, 4s, 8s, 16s...
            wait_time = base_delay * (2 ** attempt)
            
            # 지터 추가 (0.5~1.5 배율 랜덤)
            import random
            jitter = random.uniform(0.5, 1.5)
            wait_time *= jitter
        
        print(f"[Rate Limit] {wait_time:.1f}초 후 재시도 (attempt {attempt + 1})")
        time.sleep(wait_time)
        
        # 재시도
        response = requests.post(url, json=payload)
        
        if response.status_code != 429:
            return response
    
    raise Exception(f"Rate limit 재시도 {max_retries}회 실패")


실제 사용

response = requests.post(url, json=payload)

if response.status_code == 429:
    print(f"[Rate Limit] 429 오류 발생: {response.text}")
    response = exponential_backoff_with_jitter()

오류 4: Invalid Request Error - 컨텍스트 길이 초과

# ❌ 잘못된 접근 - 토큰 수 무시
prompt = very_long_text  # 수십만 토큰 가능


✅ 해결 방법: 토큰 수 추정 및 분할

import tiktoken

def count_tokens(text: str, model: str = "claude-sonnet-4-20250514") -> int:
    """토큰 수 추정 (대략적)"""
    # Claude 모델은 토크나이저 사용 필요
    # 여기서는 대략적 계산 (영문 기준 4자 ≈ 1토큰)
    return len(text) // 4

def split_by_tokens(text: str, max_tokens: int = 100000) -> list:
    """토큰 기준 분할"""
    words = text.split()
    chunks = []
    current_chunk = []
    current_tokens = 0
    
    for word in words:
        word_tokens = len(word) // 4 + 1
        
        if current_tokens + word_tokens > max_tokens:
            chunks.append(" ".join(current_chunk))
            current_chunk = [word]
            current_tokens = word_tokens
        else:
            current_chunk.append(word)
            current_tokens += word_tokens
    
    if current_chunk:
        chunks.append(" ".join(current_chunk))
    
    return chunks


실제 사용

MAX_CONTEXT = 180000  # Claude Sonnet 4 컨텍스트 한도

prompt_tokens = count_tokens(prompt)
print(f"프롬프트 토큰 수: {prompt_tokens:,}")

if prompt_tokens > MAX_CONTEXT:
    print(f"컨텍스트 초과! {len(prompt)} 토큰으로 분할 중...")
    chunks = split_by_tokens(prompt, MAX_CONTEXT)
    print(f"{len(chunks)}개 청크로 분할됨")
    
    # 각 청크 처리 후 결과 병합
    results = []
    for i, chunk in enumerate(chunks):
        print(f"청크 {i+1}/{len(chunks)} 처리 중...")
        result = process_chunk(chunk)
        results.append(result)

오류 5: Webhook 서명 검증 실패

# ❌ 잘못된 접근 - 서명 검증 없음
@app.post("/webhook")
async def webhook(request: Request):
    data = await request.json()
    process(data)
    return {"status": "ok"}


✅ 해결 방법: HMAC-SHA256 서명 검증

from fastapi import FastAPI, Request, HTTPException, Header
import hmac
import hashlib

WEBHOOK_SECRET = os.getenv("WEBHOOK_SECRET", "your-secret-key")

@app.post("/webhook")
async def webhook(
    request: Request,
    x_signature: str = Header(None, alias="X-Webhook-Signature")
):
    body = await request.body()
    
    # 서명 검증
    if not x_signature:
        raise HTTPException(status_code=401, detail="Missing signature")
    
    # HMAC-SHA256 서명 계산
    expected_signature = hmac.new(
        WEBHOOK_SECRET.encode(),
        body,
        hashlib.sha256
    ).hexdigest()
    
    expected_header = f"sha256={expected_signature}"
    
    if not hmac.compare_digest(expected_header, x_signature):
        raise HTTPException(status_code=401, detail="Invalid signature")
    
    data = await request.json()
    await process_webhook(data)
    
    return {"status": "ok"}


테스트용 서명 생성 함수

def generate_webhook_signature(payload: dict, secret: str) -> str:
    """Webhook 테스트용 서명 생성"""
    body = json.dumps(payload).encode()

관련 리소스
관련 문서