웹훅으로 전화 자동화 구현하기: Python + FastAPI 실전 코드

AI 전화 에이전트가 전화를 받고 걸 수 있게 됐습니다. 그런데 진짜 자동화는 여기서 시작입니다. 전화가 끝나면 CRM에 기록을 남기고, 주문이 들어오면 확인 전화를 걸고, 부재중이면 슬랙으로 알림을 보내는 — 이런 연결을 만들어야 합니다.

n8n, Make, Zapier 같은 노코드 도구로 연결하는 방법도 있지만, 이 글에서는 개발자가 직접 Python으로 구현하는 방법을 다룹니다.

웹훅이란

ClawOps에서 이벤트가 발생하면 지정된 URL로 HTTP POST 요청을 보내는 것입니다:

전화 이벤트 발생 → ClawOps 서버 → HTTP POST → 내 서버 → 비즈니스 로직

주요 이벤트:

이벤트	설명	활용
call.initiated	전화 발신 시작	발신 로그 기록
call.ringing	벨 울리는 중	—
call.answered	상대방이 받음	통화 시작 시간 기록
call.completed	통화 종료	CRM 기록, 후처리
message.received	문자 수신	자동 응답

기본 웹훅 서버 (FastAPI)

pip install fastapi uvicorn clawops

from fastapi import FastAPI, Request
from datetime import datetime

app = FastAPI()

@app.post("/webhook/call-status")
async def call_status(request: Request):
    """통화 상태 변경 시 호출됨"""
    data = await request.json()

    call_id = data.get("call_id")
    status = data.get("status")
    direction = data.get("direction")
    from_number = data.get("from")
    to_number = data.get("to")
    duration = data.get("duration")

    print(f"[{datetime.now()}] {call_id}: {status} ({direction})")

    if status == "completed":
        print(f"  통화 종료 — {from_number} → {to_number}, {duration}초")
        # 여기서 비즈니스 로직 실행

    return {"status": "ok"}

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

시나리오 1: 부재중 전화 슬랙 알림

놓친 전화를 슬랙으로 알려줍니다:

import httpx
from fastapi import FastAPI, Request

app = FastAPI()

SLACK_WEBHOOK = "https://hooks.slack.com/services/T.../B.../..."

@app.post("/webhook/missed-call")
async def missed_call(request: Request):
    data = await request.json()

    status = data.get("status")
    direction = data.get("direction")

    # 인바운드 + 미응답인 경우
    if direction == "inbound" and status == "completed":
        duration = data.get("duration", 0)
        if duration == 0:  # 통화 시간 0초 = 부재중
            caller = data.get("from")
            called_at = data.get("created_at")

            # 슬랙 알림 발송
            async with httpx.AsyncClient() as http:
                await http.post(SLACK_WEBHOOK, json={
                    "text": f"📞 부재중 전화\n"
                            f"발신자: {caller}\n"
                            f"시각: {called_at}\n"
                            f"콜백이 필요합니다."
                })

    return {"status": "ok"}

시나리오 2: 주문 접수 → 자동 확인 전화

쇼핑몰에서 주문이 들어오면 자동으로 확인 전화를 걸어줍니다:

from fastapi import FastAPI, Request
from clawops.agent import ClawOpsAgent, OpenAIRealtime
import asyncio

app = FastAPI()

@app.post("/webhook/new-order")
async def new_order(request: Request):
    """쇼핑몰 주문 웹훅 수신"""
    order = await request.json()

    customer_phone = order["customer_phone"]
    customer_name = order["customer_name"]
    items = order["items"]
    total = order["total_price"]

    # AI가 확인 전화 발신
    agent = ClawOpsAgent(
        from_="07052351234",
        session=OpenAIRealtime(
            system_prompt=f"""{customer_name}님에게 주문 확인 전화를 합니다.

주문 내용:
- 상품: {', '.join(items)}
- 결제 금액: {total:,}원

할 일:
1. "{customer_name}님, 주문해주셔서 감사합니다"로 시작
2. 주문 내용을 읽어줍니다
3. "배송지 확인을 위해 주소를 말씀해주세요"
4. 주소 확인 후 "감사합니다, 내일 오전 중 발송됩니다"
5. 통화 종료""",
            voice="marin",
            language="ko",
        ),
    )

    async def make_call():
        session = await agent.call(customer_phone, timeout=30)
        await session.wait()
        await agent.disconnect()

    asyncio.create_task(make_call())

    return {"status": "call_initiated"}

시나리오 3: 통화 종료 → CRM 자동 기록

통화가 끝나면 트랜스크립트를 분석해서 CRM에 자동 기록합니다:

from fastapi import FastAPI, Request
from clawops import ClawOps
import httpx

app = FastAPI()
client = ClawOps()

CRM_API = "https://api.my-crm.com"
CRM_TOKEN = "crm-api-token"

@app.post("/webhook/call-completed")
async def call_completed(request: Request):
    data = await request.json()

    if data.get("status") != "completed":
        return {"status": "skipped"}

    call_id = data.get("call_id")
    caller = data.get("from")
    duration = data.get("duration")

    # 트랜스크립트 조회
    transcript = client.calls.get_transcript(call_id)

    if transcript.status == "completed":
        # 전체 대화 텍스트 추출
        conversation = "\n".join(
            f"[{seg.speaker}] {seg.text}"
            for seg in (transcript.segments or [])
        )

        # 통화 요약 조회
        summary = client.calls.get_summary(call_id)
        summary_text = ""
        if summary.status == "completed" and summary.result_json:
            summary_text = summary.result_json.get("coreSummary", "")

        # CRM에 기록
        async with httpx.AsyncClient() as http:
            await http.post(
                f"{CRM_API}/activities",
                headers={"Authorization": f"Bearer {CRM_TOKEN}"},
                json={
                    "type": "call",
                    "phone": caller,
                    "duration": duration,
                    "summary": summary_text,
                    "transcript": conversation,
                    "call_id": call_id,
                },
            )

    return {"status": "recorded"}

시나리오 4: 예약 리마인더 (문자 + 전화)

예약 시간 전에 문자를 보내고, 확인이 없으면 전화를 걸어줍니다:

from clawops import ClawOps
import time

client = ClawOps()

def send_reminder(appointment):
    """예약 1시간 전 실행"""
    name = appointment["name"]
    phone = appointment["phone"]
    time_str = appointment["time"]

    # 1단계: 문자 발송
    msg = client.messages.create(
        to=phone,
        from_="07052351234",
        body=f"[강남내과] {name}님, 오늘 {time_str} 예약 안내드립니다. "
             f"방문이 어려우시면 070-5235-1234로 연락 주세요.",
    )
    print(f"리마인더 문자 발송: {msg.message_id}")

    # 2단계: 10분 후 확인 없으면 전화 발신
    time.sleep(600)

    # 전화로 확인
    call = client.calls.create(
        to=phone,
        from_="07052351234",
        url=f"https://my-server.com/reminder-call?name={name}&time={time_str}",
    )
    print(f"리마인더 전화 발신: {call.call_id}")

웹훅 보안: 서명 검증

프로덕션에서는 웹훅 요청이 정말 ClawOps에서 온 것인지 확인해야 합니다. 서명 검증으로 위조 요청을 방지합니다:

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

app = FastAPI()

WEBHOOK_SECRET = "whsec_..."

def verify_signature(payload: bytes, signature: str) -> bool:
    expected = hmac.new(
        WEBHOOK_SECRET.encode(),
        payload,
        hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(f"sha256={expected}", signature)

@app.post("/webhook/secure")
async def secure_webhook(request: Request):
    body = await request.body()
    signature = request.headers.get("X-ClawOps-Signature", "")

    if not verify_signature(body, signature):
        raise HTTPException(status_code=401, detail="Invalid signature")

    data = await request.json()
    # 안전하게 처리
    return {"status": "ok"}

웹훅 URL 등록

번호별 웹훅

from clawops import ClawOps

client = ClawOps()

# 번호 생성 시 웹훅 등록
number = client.numbers.create(
    webhook_url="https://my-server.com/webhook/voice"
)

# 기존 번호 웹훅 변경
client.numbers.update(
    "07052351234",
    webhook_url="https://my-server.com/webhook/voice-v2",
)

통화별 상태 콜백

call = client.calls.create(
    to="01012345678",
    from_="07052351234",
    url="https://my-server.com/voice",
    status_callback="https://my-server.com/webhook/call-status",
    status_callback_event="initiated ringing answered completed",
)

배포 체크리스트

프로덕션에 배포할 때:

• HTTPS 적용 (웹훅은 HTTPS만 지원)
• 서명 검증 활성화
• 타임아웃 설정 (웹훅 응답은 5초 이내 권장)
• 에러 핸들링 및 로깅
• 웹훅 실패 시 재시도 로직 (ClawOps가 자동 재시도)
• 중복 처리 방지 (멱등성)

다음 단계

• 웹훅 공식 문서 — 이벤트 전체 목록, 서명 검증 상세
• VoiceML 가이드 — 통화 흐름 XML 제어
• n8n/Make/Zapier 연동 — 노코드로 연결하기
• Python SDK 입문 — SDK 전체 개요
• AI 전화 에이전트 만들기 — AI 에이전트 구축 가이드

---

웹훅 하나로 전화가 비즈니스 시스템과 연결됩니다. 지금 시작하세요.