[Claude 활용 24회 — AI에게 일을 위임하는 법] 21/24화: Agent SDK 제품 만들기 4단계 — 스크립트를 SaaS로 바꾸는 법

15화에서 만든 예약 작업이 석 달째 잘 돌고 있었다. 매일 아침 6시, 내가 지정한 키워드의 시장 동향이 마크다운 파일로 떨어졌다. claude -p 파이프라인에 MCP 서버를 물린 구조. 하루도 빠지지 않았다.

그런데 이 시스템이 소문났다. 옆 팀 동료가 슬쩍 물었다. “그거 나도 쓸 수 있어?” 스크립트를 보내줬다. README도 달았다. 30분 뒤 메시지가 왔다. “claude 명령이 없다는데?” 설치법을 알려줬다. 또 30분. “MCP 서버 설정 파일 어디에 넣어?” 그날 오후를 통째로 날렸다.

일주일 뒤 세 번째 사람한테서 연락이 왔을 때 깨달았다. 이건 내가 옆에 앉아서 환경을 맞춰주는 방식으로는 확장할 수 없다. 스크립트가 아니라 제품이 필요한 순간이었다. Agent SDK 제품 만들기는 바로 이 간극 — “나는 잘 되는데 남은 못 쓰는” 상태에서 시작한다.

이번 회차에서는 Claude의 에이전트 루프를 query() 한 줄로 호출하고, 그걸 웹 서비스로 감싸서, 아무나 브라우저에서 키워드를 입력하면 브리핑을 받을 수 있게 만든다. 15화에서 만든 자동화가 21화에서 상품이 되는 과정이다. PART 6 “하나의 회사 만들기”의 첫 문이 열린다.

“되는 스크립트”에서 “쓸 수 있는 제품”으로 — 세 가지 간극

20년간 백엔드 시스템을 만들면서 깨달은 게 하나 있다. 내부 도구가 제품이 되려면 세 가지 간극을 동시에 건너야 한다.

첫째, 환경 간극. 내 스크립트는 내 PC에 깔린 CLI, 내가 설정한 MCP 서버, 내 디렉터리 구조에 의존한다. 상대방의 환경에는 이 중 아무것도 없다. README에 “사전 요구사항”을 세 쪽이나 적어도, 그 사람의 OS와 셸과 Python 버전이 다르면 처음부터 막힌다. 3화에서 “첫 30분”에 겪는 온보딩 고통을 기억하는가. 그걸 매 사용자마다 반복할 수는 없다.

둘째, 인터페이스 간극. 터미널에서 cron으로 도는 스크립트를 쓸 줄 아는 사람은 개발자뿐이다. 기획자가, 마케터가, 영업팀이 쓰려면 최소한 웹 폼 하나는 있어야 한다. “여기에 키워드 넣고 버튼 누르세요.” 이 한 줄이 스크립트와 제품의 경계선이다.

셋째, 비용 간극. 내가 혼자 쓸 때는 MAX 구독 안에서 무료처럼 느껴진다. 하지만 50명이 동시에 쓰면? 토큰 비용이 발생한다. API 호출 단가가 존재한다. 18화에서 이미 경고했다 — 생산성이 24% 올라도 예산이 터지면 라이선스부터 끊긴다. 제품에는 원가 구조가 있어야 한다.

Agent SDK는 이 세 간극을 동시에 메운다. 환경을 서버 한 대에 고정하고(환경 간극 해소), HTTP API로 인터페이스를 표준화하고(인터페이스 간극 해소), 요청 단위로 토큰 사용량을 추적할 수 있게 한다(비용 간극 해소). 핵심은 딱 하나: query() 함수. Claude Code가 터미널에서 돌리는 에이전트 루프 전체를 내 코드 안에서 프로그래밍 가능한 함수로 호출할 수 있다는 뜻이다.

AI가 코드를 짜주는 시대가 아니다. AI에게 일을 위임하는 시대다. 그리고 그 위임을 남이 돈 내고 쓸 수 있는 형태로 포장하는 것 — 그게 Agent SDK 제품 만들기의 본질이다.

Agent SDK 기본 구조 — query()가 여는 제품화의 문

Agent SDK를 이해하려면 그 아래 깔린 계층부터 보는 게 빠르다. 바닥에서 위로 올라가자.

헤드리스 CLI — 자동화의 시작점

15화에서 예약 작업을 만들 때 사용한 방식이 바로 헤드리스 모드다. claude -p 플래그를 붙이면 Claude Code가 대화형 터미널 대신 파이프 모드(pipe mode)로 동작한다. 표준 입력으로 프롬프트를 밀어넣고, 표준 출력으로 결과를 뽑아낸다.

# 기본 사용법: stdin → claude → stdout
echo '최근 7일간 생성형 AI 시장 동향을 3문단으로 요약해줘' | claude -p

# JSON 스트리밍 출력 — 자동화의 핵심
echo '시장 동향 분석' | claude -p --output-format stream-json

--output-format stream-json을 붙이면 결과가 한 줄짜리 JSON 이벤트 스트림(JSONL)으로 나온다. 각 줄은 {"type": "assistant", "content": [...]} 같은 형태다. 이걸 파싱하면 프로그램이 Claude의 출력을 실시간으로 소비할 수 있다.

Python에서 직접 서브프로세스로 부르면 이렇게 된다:

import asyncio
import json


async def run_claude_raw(prompt: str) -> str:
    """claude -p를 서브프로세스로 직접 호출한다. 가장 낮은 수준의 통합."""
    proc = await asyncio.create_subprocess_exec(
        "claude", "-p",
        "--output-format", "stream-json",
        stdin=asyncio.subprocess.PIPE,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE,
    )
    stdout_bytes, _ = await proc.communicate(
        input=prompt.encode("utf-8")
    )

    result_text = ""
    for line in stdout_bytes.decode("utf-8").splitlines():
        if not line.strip():
            continue
        try:
            event = json.loads(line)
        except json.JSONDecodeError:
            continue
        if event.get("type") == "assistant":
            for block in event.get("content", []):
                if block.get("type") == "text":
                    result_text += block["text"]

    return result_text

동작은 한다. 하지만 이 방식에는 문제가 있다.

• 에러 처리를 직접 구현해야 한다. CLI가 비정상 종료하면?
• 세션 관리를 직접 해야 한다. 이어서 물으려면?
• 도구(Tool) 호출 결과를 파싱하는 로직이 전부 내 몫이다.
• 타임아웃, 재시도, 동시성 제어 — 프로덕션에 필요한 것 전부가 빠져 있다.

그래서 SDK가 존재한다.

query() — 프로그래밍 가능한 에이전트 루프

Anthropic의 Agent SDK는 이 서브프로세스 관리를 캡슐화한 공식 라이브러리다. Python 패키지 이름은 claude-code-sdk.

# 설치
pip install claude-code-sdk

핵심 함수는 query() 하나다. 비동기 제너레이터(async generator)로 동작하며, Claude Code가 에이전트 루프를 돌면서 생성하는 이벤트를 하나씩 내보낸다(yield).

import asyncio
from claude_code_sdk import query, ClaudeCodeOptions


async def basic_example():
    """가장 단순한 query() 호출."""
    options = ClaudeCodeOptions(
        max_turns=5,
        system_prompt=(
            "한국어로 응답하세요. "
            "출처와 날짜를 반드시 포함하세요."
        ),
    )

    full_text = ""
    async for event in query(
        prompt="최근 7일간 '클라우드 인프라' 관련 뉴스 3건을 요약해줘",
        options=options,
    ):
        # event는 Message 객체. content 블록을 순회한다.
        if hasattr(event, "content"):
            for block in event.content:
                if hasattr(block, "text"):
                    full_text += block.text

    print(full_text)


asyncio.run(basic_example())

이것만 놓고 보면 “서브프로세스 직접 호출이랑 뭐가 다르냐”고 물을 수 있다. 차이는 ClaudeCodeOptions에 들어 있다. 제품을 만들 때 필요한 대부분의 제어권이 이 설정 객체에 담겨 있기 때문이다.

options = ClaudeCodeOptions(
    # 에이전트가 도구를 호출하며 반복하는 최대 턴 수
    max_turns=10,

    # 시스템 프롬프트 — 제품의 성격을 결정한다
    system_prompt="당신은 시장 분석가입니다...",

    # 허용 도구 — 최소 권한 원칙 (6화 참고)
    allowed_tools=["Read", "Bash"],

    # 작업 디렉터리 — 에이전트의 파일 접근 범위를 제한
    cwd="/opt/briefing-workspace",

    # MCP 서버 — 제품이 접근하는 데이터 소스 (10화 참고)
    mcp_servers={
        "news": {
            "command": "python",
            "args": ["-m", "news_mcp_server"],
        },
    },

    # 권한 모드 — 프로덕션에서는 반드시 제한
    permission_mode="default",
)

여기서 보이는가? 9화의 Skill, 10화의 MCP 서버, 6화의 권한 제어 — 그동안 배운 것들이 모두 ClaudeCodeOptions의 필드로 들어온다. 개별 기능이 아니라 하나의 제품 설정으로 합쳐지는 순간이다.

스트리밍 — 사용자가 기다리지 않게

제품의 관점에서 스트리밍은 사치가 아니라 필수다. 시장 동향 브리핑이 3,000자짜리 리포트라면, 전체가 생성될 때까지 기다리는 건 보통 20~40초다. 사용자는 빈 화면 앞에서 5초만 지나도 불안해한다. 글자가 한 줄씩 나타나면 같은 40초도 “일하고 있구나” 하고 기다린다.

query()는 태생이 비동기 제너레이터다. 이벤트가 생길 때마다 바로 yield한다. 이걸 그대로 SSE(Server-Sent Events)로 릴레이하면 브라우저에서 실시간 스트리밍을 볼 수 있다.

from claude_code_sdk import query, ClaudeCodeOptions


async def stream_briefing(keyword: str):
    """키워드 기반 브리핑을 스트리밍으로 생성한다."""
    options = ClaudeCodeOptions(
        max_turns=5,
        system_prompt="시장 분석가로서 객관적으로 요약하세요.",
    )

    async for event in query(
        prompt=f"'{keyword}' 관련 최근 7일간 시장 동향을 분석해줘",
        options=options,
    ):
        if hasattr(event, "content"):
            for block in event.content:
                if hasattr(block, "text"):
                    # 텍스트 조각이 생길 때마다 즉시 반환
                    yield block.text

이 제너레이터를 HTTP 응답에 연결하는 부분은 실습 섹션에서 완성한다.

세션 체크포인트 — 대화를 이어가는 제품

15화의 예약 작업은 단발 요청이었다. 키워드를 던지면 리포트가 나온다. 끝. 그런데 제품이 되면 사용자는 이렇게 쓴다:

1. “‘생성형 AI’ 시장 동향 분석해줘” → 리포트 수신
2. “아까 리포트에서 언급한 A사의 투자 규모를 더 파줘” → 같은 맥락에서 추가 질문
3. 다음 날 다시 와서 “어제 분석 이어서 해줘” → 세션 재개

이것을 가능하게 하는 것이 세션 체크포인트다. query()는 호출 시 세션 ID를 반환하고, 다음 호출에서 그 ID를 넘기면 이전 대화를 이어받는다.

from claude_code_sdk import query, ClaudeCodeOptions

# === 1단계: 세션 생성 ===
session_id = None

async for event in query(
    prompt="'생성형 AI' 시장 동향을 분석해줘",
    options=ClaudeCodeOptions(max_turns=5),
):
    # SDK가 세션 ID를 이벤트에 포함해 돌려준다
    if hasattr(event, "session_id") and event.session_id:
        session_id = event.session_id
    # ... 결과 처리 ...


# === 2단계: 같은 세션 이어서 ===
async for event in query(
    prompt="A사의 투자 규모를 더 구체적으로 분석해줘",
    options=ClaudeCodeOptions(
        session_id=session_id,   # 이전 세션을 이어받는다
        max_turns=5,
    ),
):
    # ... 결과 처리 ...
    pass

세션 ID를 데이터베이스에 사용자별로 저장하면 “로그인 → 이전 분석 이어서”라는 제품 경험이 만들어진다. 8화에서 배운 체크포인트 개념이 여기서 제품 기능으로 변환된 것이다.

기본 구조를 훑었다. 이제 이것을 실제로 돌아가는 웹 서비스로 조립한다.

실습 — 15화의 예약 작업을 웹 서비스로 감싸기

15화에서 만든 것: cron → claude -p → 파일 저장. 나만 쓰는 자동화.

21화에서 만들 것: 웹 폼 → FastAPI → Agent SDK query() → 스트리밍 응답. 누구나 쓰는 서비스.

변환의 핵심은 단 하나, 인터페이스를 바꾸는 것이다. 에이전트 로직 자체는 15화와 거의 같다.

프로젝트 뼈대

디렉터리 구조부터 잡는다. 실제로 복사해서 쓸 수 있는 최소 형태다.

briefing-service/
├── pyproject.toml
├── src/
│   └── briefing/
│       ├── __init__.py
│       ├── main.py          # FastAPI 앱 진입점
│       ├── agent.py         # Agent SDK 래퍼
│       ├── models.py        # 요청/응답 스키마
│       └── cost_tracker.py  # 토큰 비용 추적
└── README.md

pyproject.toml의 의존성:

[project]
name = "briefing-service"
version = "0.1.0"
requires-python = ">=3.11"
dependencies = [
    "fastapi>=0.115",
    "uvicorn[standard]>=0.30",
    "claude-code-sdk>=0.1",
    "pydantic>=2.0",
]

요청/응답 스키마 — models.py

Pydantic으로 입출력 형태를 고정한다. 제품은 “무엇이 들어오고 무엇이 나가는가”가 계약이다.

"""briefing/models.py — 요청/응답 데이터 모델."""
from pydantic import BaseModel, Field


class BriefingRequest(BaseModel):
    """브리핑 생성 요청."""
    keyword: str = Field(
        ...,
        min_length=2,
        max_length=100,
        description="분석할 키워드",
        examples=["생성형 AI", "클라우드 인프라"],
    )
    period_days: int = Field(
        default=7,
        ge=1,
        le=30,
        description="분석 기간(일)",
    )
    session_id: str | None = Field(
        default=None,
        description="이전 세션을 이어받을 때 사용",
    )


class BriefingMeta(BaseModel):
    """브리핑 생성 완료 후 반환되는 메타데이터."""
    session_id: str
    input_tokens: int
    output_tokens: int
    estimated_cost_usd: float

주목할 필드는 session_id와 estimated_cost_usd다. 전자는 “이어서 물어볼 수 있는 제품”을, 후자는 “원가를 추적할 수 있는 사업”을 가능하게 한다.

Agent SDK 래퍼 — agent.py

Agent SDK 호출 로직을 한 곳에 모은다. 비즈니스 로직(프롬프트 설계, 도구 설정)과 HTTP 계층을 분리하는 기본 원칙이다.

"""briefing/agent.py — Agent SDK를 감싸는 비즈니스 계층."""
from __future__ import annotations

import json
from dataclasses import dataclass, field
from typing import AsyncIterator

from claude_code_sdk import ClaudeCodeOptions, query


SYSTEM_PROMPT = """\
당신은 시장 동향 분석가입니다.
- 최근 {period_days}일간의 동향을 객관적으로 분석합니다.
- 출처와 날짜를 반드시 포함합니다.
- 분석 결과를 '핵심 요약 → 세부 동향 → 시사점' 순서로 구성합니다.
- 근거 없는 전망은 '추정'임을 명시합니다.
"""


@dataclass
class BriefingResult:
    """스트리밍 중 수집되는 결과 데이터."""
    session_id: str = ""
    input_tokens: int = 0
    output_tokens: int = 0
    chunks: list[str] = field(default_factory=list)


async def generate_briefing(
    keyword: str,
    period_days: int = 7,
    session_id: str | None = None,
) -> AsyncIterator[str | BriefingResult]:
    """
    키워드 기반 브리핑을 생성하고 텍스트 조각을 yield한다.
    마지막에 BriefingResult를 yield해서 메타데이터를 전달한다.
    """
    options = ClaudeCodeOptions(
        max_turns=5,
        system_prompt=SYSTEM_PROMPT.format(period_days=period_days),
        allowed_tools=["Read", "Bash"],
    )

    # 세션 이어받기
    if session_id:
        options.session_id = session_id

    prompt = (
        f"'{keyword}' 관련 최근 {period_days}일간 "
        f"시장 동향을 분석해줘."
    )

    result = BriefingResult()

    async for event in query(prompt=prompt, options=options):
        # 세션 ID 캡처
        if hasattr(event, "session_id") and event.session_id:
            result.session_id = event.session_id

        # 토큰 사용량 캡처
        if hasattr(event, "usage"):
            result.input_tokens += getattr(
                event.usage, "input_tokens", 0
            )
            result.output_tokens += getattr(
                event.usage, "output_tokens", 0
            )

        # 텍스트 조각 스트리밍
        if hasattr(event, "content"):
            for block in event.content:
                if hasattr(block, "text") and block.text:
                    result.chunks.append(block.text)
                    yield block.text   # <-- 이 줄이 실시간 스트리밍의 핵심

    # 스트리밍 완료 후 메타데이터 전달
    yield result

이 함수의 설계 포인트:

• 제너레이터의 이중 역할: 텍스트 조각(str)을 먼저 yield하고, 마지막에 BriefingResult 객체를 yield한다. 호출하는 쪽에서 타입을 확인해 분기한다.
• 세션 ID 자동 캡처: 첫 요청이면 새 세션이 생기고, 기존 ID를 넘기면 이어받는다. 사용자는 이 차이를 의식하지 않는다.
• 토큰 사용량 추적: 비용 계산의 원천 데이터. 나중에 사용자별·요청별로 집계하는 기반이 된다.
• 도구 제한: allowed_tools에 최소한만 넣는다. 6화에서 “프롬프트로 ‘하지 마’는 통제가 아니다”라고 한 것을 코드로 실현하는 부분이다.

스트리밍 응답 릴레이 — main.py

FastAPI 엔드포인트에서 Agent SDK의 스트리밍을 SSE(Server-Sent Events)로 중계한다. 브라우저의 EventSource API가 이걸 직접 소비할 수 있다.

"""briefing/main.py — FastAPI 앱."""
from __future__ import annotations

import json
from contextlib import asynccontextmanager
from typing import AsyncIterator

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse

from briefing.agent import BriefingResult, generate_briefing
from briefing.models import BriefingMeta, BriefingRequest


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
    """앱 시작/종료 시 실행되는 코드."""
    # 시작 시: 필요한 초기화 (DB 연결, 캐시 웜업 등)
    yield
    # 종료 시: 정리


app = FastAPI(
    title="Briefing Service",
    version="0.1.0",
    lifespan=lifespan,
)

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],   # 프로덕션에서는 도메인 제한
    allow_methods=["POST"],
    allow_headers=["*"],
)

# --- 토큰 비용 상수 (모델에 따라 다름, 실제 운영 시 설정 파일로 분리) ---
INPUT_PRICE_PER_1M = 3.0     # USD / 1M input tokens
OUTPUT_PRICE_PER_1M = 15.0   # USD / 1M output tokens


def _estimate_cost(input_tokens: int, output_tokens: int) -> float:
    return (
        (input_tokens / 1_000_000) * INPUT_PRICE_PER_1M
        + (output_tokens / 1_000_000) * OUTPUT_PRICE_PER_1M
    )


@app.post("/api/v1/briefings")
async def create_briefing(req: BriefingRequest) -> StreamingResponse:
    """브리핑 생성 — SSE 스트리밍 응답."""

    async def event_stream() -> AsyncIterator[str]:
        meta: BriefingResult | None = None

        async for chunk in generate_briefing(
            keyword=req.keyword,
            period_days=req.period_days,
            session_id=req.session_id,
        ):
            if isinstance(chunk, str):
                # 텍스트 조각 → SSE data 이벤트
                payload = json.dumps(
                    {"type": "text", "content": chunk},
                    ensure_ascii=False,
                )
                yield f"data: {payload}\n\n"
            elif isinstance(chunk, BriefingResult):
                meta = chunk

        # 스트리밍 완료 → 메타데이터 이벤트
        if meta:
            cost = _estimate_cost(
                meta.input_tokens, meta.output_tokens,
            )
            done_payload = json.dumps(
                {
                    "type": "done",
                    "session_id": meta.session_id,
                    "input_tokens": meta.input_tokens,
                    "output_tokens": meta.output_tokens,
                    "estimated_cost_usd": round(cost, 6),
                },
                ensure_ascii=False,
            )
            yield f"data: {done_payload}\n\n"

        yield "data: [DONE]\n\n"

    return StreamingResponse(
        event_stream(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",   # nginx 뒤에 있을 때 필수
        },
    )


@app.get("/healthz")
async def health() -> dict[str, str]:
    return {"status": "ok"}

몇 가지 포인트를 짚자.

• X-Accel-Buffering: no: nginx 같은 리버스 프록시 뒤에 있으면 SSE가 버퍼링돼서 한 번에 몰려 나온다. 이 헤더를 반드시 넣어야 실시간 스트리밍이 동작한다. 이거 빼먹고 “왜 스트리밍이 안 되지?” 하며 30분 삽질한 적 있다.
• 비용 추정 응답: 스트리밍이 끝나면 마지막 SSE 이벤트에 토큰 사용량과 추정 비용을 실어 보낸다. 프론트엔드에서 “이 분석에 약 $0.04가 사용되었습니다”를 표시할 수 있다.
• 세션 ID 반환: 클라이언트가 이 ID를 저장했다가 다음 요청에 넣으면 대화가 이어진다.

세션 관리와 이력 조회

제품 수준으로 가려면 세션 ID를 데이터베이스에 저장하고, 사용자별로 이력을 조회할 수 있어야 한다. 최소한의 인메모리 구현을 보여주고, 프로덕션에서 어디를 바꾸면 되는지 표시한다.

"""briefing/sessions.py — 세션 이력 관리 (인메모리, 프로덕션은 DB로 교체)."""
from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime, timezone


@dataclass
class SessionRecord:
    session_id: str
    keyword: str
    created_at: datetime = field(
        default_factory=lambda: datetime.now(timezone.utc)
    )
    total_input_tokens: int = 0
    total_output_tokens: int = 0
    total_cost_usd: float = 0.0
    turn_count: int = 0


class SessionStore:
    """
    세션 이력 저장소.
    인메모리 dict → 프로덕션에서는 PostgreSQL / Redis로 교체.
    인터페이스(save / get / list_by_user)만 유지하면 된다.
    """

    def __init__(self) -> None:
        self._records: dict[str, SessionRecord] = {}

    def save(self, record: SessionRecord) -> None:
        self._records[record.session_id] = record

    def get(self, session_id: str) -> SessionRecord | None:
        return self._records.get(session_id)

    def list_recent(self, limit: int = 20) -> list[SessionRecord]:
        all_records = sorted(
            self._records.values(),
            key=lambda r: r.created_at,
            reverse=True,
        )
        return all_records[:limit]


# 앱 전역 인스턴스 (lifespan에서 교체 가능)
session_store = SessionStore()

인메모리라서 서버를 재시작하면 날아간다. 하지만 인터페이스를 먼저 정의해두면 나중에 PostgreSQL이든 Redis든 SessionStore만 교체하면 된다. 이 패턴은 12화에서 플러그인의 스토리지 추상화로 다뤘던 것과 같다.

로컬에서 돌려보기

# 1. 가상환경 생성 + 의존성 설치
python -m venv .venv
.venv\Scripts\activate    # Windows
pip install -e .

# 2. 서버 시작
uvicorn briefing.main:app --host 127.0.0.1 --port 8000 --reload

# 3. 다른 터미널에서 테스트
curl -N -X POST http://127.0.0.1:8000/api/v1/briefings ^
  -H "Content-Type: application/json" ^
  -d "{\"keyword\": \"생성형 AI\", \"period_days\": 7}"

curl -N은 버퍼링을 끄는 플래그다. SSE 스트리밍이 실시간으로 찍힌다. 터미널에 JSON 이벤트가 한 줄씩 흘러가면 성공이다.

이어서 물어보려면 첫 응답의 session_id를 복사해서 다음 요청에 넣는다:

curl -N -X POST http://127.0.0.1:8000/api/v1/briefings ^
  -H "Content-Type: application/json" ^
  -d "{\"keyword\": \"생성형 AI\", \"session_id\": \"세션ID값\"}"

같은 맥락에서 추가 질문이 이어지는 걸 확인할 수 있다.

여기까지가 “스크립트를 웹 서비스로 감싸는” 실습이다. 코드 전체를 합쳐도 200줄이 안 된다. 9화의 Skill, 10화의 MCP 서버, 15화의 예약 작업 — 그동안 쌓은 블록이 있으니 조립만 하면 되는 것이다.

하지만 여기서 끝이 아니다. 이걸 실제로 “팔려면” 반드시 알아야 하는 것이 하나 있다.

원가를 모르고 정액제를 붙이면 그날로 적자

제품의 기술적 구현보다 더 많은 사람이 실패하는 지점이 여기다. 비용 구조.

전통 SaaS의 경험칙은 이렇다: “서버 비용은 사용자 수에 비례해 느리게 오르고, 한계비용은 거의 0에 수렴한다.” 그래서 월 9,900원 정액제가 동작한다. 100명이 쓰든 10,000명이 쓰든 서버 비용은 비슷하니까.

Agent SDK 기반 제품은 완전히 다르다. 요청 하나마다 토큰 비용이 발생한다. 사용자가 많이 쓸수록 비용이 선형으로 — 때로는 초선형으로 — 올라간다. 전통 SaaS의 상식으로 가격을 매기면 첫 달에 적자가 난다.

요청 1건의 진짜 원가

우리가 만든 브리핑 서비스의 비용을 계산해보자.

# 비용 추정기 — 실제 서비스에서 반드시 돌려봐야 하는 계산
INPUT_PRICE_PER_1M = 3.00     # USD / 1M input tokens
OUTPUT_PRICE_PER_1M = 15.00   # USD / 1M output tokens

# 브리핑 서비스 기준 실측 평균값 (20회 실행 평균)
avg_input_tokens = 4_500      # 시스템 프롬프트 + 사용자 입력 + MCP 컨텍스트
avg_output_tokens = 2_000     # 브리핑 리포트 본문

cost_per_request = (
    (avg_input_tokens / 1_000_000) * INPUT_PRICE_PER_1M
    + (avg_output_tokens / 1_000_000) * OUTPUT_PRICE_PER_1M
)

print(f"요청 1건 원가: ${cost_per_request:.4f}")
# → 요청 1건 원가: $0.0435

요청 한 건에 약 4.4센트(약 60원). 싸 보인다.

하지만 에이전트가 도구를 여러 번 호출하면(max_turns=5) 턴마다 토큰이 추가된다. 실측해보면 평균 3턴을 돌고, 최악의 경우 5턴 풀로 돌며 입력 토큰이 12,000까지 올라간다. 최악 케이스의 원가는 1건에 $0.066 — 평균의 1.5배다.

정액제 실패 시나리오

“월 9,900원(약 $7)에 무제한 브리핑”으로 가격을 매겼다고 치자.

# 시나리오: 월 $7 정액제, 50명 유저

users = 50
monthly_revenue = 7.0 * users           # $350

# 사용자별 하루 평균 요청: 3건 (아침 브리핑 + 추가 질문 2건)
requests_per_day = 3
monthly_requests = users * requests_per_day * 30   # 4,500건

avg_cost = 0.0435   # 평균 요청 원가
monthly_cost = monthly_requests * avg_cost          # $195.75

# 서버 비용 (클라우드 VM + DB)
infra_cost = 30.0

total_cost = monthly_cost + infra_cost              # $225.75
profit = monthly_revenue - total_cost               # $124.25

print(f"월 매출: ${monthly_revenue:.0f}")
print(f"월 토큰 원가: ${monthly_cost:.0f}")
print(f"월 인프라: ${infra_cost:.0f}")
print(f"월 이익: ${profit:.0f}")
# → 이익 $124. 괜찮아 보인다.

$124 이익. 나쁘지 않아 보인다. 하지만 이건 평균의 함정이다.

# 현실 시나리오: 파워 유저 등장

# 50명 중 5명이 하루 15건씩 쓴다 (리서치 용도)
power_user_requests = 5 * 15 * 30                   # 2,250건
# 나머지 45명은 하루 2건
normal_user_requests = 45 * 2 * 30                  # 2,700건

total_requests = power_user_requests + normal_user_requests  # 4,950건

# 파워 유저는 복잡한 질문이 많아 평균 원가도 높다
power_cost = 2_250 * 0.066                          # $148.50
normal_cost = 2_700 * 0.0435                        # $117.45

total_token_cost = power_cost + normal_cost          # $265.95
total_cost = total_token_cost + infra_cost           # $295.95
profit = monthly_revenue - total_cost                # $54.05

print(f"현실 월 이익: ${profit:.0f}")
# → $54. 한 명 더 파워유저 되면 적자 전환.

파워 유저 5명이 원가의 56%를 차지한다. 이런 사용자가 한두 명 더 생기면 바로 적자다. 18화에서 마이크로소프트 리서치의 극단 사례를 언급했다 — 월 140만 달러가 한 조직에서 나갔다. 규모만 다를 뿐 구조는 같다.

요금제 설계 3원칙

Agent SDK 기반 제품의 가격을 매길 때 지켜야 할 원칙이 셋 있다.

원칙 1: 원가의 3배를 바닥으로 잡는다.

요청 1건 원가 $0.04라면, 사용자에게 1건당 최소 $0.12를 부과하는 구조를 설계해야 한다. 이 마진에 인프라 비용, 내 인건비, 예상치 못한 비용 폭증 버퍼가 포함된다. 정액제를 쓰더라도 내부적으로는 요청 수 × 3배 단가로 환산해서 한도를 설정한다.

원칙 2: 사용량 상한을 반드시 건다.

코드로 보면 이렇다:

# 사용자별 월간 토큰 상한 설정
MAX_MONTHLY_TOKENS_PER_USER = 500_000   # 입출력 합산
MAX_MONTHLY_COST_PER_USER = 5.0         # USD

async def check_usage_limit(user_id: str) -> bool:
    """사용량 초과 여부를 확인한다. True면 차단."""
    usage = await get_monthly_usage(user_id)  # DB 조회
    if usage.total_tokens > MAX_MONTHLY_TOKENS_PER_USER:
        return True
    if usage.total_cost_usd > MAX_MONTHLY_COST_PER_USER:
        return True
    return False

상한을 안 걸면 파워 유저 한 명이 사업을 말아먹을 수 있다. “무제한”이라는 단어를 쓰고 싶으면, 최소한 “공정 사용 정책(Fair Use Policy)” 뒤에 숫자 기준을 숨겨놓아야 한다.

원칙 3: 실측 데이터 없이 가격표를 만들지 않는다.

최소 2주, 최소 20명의 실 사용 데이터를 모은 뒤에 가격을 결정한다. 내가 실측한 평균 4,500 입력 토큰이 실제 사용자 환경에서는 8,000이 될 수도 있다. 시스템 프롬프트를 약간만 바꿔도 토큰 수가 2배로 뛸 수 있기 때문이다. Claude Code의 공식 저장소에서도 토큰 사용량은 작업 복잡도에 따라 크게 변동한다고 안내한다.

이 세 원칙이 없으면 Agent SDK 제품 만들기는 기술적으로 성공하고 사업적으로 실패하는, 개발자가 가장 많이 걸리는 함정에 빠진다.

이번 회차의 수익화 지점

오늘 만든 코드 200줄을 기준으로, 바로 상품화할 수 있는 형태가 두 가지다.

하나, Agent SDK 래퍼를 템플릿으로 판다. 브리핑 서비스의 뼈대 — FastAPI + Agent SDK + SSE 스트리밍 + 비용 추적 — 를 보일러플레이트 템플릿으로 묶는다. 12화에서 만든 플러그인 패키징 방식과 같다. “여기에 시스템 프롬프트만 바꾸면 법률 문서 요약 서비스가 되고, 고객 문의 분류 서비스가 된다.” 구조를 파는 것이다.

둘, 특정 도메인에 특화된 마이크로 SaaS로 운영한다. “매일 아침 특정 산업 동향을 정리해주는 구독 서비스”는 실제로 돈을 내는 사람이 있는 시장이다. 뉴스레터 자동화, 경쟁사 모니터링, 키워드 트래킹 — 이런 것들이 오늘 만든 코드 위에서 돌아간다. 핵심은 원가 3원칙을 지키면서 크레딧 기반 요금제(월 100크레딧, 1크레딧 = 1브리핑)로 설계하는 것이다.

위임할 줄 아는 사람이 하나의 회사가 된다. Agent SDK는 그 위임을 상품으로 만드는 마지막 기술 계층이다.

다음 22화 예고: 만들었다. 팔 수 있다. 그런데 부업으로 이걸 팔아도 되는 걸까? 필명 브랜딩, 회사 자산과의 분리, 사업자 등록 — “회사 노트북으로 잠깐만”이 모든 걸 무너뜨리는 이야기.

관련 회차: 15화(예약 작업 원본), 12화(플러그인 패키징), 18화(비용 거버넌스)

---

자주 묻는 질문

Agent SDK의 query() 함수는 무엇이고 어떻게 제품화에 활용하나요?

query() 함수는 Claude Code가 터미널에서 실행하는 에이전트 루프 전체를 프로그래밍 가능한 단일 함수로 호출할 수 있게 해주는 Agent SDK의 핵심입니다. 이를 통해 CLI 스크립트를 HTTP API로 감싸 웹 서비스로 만들 수 있으며, 누구나 브라우저에서 키워드를 입력하면 AI 브리핑을 받을 수 있는 SaaS 형태의 제품으로 전환할 수 있습니다.

개인용 AI 스크립트를 다른 사람도 쓸 수 있는 제품으로 바꾸려면 어떤 문제를 해결해야 하나요?

세 가지 간극을 동시에 건너야 합니다. 첫째는 환경 간극으로, 내 PC의 CLI·MCP 서버·디렉터리 구조에 의존하는 스크립트를 서버 한 대에 고정해야 합니다. 둘째는 인터페이스 간극으로, 터미널 대신 웹 폼 같은 비개발자용 UI가 필요합니다. 셋째는 비용 간극으로, 다수가 사용할 때 토큰 비용을 요청 단위로 추적할 수 있는 원가 구조가 있어야 합니다.

claude -p 헤드리스 모드와 Agent SDK의 차이점은 무엇인가요?

claude -p 헤드리스 모드는 표준 입력으로 프롬프트를 넣고 표준 출력으로 결과를 받는 파이프 방식으로, 개인 자동화(예: cron 예약 작업)에 적합합니다. Agent SDK는 이 헤드리스 CLI 위에 구축된 상위 계층으로, query() 함수를 통해 에이전트 루프를 코드 안에서 직접 호출하고 HTTP API로 감싸 여러 사용자가 접근 가능한 웹 서비스 형태의 제품으로 만들 수 있게 해줍니다.