FastAPI로 CLI를 REST API로 확장하기

CLI의 한계

PromoEngine은 처음에 CLI로 만들었다. 잘 동작했지만, 두 가지 한계가 생겼다.

1. 비개발자가 사용할 수 없다: 마케팅 담당자에게 “터미널에서 promo generate 치세요”라고 할 수 없다
2. 다른 서비스에서 호출할 수 없다: 웹 앱이나 슬랙 봇에서 콘텐츠를 생성하려면 API가 필요하다

CLI의 핵심 로직을 그대로 두고, FastAPI로 HTTP 인터페이스를 씌우기로 했다.

구조 분리

먼저 CLI 로직과 비즈니스 로직을 분리한다.

promoengine/
├── core/
│   ├── generator.py    # 비즈니스 로직 (채널별 생성)
│   ├── templates.py    # 프롬프트 템플릿
│   └── storage.py      # SQLite 저장
├── cli/
│   └── main.py         # argparse CLI
└── api/
    └── main.py         # FastAPI 서버

core/는 CLI와 API 모두에서 공유한다. 이 구조면 CLI를 건드리지 않고 API를 추가할 수 있다.

FastAPI 엔드포인트

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

from promoengine.core.generator import generate_all, generate_channel

app = FastAPI(title="PromoEngine API", version="0.1.0")

class GenerateRequest(BaseModel):
    content: str
    channels: list[str] = ['reddit', 'twitter', 'blog', 'email', 'producthunt']
    language: str = 'ko'

class GenerateResponse(BaseModel):
    results: dict[str, str]
    total_channels: int

@app.post("/api/generate", response_model=GenerateResponse)
async def api_generate(req: GenerateRequest):
    try:
        results = await generate_all(
            content=req.content,
            channels=req.channels,
            language=req.language,
        )
        return GenerateResponse(
            results=results,
            total_channels=len(results),
        )
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))

Pydantic 모델로 요청/응답을 정의하면, 자동으로 입력 검증과 API 문서(Swagger)가 생성된다.

비동기 처리

Claude API 호출은 시간이 걸린다. 5개 채널을 모두 생성하면 10~30초. 동기 처리하면 다른 요청이 블로킹된다.

from fastapi import BackgroundTasks

class TaskResponse(BaseModel):
    task_id: str
    status: str

@app.post("/api/generate/async", response_model=TaskResponse)
async def api_generate_async(
    req: GenerateRequest,
    background_tasks: BackgroundTasks,
):
    task_id = str(uuid.uuid4())
    background_tasks.add_task(
        run_generation, task_id, req.content, req.channels
    )
    return TaskResponse(task_id=task_id, status="processing")

@app.get("/api/tasks/{task_id}")
async def get_task(task_id: str):
    result = storage.get_task(task_id)
    if not result:
        raise HTTPException(status_code=404, detail="Task not found")
    return result

긴 작업은 백그라운드로 처리하고, 클라이언트가 task_id로 폴링한다.

인증

API를 외부에 노출하면 인증이 필요하다. 간단하게 API 키 방식을 사용한다.

from fastapi import Depends, Security
from fastapi.security import APIKeyHeader

api_key_header = APIKeyHeader(name="X-API-Key")

async def verify_api_key(api_key: str = Security(api_key_header)):
    if api_key != os.environ.get("API_KEY"):
        raise HTTPException(status_code=403, detail="Invalid API key")
    return api_key

@app.post("/api/generate", dependencies=[Depends(verify_api_key)])
async def api_generate(req: GenerateRequest):
    ...

자동 API 문서

FastAPI의 가장 큰 장점은 Swagger UI가 자동으로 생성된다는 것이다. /docs에 접속하면 모든 엔드포인트를 브라우저에서 테스트할 수 있다.

Pydantic 모델에 description을 추가하면 문서 품질이 올라간다.

class GenerateRequest(BaseModel):
    content: str = Field(description="변환할 원본 콘텐츠 (마크다운)")
    channels: list[str] = Field(
        default=['reddit', 'twitter', 'blog'],
        description="대상 채널 목록"
    )

배포

# 로컬 실행
uvicorn promoengine.api.main:app --host 0.0.0.0 --port 8000

# Docker
FROM python:3.12-slim
COPY . /app
WORKDIR /app
RUN pip install .
CMD ["uvicorn", "promoengine.api.main:app", "--host", "0.0.0.0", "--port", "8000"]

정리

CLI를 API로 확장할 때 핵심은 비즈니스 로직을 분리해두는 것이다. core/가 독립적이면, CLI든 API든 Slack 봇이든 인터페이스만 바꿔 끼울 수 있다. FastAPI는 이 과정을 Pydantic 타입 시스템과 자동 문서로 깔끔하게 만들어준다.