AI Agents · 강의

에이전트를 웹훅에 연결하기

웹훅 이벤트를 수신하고 이에 따라 에이전트 작업 흐름을 실행합니다.

레슨 2/413개 단계

에이전트를 웹훅에 연결하기은(는) CoddyKit의 무료 AI Agents 강의입니다. 이것은 4개 중 2번째 강의입니다. 아래에서 전체 강의를 무료로 읽을 수 있으며, 내장 코드 에디터와 24/7 AI 튜터와 함께 브라우저에서 직접 실습할 수 있습니다. 이 강의는 AI Agents 학습 경로의 일부이며, 진행 상황이 웹과 CoddyKit 앱에 동기화됩니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.

웹훅이란 무엇입니까

웹훅은 HTTP 콜백입니다. 외부 서비스에서 이벤트가 발생하면 이벤트 데이터가 포함된 POST 요청을 사용자의 엔드포인트로 보냅니다. 에이전트는 페이로드를 처리하고 작업을 수행합니다.

웹훅은 이벤트가 발생하는 즉시 도착하는 푸시 기반 방식인 반면, 폴링은 반복해서 확인하는 방식입니다.

FastAPI 웹훅 엔드포인트

FastAPI를 사용하면 웹훅 수신기를 쉽게 만들 수 있습니다. POST 경로를 정의하고 JSON 본문을 분석한 다음 에이전트 로직으로 넘기십시오.

from fastapi import FastAPI, Request
from pydantic import BaseModel

app = FastAPI()

class WebhookPayload(BaseModel):
    event: str
    data: dict

@app.post('/webhook')
async def receive_webhook(payload: WebhookPayload):
    print(f'Received event: {payload.event}')
    print(f'Data: {payload.data}')
    
    # Route to the right agent handler
    if payload.event == 'email.received':
        await handle_email_event(payload.data)
    elif payload.event == 'file.uploaded':
        await handle_file_event(payload.data)
    
    return {'status': 'accepted'}

async def handle_email_event(data: dict):
    print(f'Processing email from: {data.get("from")}')

async def handle_file_event(data: dict):
    print(f'Processing file: {data.get("filename")}')

웹훅 서명 검증

웹훅 요청이 예상한 발신자에게서 왔는지 항상 검증하십시오. 대부분의 서비스는 공유 비밀을 사용하여 HMAC-SHA256으로 페이로드에 서명합니다. 서명이 유효하지 않은 요청은 거부하십시오.

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

app = FastAPI()
WEBHOOK_SECRET = 'your-webhook-secret-here'

def verify_signature(payload_bytes: bytes, signature_header: str) -> bool:
    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        payload_bytes,
        hashlib.sha256
    ).hexdigest()
    received = signature_header.replace('sha256=', '')
    return hmac.compare_digest(expected, received)

@app.post('/webhook/verified')
async def verified_webhook(request: Request):
    payload_bytes = await request.body()
    signature = request.headers.get('X-Signature', '')
    
    if not verify_signature(payload_bytes, signature):
        raise HTTPException(status_code=401, detail='Invalid signature')
    
    # Safe to process
    import json
    data = json.loads(payload_bytes)
    return {'status': 'verified', 'event': data.get('event')}

멱등성 키

외부 서비스는 실패한 웹훅 전달을 자주 재시도합니다. 멱등성 키는 각 이벤트와 함께 전송되는 고유한 ID입니다. 처리한 키를 저장하고 중복 항목은 건너뛰십시오.

from fastapi import FastAPI, Request, HTTPException
import redis
import json

app = FastAPI()
r = redis.Redis(host='localhost', port=6379, decode_responses=True)

@app.post('/webhook/idempotent')
async def idempotent_webhook(request: Request):
    payload = await request.json()
    
    # Extract idempotency key from header or payload
    idempotency_key = request.headers.get('Idempotency-Key') or payload.get('event_id')
    
    if not idempotency_key:
        raise HTTPException(status_code=400, detail='Missing idempotency key')
    
    redis_key = f'webhook:processed:{idempotency_key}'
    
    # Check if already processed
    if r.exists(redis_key):
        print(f'Duplicate event {idempotency_key}, skipping')
        return {'status': 'duplicate', 'idempotency_key': idempotency_key}
    
    # Process event
    # ... agent logic here ...
    
    # Mark as processed (expire after 24h)
    r.setex(redis_key, 86400, '1')
    return {'status': 'processed', 'idempotency_key': idempotency_key}

재시도 중복 제거 전략

멱등성 키 외에도 중복 제거 기간을 고려하십시오. 짧은 기간 안에 동일한 이벤트 내용을 받았다면 재시도일 가능성이 높습니다. 이벤트 해시를 비교하여 재시도를 감지하고 삭제하십시오.

import hashlib
import json
from datetime import datetime

# In-memory store; use Redis in production
recent_hashes = {}
DEDUP_WINDOW_SECONDS = 300  # 5 minutes

def is_duplicate(payload: dict) -> bool:
    # Hash the event content
    content = json.dumps(payload, sort_keys=True)
    event_hash = hashlib.md5(content.encode()).hexdigest()
    
    now = datetime.utcnow().timestamp()
    
    # Clean up old entries
    expired = [h for h, ts in recent_hashes.items() if now - ts > DEDUP_WINDOW_SECONDS]
    for h in expired:
        del recent_hashes[h]
    
    if event_hash in recent_hashes:
        return True
    
    recent_hashes[event_hash] = now
    return False

# Test
payload = {'event': 'payment.completed', 'amount': 100}
print('First:', is_duplicate(payload))   # False
print('Second:', is_duplicate(payload))  # True (duplicate)

에이전트 비동기 실행

웹훅 처리기는 빠르게 응답하고(5초 이내) 에이전트 로직은 백그라운드에서 처리해야 합니다. 시간 초과를 방지하려면 FastAPI의 BackgroundTasks를 사용하십시오.

from fastapi import FastAPI, BackgroundTasks
import asyncio

app = FastAPI()

async def run_agent_job(event: str, data: dict):
    print(f'Agent starting for event: {event}')
    await asyncio.sleep(2)  # Simulate LLM call
    print(f'Agent finished for event: {event}')

@app.post('/webhook/async')
async def async_webhook(request_data: dict, background_tasks: BackgroundTasks):
    event = request_data.get('event', 'unknown')
    data = request_data.get('data', {})
    
    # Respond immediately
    background_tasks.add_task(run_agent_job, event, data)
    
    return {'status': 'accepted', 'message': 'Processing in background'}

복잡한 페이로드 분석

서비스마다 서로 다른 페이로드 형식을 보냅니다. 각 서비스에 전용 분석 함수를 작성하여 에이전트가 항상 정규화된 이벤트 객체를 받도록 하십시오.

from dataclasses import dataclass
from typing import Optional

@dataclass
class NormalizedEvent:
    event_type: str
    source: str
    resource_id: str
    metadata: dict

def parse_github_webhook(payload: dict) -> NormalizedEvent:
    return NormalizedEvent(
        event_type='github.' + payload.get('action', 'unknown'),
        source='github',
        resource_id=str(payload.get('repository', {}).get('id', '')),
        metadata={
            'repo': payload.get('repository', {}).get('full_name'),
            'sender': payload.get('sender', {}).get('login')
        }
    )

def parse_stripe_webhook(payload: dict) -> NormalizedEvent:
    return NormalizedEvent(
        event_type=payload.get('type', 'unknown'),
        source='stripe',
        resource_id=payload.get('id', ''),
        metadata={'amount': payload.get('data', {}).get('object', {}).get('amount')}
    )

# Usage
github_payload = {'action': 'opened', 'repository': {'id': 123, 'full_name': 'user/repo'}, 'sender': {'login': 'alice'}}
event = parse_github_webhook(github_payload)
print(event)

웹훅 응답 코드가 중요한 이유

올바른 HTTP 상태를 반환하십시오. 2xx는 발신자에게 웹훅이 수락되었음을 알립니다. 4xx는 클라이언트 오류를 의미합니다(잘못된 페이로드). 5xx 또는 시간 초과가 발생하면 발신자가 재시도합니다.

200: 수락 및 처리 완료
202: 비동기 처리를 위해 수락
400: 잘못된 요청(필드 누락)
401: 잘못된 서명

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse

app = FastAPI()

@app.post('/webhook/proper-responses')
async def proper_webhook(request: Request):
    try:
        payload = await request.json()
    except Exception:
        raise HTTPException(status_code=400, detail='Invalid JSON body')
    
    required_fields = ['event', 'data']
    for field in required_fields:
        if field not in payload:
            raise HTTPException(status_code=400, detail=f'Missing field: {field}')
    
    event = payload['event']
    known_events = ['email.received', 'file.uploaded', 'payment.completed']
    
    if event not in known_events:
        # Acknowledge unknown events gracefully - do not retry
        return JSONResponse(status_code=200, content={'status': 'ignored', 'reason': 'unknown event'})
    
    # Start background processing
    return JSONResponse(status_code=202, content={'status': 'accepted'})

로컬에서 웹훅 테스트하기

ngrok을 사용하면 로컬 서버를 인터넷에 공개하여 테스트할 수 있습니다. ngrok http 8000을 실행하면 로컬 FastAPI 앱으로 연결되는 공개 URL을 얻을 수 있습니다.

# Start your FastAPI app
# uvicorn main:app --reload --port 8000

# In another terminal, start ngrok:
# ngrok http 8000
# You get: https://abc123.ngrok.io

# Now configure your webhook in Stripe/GitHub/etc. to:
# https://abc123.ngrok.io/webhook

# Test with curl:
import subprocess

def test_webhook_locally():
    test_payload = '{"event": "email.received", "data": {"from": "test@example.com"}}'
    # In real usage you would run this in terminal:
    # curl -X POST http://localhost:8000/webhook \
    #   -H 'Content-Type: application/json' \
    #   -d '{"event": "email.received", "data": {"from": "test@example.com"}}'
    print('Test payload:', test_payload)
    print('Send to: http://localhost:8000/webhook')

test_webhook_locally()

웹훅 이벤트 로깅

수신되는 모든 웹훅에 타임스탬프, 소스, 이벤트 유형, 처리 결과를 기록하십시오. 이러한 감사 추적은 누락된 이벤트나 중복 처리 문제를 디버깅하는 데 필수적입니다.

import logging
import json
from datetime import datetime
import sys

logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s %(levelname)s %(message)s',
    stream=sys.stdout
)
logger = logging.getLogger('webhook')

def log_webhook_event(event_id: str, event_type: str, source: str, status: str, details: dict = None):
    logger.info(json.dumps({
        'timestamp': datetime.utcnow().isoformat(),
        'event_id': event_id,
        'event_type': event_type,
        'source': source,
        'status': status,
        'details': details or {}
    }))

# Usage in webhook handler
log_webhook_event(
    event_id='evt_123',
    event_type='email.received',
    source='gmail',
    status='processed',
    details={'from': 'user@example.com', 'action_taken': 'reply_sent'}
)

수신 웹훅 요청률 제한

요청률 제한을 사용하여 웹훅 엔드포인트가 과도한 요청으로 압도되지 않도록 보호하십시오. slowapi 라이브러리를 사용하면 최소한의 코드로 FastAPI에 요청률 제한을 추가할 수 있습니다.

from fastapi import FastAPI, Request
from slowapi import Limiter, _rate_limit_exceeded_handler
from slowapi.util import get_remote_address
from slowapi.errors import RateLimitExceeded

limiter = Limiter(key_func=get_remote_address)
app = FastAPI()
app.state.limiter = limiter
app.add_exception_handler(RateLimitExceeded, _rate_limit_exceeded_handler)

@app.post('/webhook/limited')
@limiter.limit('100/minute')
async def rate_limited_webhook(request: Request):
    payload = await request.json()
    return {'status': 'accepted', 'event': payload.get('event')}

지식 확인: 웹훅

에이전트를 위한 웹훅 모범 사례에 대한 이해도를 테스트해 보십시오.

인터랙티브 레슨에서 답해보세요

프로덕션 환경의 웹훅

프로덕션 환경에서는 서명 검증, 멱등성 키, 백그라운드 처리, 구조화된 로깅, 요청률 제한을 모두 결합하십시오. TLS 종료와 추가 보호를 위해 nginx와 같은 역방향 프록시 뒤에 배포하십시오.

무료로 시작

AI 튜터와 함께 AI Agents을(를) 배우세요 — 무료

브라우저에서 실제 코드를 작성하고 실행하며, 24/7 AI 튜터로부터 즉각적인 도움을 받고, 웹이나 앱에서 중단한 부분부터 계속 학습하세요.

코스: 60
레슨: 239

무료로 학습 시작 전체 코스 보기

자주 묻는 질문

“에이전트를 웹훅에 연결하기” 강의는 무료인가요?

네 — “에이전트를 웹훅에 연결하기” 전체 내용을 이 웹사이트에서 무료로 읽을 수 있습니다. 인터랙티브하게 실습하려면(내장 코드 에디터와 24/7 AI 튜터), CoddyKit PRO로 업그레이드하면 AI Agents 강의 전체를 잠금 해제할 수 있습니다. AI Agents 강의에는 총 4개의 강의가 포함되어 있습니다.

“에이전트를 웹훅에 연결하기”에서 뭘 배우나요?

웹훅 이벤트를 수신하고 이에 따라 에이전트 작업 흐름을 실행합니다. 브라우저에서 직접 실행하는 실습 코드로 AI Agents을(를) 배우며, 24/7 AI 튜터가 강의를 진행하면서 질문에 답변해줍니다.

AI Agents을(를) 시작하는 데 경험이 필요한가요?

사전 경험은 필요하지 않습니다. CoddyKit의 AI Agents은(는) 초급자부터 고급 학습자까지를 위해 구성되어 있으므로, 여기서 시작하거나 처음부터 시작할 수 있으며 자신의 속도대로 진행할 수 있습니다. 이것은 4개 중 2번째 강의입니다.

“에이전트를 웹훅에 연결하기” 강의는 얼마나 걸리나요?

대부분의 CoddyKit 강의는 약 5~10분이 소요됩니다. 각 강의는 간결하고 인터랙티브하여 꾸준한 진행이 가능하며, 웹과 앱에서 중단한 부분부터 바로 시작할 수 있습니다.

이 AI Agents 강의에서 코드를 작성하고 실행할 수 있나요?

네. 모든 AI Agents 강의에는 내장 코드 에디터가 포함되어 있으므로, 브라우저에서 바로 실제 코드를 작성하고 실행한 후 즉시 AI 피드백을 받을 수 있습니다 — 로컬 설정이 필요 없습니다.

이 강의의 모든 강의

← AI Agents(으)로 돌아가기