이미지 로딩 중...
AI Generated
2025. 11. 12. · 3 Views
Python으로 AI 에이전트 만들기 11편 FastAPI로 AI 서비스 배포
AI 에이전트를 실제 서비스로 배포하는 방법을 배웁니다. FastAPI를 활용해 RESTful API를 구축하고, 비동기 처리, 에러 핸들링, 보안, 성능 최적화까지 실무에 필요한 모든 것을 다룹니다.
목차
- FastAPI_기본_설정
- 비동기_엔드포인트_구현
- Pydantic_모델_검증
- 미들웨어_활용
- CORS_설정
- 스트리밍_응답_구현
- 의존성_주입_시스템
- 배경_작업_처리
- API_문서_자동화
- 환경_변수_관리
1. FastAPI_기본_설정
시작하며
여러분이 멋진 AI 에이전트를 만들었는데, 이걸 실제로 다른 사람들이 사용할 수 있게 하려면 어떻게 해야 할까요? 로컬 환경에서만 돌아가는 스크립트로는 한계가 있습니다.
실제 서비스를 만들려면 웹 서버가 필요합니다. Flask도 있고 Django도 있지만, AI 서비스에는 FastAPI가 압도적으로 유리합니다.
왜일까요? FastAPI는 이름 그대로 '빠른' API를 만들 수 있습니다.
비동기 처리를 기본으로 지원하고, 자동 문서화, 데이터 검증까지 모두 제공합니다. AI 모델 추론처럼 시간이 걸리는 작업을 처리하기에 완벽한 프레임워크입니다.
개요
간단히 말해서, FastAPI는 Python으로 고성능 API를 쉽게 만들 수 있게 해주는 웹 프레임워크입니다. 전통적인 Flask나 Django와 달리 FastAPI는 처음부터 비동기 처리를 염두에 두고 설계되었습니다.
이게 왜 중요할까요? AI 모델이 답변을 생성하는 동안 다른 사용자의 요청도 동시에 처리할 수 있기 때문입니다.
한 명의 요청 때문에 다른 모든 사용자가 기다릴 필요가 없습니다. 또한 FastAPI는 Python 3.6+의 타입 힌팅을 활용합니다.
타입을 지정하면 자동으로 데이터 검증, 직렬화, API 문서까지 생성해줍니다. 일일이 검증 코드를 작성하거나 문서를 따로 관리할 필요가 없습니다.
핵심 특징은 세 가지입니다: 1) 빠른 성능(Node.js, Go와 동등한 수준), 2) 적은 코드로 많은 기능 구현, 3) 자동 대화형 API 문서. 이런 특징들이 개발 시간을 대폭 줄여주고 유지보수를 쉽게 만들어줍니다.
코드 예제
from fastapi import FastAPI
from pydantic import BaseModel
# FastAPI 앱 인스턴스 생성
app = FastAPI(title="AI Agent API", version="1.0.0")
# 요청 데이터 모델 정의
class ChatRequest(BaseModel):
message: str
user_id: str
# 기본 엔드포인트 - 서버 상태 확인
@app.get("/")
async def root():
return {"status": "running", "message": "AI Agent API is ready"}
# AI 채팅 엔드포인트
@app.post("/chat")
async def chat(request: ChatRequest):
# 실제로는 여기서 AI 모델을 호출합니다
response = f"You said: {request.message}"
return {"response": response, "user_id": request.user_id}
설명
이것이 하는 일: FastAPI 앱을 초기화하고 AI 에이전트와 통신할 수 있는 기본 엔드포인트를 만듭니다. 첫 번째로, FastAPI 앱 인스턴스를 생성합니다.
title과 version을 지정하면 자동 생성되는 API 문서에 이 정보가 표시됩니다. 이렇게 하면 API를 사용하는 다른 개발자들이 어떤 버전을 사용하고 있는지 명확히 알 수 있습니다.
그 다음으로, Pydantic 모델인 ChatRequest를 정의합니다. 이 클래스는 단순해 보이지만 매우 강력합니다.
FastAPI가 이 모델을 보고 자동으로 JSON 데이터를 검증하고, 잘못된 형식의 요청은 자동으로 거부합니다. message와 user_id 필드가 없거나 타입이 맞지 않으면 클라이언트에게 명확한 에러 메시지를 보냅니다.
마지막으로, 두 개의 엔드포인트를 정의합니다. GET / 엔드포인트는 서버가 정상 작동하는지 확인하는 헬스 체크용입니다.
POST /chat 엔드포인트는 실제 AI 채팅을 처리합니다. async def를 사용해 비동기 함수로 만들었기 때문에 여러 요청을 동시에 처리할 수 있습니다.
여러분이 이 코드를 실행하면 uvicorn main:app --reload 명령어로 개발 서버를 시작할 수 있습니다. 브라우저에서 http://localhost:8000/docs에 접속하면 자동 생성된 대화형 API 문서를 볼 수 있고, 직접 API를 테스트해볼 수도 있습니다.
Postman이나 curl 없이도 바로 테스트 가능하다는 게 FastAPI의 큰 장점입니다.
실전 팁
💡 개발 중에는 uvicorn main:app --reload 옵션을 사용하세요. 코드를 수정하면 자동으로 서버가 재시작되어 개발 속도가 빨라집니다.
💡 API 버전을 URL에 포함시키는 것도 좋은 방법입니다. @app.post("/v1/chat")처럼 작성하면 나중에 API를 업데이트할 때 기존 클라이언트와의 호환성을 유지할 수 있습니다.
💡 production 환경에서는 여러 워커를 사용하세요. uvicorn main:app --workers 4로 실행하면 멀티코어 CPU를 활용해 더 많은 요청을 처리할 수 있습니다.
💡 환경에 따라 다른 설정을 사용하려면 .env 파일을 만들고 python-dotenv 라이브러리로 로드하세요. API 키나 데이터베이스 URL 같은 민감한 정보를 코드에 하드코딩하지 않아야 합니다.
2. 비동기_엔드포인트_구현
시작하며
여러분의 AI 서비스에 사용자가 한 명씩 들어올 때는 문제가 없습니다. 하지만 동시에 10명, 100명이 요청을 보내면 어떻게 될까요?
일반적인 동기 방식이라면 첫 번째 사용자의 AI 응답이 끝날 때까지 나머지 99명이 기다려야 합니다. 이런 문제는 실제 서비스에서 치명적입니다.
AI 모델 추론은 보통 수 초가 걸리는데, 그 동안 서버가 아무것도 하지 않고 기다린다면 엄청난 리소스 낭비입니다. 사용자 경험도 형편없어지고, 서버 비용도 기하급수적으로 증가합니다.
바로 이럴 때 필요한 것이 비동기 프로그래밍입니다. AI 모델이 생각하는 동안 다른 요청을 처리할 수 있습니다.
같은 서버로 훨씬 많은 사용자를 감당할 수 있습니다.
개요
간단히 말해서, 비동기 프로그래밍은 시간이 걸리는 작업을 기다리는 동안 다른 일을 할 수 있게 해주는 프로그래밍 방식입니다. Python의 async/await 키워드를 사용하면 FastAPI가 자동으로 비동기 처리를 해줍니다.
AI 모델 API를 호출하거나 데이터베이스를 조회하는 동안, 서버는 다른 요청을 받아서 처리를 시작할 수 있습니다. 마치 요리사가 여러 요리를 동시에 하는 것처럼, 오븐에서 굽는 동안 다른 재료를 손질하는 것과 같습니다.
기존 동기 방식에서는 한 번에 하나의 요청만 처리했다면, 이제는 I/O 대기 시간을 활용해 여러 요청을 동시에 처리할 수 있습니다. 특히 외부 API 호출(OpenAI, Anthropic 등)이나 데이터베이스 쿼리처럼 네트워크를 통한 작업에서 효과가 극적입니다.
핵심 특징은: 1) CPU를 효율적으로 사용, 2) 동시 처리 가능한 요청 수 증가, 3) 응답 시간 개선. 실제로 동기 방식 대비 5-10배 많은 요청을 같은 하드웨어에서 처리할 수 있습니다.
코드 예제
import asyncio
import httpx
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
app = FastAPI()
class AIRequest(BaseModel):
prompt: str
max_tokens: int = 100
# 비동기 HTTP 클라이언트로 외부 AI API 호출
async def call_ai_api(prompt: str, max_tokens: int):
async with httpx.AsyncClient(timeout=30.0) as client:
try:
response = await client.post(
"https://api.openai.com/v1/completions",
json={"prompt": prompt, "max_tokens": max_tokens},
headers={"Authorization": "Bearer YOUR_API_KEY"}
)
return response.json()
except httpx.TimeoutException:
raise HTTPException(status_code=504, detail="AI service timeout")
# 비동기 엔드포인트
@app.post("/generate")
async def generate(request: AIRequest):
# await으로 비동기 함수 호출 - 대기 중 다른 요청 처리 가능
result = await call_ai_api(request.prompt, request.max_tokens)
return {"result": result}
설명
이것이 하는 일: 외부 AI API를 비동기로 호출하여 여러 사용자의 요청을 동시에 처리합니다. 첫 번째로, httpx.AsyncClient를 사용합니다.
일반적인 requests 라이브러리는 동기 방식이라 FastAPI의 비동기 장점을 활용할 수 없습니다. httpx는 requests와 비슷한 API를 제공하면서도 비동기를 지원합니다.
async with 문법을 사용해 클라이언트를 자동으로 정리하고, timeout을 설정해 응답이 너무 오래 걸리면 에러를 발생시킵니다. 그 다음으로, call_ai_api 함수를 async def로 정의했습니다.
이 함수 안에서 await client.post()를 호출하면, 네트워크 응답을 기다리는 동안 Python의 이벤트 루프가 다른 작업을 처리할 수 있습니다. 만약 10명이 동시에 요청했다면, 10개의 API 호출이 거의 동시에 전송되고, 각자의 응답을 기다립니다.
마지막으로, /generate 엔드포인트에서 await call_ai_api()를 호출합니다. 여기서 핵심은 await 키워드입니다.
이게 없으면 코루틴 객체만 반환되고 실제 실행은 되지 않습니다. await를 만나면 FastAPI는 "이 요청은 잠시 대기 상태로 두고 다른 요청을 처리하자"고 판단합니다.
여러분이 이 코드를 사용하면 동일한 서버 스펙으로 훨씬 많은 동시 사용자를 지원할 수 있습니다. 예를 들어, AI API 응답이 평균 2초 걸린다면, 동기 방식에서는 초당 0.5명(1/2)만 처리 가능하지만, 비동기 방식에서는 수십 명을 동시에 처리할 수 있습니다.
이는 서버 비용 절감과 직결됩니다.
실전 팁
💡 모든 I/O 작업에 비동기를 사용하세요. 데이터베이스는 asyncpg, Redis는 aioredis, HTTP는 httpx처럼 비동기 라이브러리를 선택해야 진정한 성능 향상을 얻을 수 있습니다.
💡 CPU 집약적인 작업(이미지 처리, 복잡한 계산)은 asyncio.to_thread()나 ProcessPoolExecutor를 사용하세요. 비동기는 I/O 대기에 효과적이지 GIL 때문에 CPU 작업은 여전히 블로킹됩니다.
💡 동시 요청 수를 제한하는 것도 중요합니다. asyncio.Semaphore를 사용해 동시에 처리할 수 있는 AI API 호출을 제한하면, 외부 API의 rate limit를 초과하지 않습니다.
💡 타임아웃을 반드시 설정하세요. 외부 서비스가 응답하지 않으면 요청이 영원히 대기할 수 있습니다. 적절한 타임아웃(보통 30초)으로 리소스 누수를 방지하세요.
3. Pydantic_모델_검증
시작하며
여러분의 API에 누군가 이상한 데이터를 보낸다면 어떻게 될까요? 문자열이 와야 할 곳에 숫자가 오거나, 필수 필드가 빠지거나, 너무 긴 텍스트가 들어올 수 있습니다.
이런 잘못된 데이터가 AI 모델에 그대로 전달되면 에러가 발생하거나 예상치 못한 결과가 나옵니다. 전통적으로는 일일이 if request.get('message') and isinstance(request['message'], str) 같은 검증 코드를 작성해야 했습니다.
필드가 10개, 20개가 되면 검증 코드만 수십 줄이 됩니다. 게다가 하나라도 빠뜨리면 보안 취약점이 될 수 있습니다.
바로 이럴 때 필요한 것이 Pydantic 모델입니다. 데이터 구조를 정의하면 검증, 파싱, 에러 메시지까지 자동으로 처리해줍니다.
안전하고 깔끔한 API를 만드는 핵심 도구입니다.
개요
간단히 말해서, Pydantic은 Python 타입 힌팅을 활용한 데이터 검증 라이브러리입니다. FastAPI의 핵심 기능 중 하나죠.
FastAPI에서 Pydantic 모델을 매개변수로 사용하면, 들어오는 JSON 데이터가 자동으로 검증됩니다. 타입이 맞지 않거나, 필수 필드가 없거나, 값의 범위가 벗어나면 422 에러와 함께 상세한 에러 메시지를 클라이언트에게 보냅니다.
여러분은 검증된 데이터만 받아서 비즈니스 로직에 집중할 수 있습니다. 예를 들어, AI 에이전트에 질문을 보낼 때 질문 길이를 제한하고, 사용자 ID 형식을 검증하고, 선택적 파라미터의 기본값을 설정하는 모든 것을 Pydantic이 처리합니다.
코드 한 줄 추가하지 않고도 견고한 API를 만들 수 있습니다. 핵심 특징은: 1) 자동 타입 변환(문자열 "123"을 정수 123으로), 2) 복잡한 검증 규칙(정규식, 커스텀 검증자), 3) 명확한 에러 메시지.
이런 특징들이 API의 안정성과 사용자 경험을 크게 향상시킵니다.
코드 예제
from fastapi import FastAPI
from pydantic import BaseModel, Field, validator
from typing import Optional, List
app = FastAPI()
class Message(BaseModel):
role: str = Field(..., regex="^(user|assistant|system)$")
content: str = Field(..., min_length=1, max_length=4000)
class ChatRequest(BaseModel):
messages: List[Message]
temperature: float = Field(default=0.7, ge=0.0, le=2.0)
max_tokens: Optional[int] = Field(None, gt=0, le=4000)
user_id: str = Field(..., min_length=3, max_length=50)
# 커스텀 검증자
@validator('messages')
def check_messages_not_empty(cls, v):
if not v:
raise ValueError('messages list cannot be empty')
return v
@app.post("/chat")
async def chat(request: ChatRequest):
# 이 시점에서 request는 이미 완벽하게 검증됨
return {"message_count": len(request.messages)}
설명
이것이 하는 일: 복잡한 검증 규칙을 모델로 정의하여 잘못된 데이터가 API에 들어오는 것을 방지합니다. 첫 번째로, Message 모델을 중첩해서 사용합니다.
Field를 사용하면 다양한 검증 규칙을 선언적으로 정의할 수 있습니다. regex 매개변수로 role이 정확히 세 가지 값 중 하나인지 확인하고, min_length와 max_length로 content가 비어있지 않고 너무 길지 않은지 검증합니다.
이렇게 하면 AI 모델이 처리할 수 없는 형식의 메시지를 미리 차단할 수 있습니다. 그 다음으로, ChatRequest 모델에서 복잡한 타입을 사용합니다.
List[Message]는 메시지 리스트를 의미하고, 각 메시지도 자동으로 검증됩니다. temperature 필드는 ge(greater than or equal)와 le(less than or equal)로 범위를 제한합니다.
OpenAI API 같은 외부 서비스의 제약사항을 여기서 미리 검증할 수 있습니다. Optional[int]는 이 필드가 없어도 된다는 의미입니다.
마지막으로, @validator 데코레이터로 커스텀 검증 로직을 추가합니다. 단순한 타입이나 범위 검증을 넘어서 비즈니스 로직에 맞는 검증이 필요할 때 사용합니다.
예를 들어, messages 리스트가 비어있으면 안 되는 규칙을 검증합니다. 검증 실패 시 ValueError를 발생시키면 FastAPI가 이를 422 에러로 변환해 클라이언트에게 전달합니다.
여러분이 이 코드를 사용하면 API가 훨씬 견고해집니다. 잘못된 요청은 엔드포인트 함수에 도달하기 전에 차단되고, 명확한 에러 메시지로 클라이언트 개발자가 무엇을 고쳐야 하는지 알 수 있습니다.
디버깅 시간이 대폭 줄어들고, production 환경에서의 예상치 못한 에러도 줄어듭니다.
실전 팁
💡 Field의 description 매개변수를 사용하면 자동 생성되는 API 문서에 각 필드에 대한 설명이 추가됩니다. 다른 개발자가 여러분의 API를 사용할 때 큰 도움이 됩니다.
💡 Config 클래스로 모델 동작을 커스터마이징할 수 있습니다. class Config: extra = "forbid"를 추가하면 정의하지 않은 추가 필드를 거부합니다. 보안상 중요한 API에서 유용합니다.
💡 응답 모델도 Pydantic으로 정의하세요. @app.post("/chat", response_model=ChatResponse)처럼 지정하면 응답도 검증되고 문서화됩니다. 실수로 민감한 정보를 노출하는 것도 방지할 수 있습니다.
💡 model_validator (Pydantic v2)를 사용하면 여러 필드를 동시에 검증할 수 있습니다. 예를 들어, start_date가 end_date보다 빠른지 검증하는 것처럼 필드 간 관계를 확인할 때 유용합니다.
4. 미들웨어_활용
시작하며
여러분의 API에서 에러가 발생했을 때, 정확히 어떤 요청이 문제를 일으켰는지 어떻게 알 수 있을까요? 각 엔드포인트마다 로깅 코드를 추가하면 코드가 지저분해지고, 하나라도 빠뜨리면 추적이 불가능합니다.
실제 서비스에서는 모든 요청을 로깅하고, 응답 시간을 측정하고, 에러를 일관되게 처리해야 합니다. 또한 인증, CORS, 압축 같은 공통 기능도 필요합니다.
이런 코드를 각 엔드포인트에 복붙하는 건 유지보수의 악몽입니다. 바로 이럴 때 필요한 것이 미들웨어입니다.
모든 요청과 응답을 가로채서 공통 로직을 실행할 수 있습니다. 한 곳에서 관리하니 일관성도 보장되고 코드도 깔끔해집니다.
개요
간단히 말해서, 미들웨어는 요청이 엔드포인트에 도달하기 전과 응답이 클라이언트에게 가기 전에 실행되는 중간 계층입니다. FastAPI의 미들웨어는 파이프라인처럼 동작합니다.
요청이 들어오면 첫 번째 미들웨어부터 순서대로 거치고, 엔드포인트에서 처리된 후, 다시 역순으로 미들웨어를 거쳐 응답이 나갑니다. 각 미들웨어는 요청을 수정하거나, 로깅하거나, 조기에 응답을 반환할 수 있습니다.
예를 들어, AI 서비스에서 각 요청의 응답 시간을 모니터링하고, 느린 요청을 찾아내고, 사용자별 API 호출 수를 제한하는 모든 것을 미들웨어로 구현할 수 있습니다. 비즈니스 로직과 완전히 분리되어 깔끔합니다.
핵심 특징은: 1) 모든 요청/응답에 자동 적용, 2) 엔드포인트 코드 수정 없이 기능 추가, 3) 재사용 가능한 컴포넌트. 이런 특징들이 코드의 가독성과 유지보수성을 크게 향상시킵니다.
코드 예제
import time
import logging
from fastapi import FastAPI, Request, status
from fastapi.responses import JSONResponse
app = FastAPI()
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# 로깅 및 타이밍 미들웨어
@app.middleware("http")
async def log_and_time_requests(request: Request, call_next):
# 요청 시작 시간 기록
start_time = time.time()
request_id = id(request) # 간단한 요청 ID
logger.info(f"[{request_id}] {request.method} {request.url.path} started")
try:
# 다음 미들웨어 또는 엔드포인트 호출
response = await call_next(request)
# 응답 시간 계산
process_time = time.time() - start_time
response.headers["X-Process-Time"] = str(process_time)
logger.info(f"[{request_id}] completed in {process_time:.3f}s - status: {response.status_code}")
return response
except Exception as e:
# 예외 처리 및 일관된 에러 응답
process_time = time.time() - start_time
logger.error(f"[{request_id}] failed in {process_time:.3f}s - error: {str(e)}")
return JSONResponse(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
content={"detail": "Internal server error", "request_id": str(request_id)}
)
설명
이것이 하는 일: 모든 API 요청의 시작과 끝을 로깅하고, 처리 시간을 측정하며, 예외를 일관된 형식으로 처리합니다. 첫 번째로, 미들웨어 함수를 @app.middleware("http") 데코레이터로 등록합니다.
이 함수는 request와 call_next 두 매개변수를 받습니다. request는 현재 HTTP 요청 객체이고, call_next는 다음 처리 단계(다른 미들웨어나 실제 엔드포인트)를 호출하는 함수입니다.
요청 시작 시간을 기록하고 간단한 요청 ID를 생성해서 로그에서 같은 요청을 추적할 수 있게 합니다. 그 다음으로, await call_next(request)를 호출합니다.
이 부분이 핵심입니다. 여기서 실제 엔드포인트가 실행되고, AI 모델이 호출되는 등 모든 처리가 일어납니다.
await 후에 반환된 response 객체를 받으면, 끝나는 시간을 측정해서 총 처리 시간을 계산합니다. 이 정보를 X-Process-Time 헤더에 추가하면 클라이언트가 API 성능을 모니터링할 수 있습니다.
마지막으로, try-except 블록으로 모든 예외를 잡습니다. 엔드포인트에서 처리되지 않은 예외가 발생하면, 미들웨어에서 이를 캐치해서 일관된 에러 응답을 반환합니다.
사용자에게는 친절한 에러 메시지를, 로그에는 상세한 정보를 남깁니다. request_id를 응답에 포함시키면 사용자가 문의할 때 해당 요청을 정확히 찾을 수 있습니다.
여러분이 이 코드를 사용하면 production 환경에서 무엇이 일어나고 있는지 완벽하게 파악할 수 있습니다. 어떤 엔드포인트가 느린지, 어떤 에러가 자주 발생하는지, 사용자가 어떤 패턴으로 API를 사용하는지 모두 로그에 남습니다.
문제가 발생하면 로그를 보고 빠르게 원인을 찾을 수 있습니다.
실전 팁
💡 미들웨어는 등록 순서가 중요합니다. 먼저 등록한 미들웨어가 요청에서는 먼저 실행되고, 응답에서는 나중에 실행됩니다. 인증 미들웨어는 보통 가장 먼저, 로깅은 그 다음에 배치하세요.
💡 구조화된 로깅을 사용하면 더 좋습니다. python-json-logger 같은 라이브러리로 로그를 JSON 형식으로 출력하면 Elasticsearch나 CloudWatch에서 분석하기 쉽습니다.
💡 미들웨어에서 너무 무거운 작업을 하지 마세요. 데이터베이스 조회 같은 건 지양하고, 간단한 로깅이나 헤더 수정 정도만 하는 게 좋습니다. 모든 요청에 영향을 주기 때문입니다.
💡 request.state를 활용하면 미들웨어와 엔드포인트 간에 데이터를 공유할 수 있습니다. 예를 들어, 인증 미들웨어에서 request.state.user = user 하면 엔드포인트에서 바로 사용할 수 있습니다.
5. CORS_설정
시작하며
여러분이 만든 AI API를 웹 프론트엔드에서 호출하려고 하는데, 브라우저 콘솔에 빨간색 에러가 뜨면서 "CORS policy" 어쩌고 하는 메시지를 본 적 있나요? API는 정상인데 브라우저가 막는 이 상황, 정말 답답합니다.
이건 보안을 위한 브라우저의 Same-Origin Policy 때문입니다. 다른 도메인의 웹사이트가 여러분의 API를 마음대로 호출하지 못하게 막는 거죠.
하지만 실제로는 자신의 프론트엔드에서 자신의 API를 호출하려는 것인데 막히는 경우가 많습니다. 바로 이럴 때 필요한 것이 CORS(Cross-Origin Resource Sharing) 설정입니다.
어떤 도메인이 여러분의 API를 호출할 수 있는지 명시적으로 허용해야 합니다. 올바르게 설정하면 프론트엔드와 백엔드가 완벽하게 연동됩니다.
개요
간단히 말해서, CORS는 다른 도메인에서 실행되는 웹 애플리케이션이 여러분의 API에 접근할 수 있도록 허용하는 메커니즘입니다. 브라우저는 보안상의 이유로 기본적으로 다른 출처(origin)로의 요청을 차단합니다.
예를 들어, https://myapp.com에서 실행되는 JavaScript가 https://api.myapp.com을 호출하면 도메인이 다르기 때문에 막힙니다. CORS 헤더를 응답에 추가하면 "이 도메인은 허용된 곳이야"라고 브라우저에게 알려줍니다.
FastAPI는 CORS를 쉽게 설정할 수 있는 미들웨어를 제공합니다. 개발 환경에서는 모든 도메인을 허용하고, production에서는 실제 프론트엔드 도메인만 허용하는 식으로 환경별로 다르게 설정할 수 있습니다.
핵심 특징은: 1) 허용할 도메인 명시, 2) HTTP 메서드 제한, 3) 인증 정보(쿠키) 포함 여부 설정. 이런 설정들이 API의 보안과 접근성 사이의 균형을 맞춰줍니다.
코드 예제
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
import os
app = FastAPI()
# 환경 변수로 허용 도메인 관리
ALLOWED_ORIGINS = os.getenv(
"ALLOWED_ORIGINS",
"http://localhost:3000,http://localhost:5173" # 개발 환경 기본값
).split(",")
# CORS 미들웨어 추가
app.add_middleware(
CORSMiddleware,
allow_origins=ALLOWED_ORIGINS, # 허용할 도메인 리스트
allow_credentials=True, # 쿠키를 포함한 요청 허용
allow_methods=["GET", "POST", "PUT", "DELETE"], # 허용할 HTTP 메서드
allow_headers=["*"], # 모든 헤더 허용 (특정 헤더만 지정 가능)
max_age=3600, # preflight 요청 캐시 시간 (초)
)
@app.get("/api/status")
async def status():
return {"status": "CORS is configured correctly"}
설명
이것이 하는 일: 브라우저가 다른 도메인의 API 호출을 허용하도록 적절한 CORS 헤더를 응답에 추가합니다. 첫 번째로, 허용할 도메인을 환경 변수에서 읽어옵니다.
코드에 하드코딩하지 않고 환경 변수를 사용하면 같은 코드로 개발/스테이징/production 환경을 운영할 수 있습니다. 개발 중에는 localhost:3000(React), localhost:5173(Vite) 같은 로컬 주소를, production에서는 실제 도메인을 설정합니다.
split(",")로 여러 도메인을 지원합니다. 그 다음으로, CORSMiddleware를 추가합니다.
allow_origins에 허용할 도메인 리스트를 전달합니다. ["*"]로 모든 도메인을 허용할 수도 있지만, 보안상 좋지 않습니다.
특히 allow_credentials=True일 때는 ["*"]를 사용할 수 없습니다. allow_credentials를 True로 설정하면 쿠키나 인증 헤더를 포함한 요청을 받을 수 있습니다.
로그인이 필요한 API에서 필수입니다. 마지막으로, HTTP 메서드와 헤더를 제한합니다.
allow_methods로 GET, POST 같은 필요한 메서드만 허용합니다. 사용하지 않는 메서드를 막으면 보안이 강화됩니다.
allow_headers는 클라이언트가 보낼 수 있는 헤더를 지정하는데, 일반적으로는 ["*"]로 모두 허용해도 괜찮습니다. max_age는 브라우저가 preflight 요청(OPTIONS) 결과를 캐시하는 시간인데, 이 값을 높이면 네트워크 요청이 줄어듭니다.
여러분이 이 코드를 사용하면 React, Vue, Angular 같은 SPA 프레임워크에서 만든 프론트엔드가 문제없이 여러분의 API를 호출할 수 있습니다. 브라우저 콘솔에 CORS 에러가 사라지고, 실제 API 응답을 받아서 화면에 표시할 수 있습니다.
모바일 앱이나 서버-to-서버 통신에서는 CORS가 적용되지 않으니 걱정하지 않아도 됩니다.
실전 팁
💡 개발 중에는 allow_origins=["*"]를 사용해도 되지만, production에 배포하기 전에 반드시 실제 도메인으로 변경하세요. 모든 도메인을 허용하면 CSRF 공격에 취약해집니다.
💡 서브도메인을 사용한다면 정규식으로 도메인을 지정할 수 있습니다. allow_origin_regex=r"https://.*\.myapp\.com"처럼 하면 api.myapp.com, admin.myapp.com 모두 허용됩니다.
💡 preflight 요청이 많이 발생하면 성능에 영향을 줄 수 있습니다. max_age를 적절히 높여서 불필요한 OPTIONS 요청을 줄이세요. 1시간(3600초)이 일반적입니다.
💡 CloudFront나 Nginx 같은 프록시를 사용한다면 거기서도 CORS를 설정할 수 있습니다. API 서버보다 프록시에서 설정하는 게 더 효율적인 경우도 있습니다.
6. 스트리밍_응답_구현
시작하며
여러분이 ChatGPT처럼 AI가 답변을 타이핑하듯이 실시간으로 보여주고 싶다면 어떻게 해야 할까요? 일반적인 API 방식으로는 AI가 전체 답변을 다 생성할 때까지 기다렸다가 한 번에 받습니다.
긴 답변이면 20-30초를 멍하니 기다려야 합니다. 사용자 경험 관점에서 이건 치명적입니다.
아무 반응이 없으면 사용자는 "서버가 죽은 건가?" 하고 걱정합니다. 실제로는 AI가 열심히 답변을 생성하고 있는데 말이죠.
중간에 취소도 할 수 없습니다. 바로 이럴 때 필요한 것이 스트리밍 응답입니다.
AI가 토큰을 생성하는 즉시 클라이언트로 전송해서, 사용자가 실시간으로 답변을 볼 수 있게 합니다. 체감 속도가 엄청나게 빨라집니다.
개요
간단히 말해서, 스트리밍 응답은 데이터를 한 번에 보내지 않고 생성되는 대로 조금씩 전송하는 방식입니다. FastAPI의 StreamingResponse를 사용하면 Python 제너레이터나 비동기 제너레이터를 응답으로 보낼 수 있습니다.
AI API들(OpenAI, Anthropic)은 대부분 스트리밍을 지원하는데, 이를 그대로 클라이언트에게 전달할 수 있습니다. Server-Sent Events(SSE) 형식으로 보내면 브라우저의 EventSource API로 쉽게 받을 수 있습니다.
예를 들어, OpenAI의 스트리밍 응답을 받아서 각 토큰을 실시간으로 클라이언트에게 전달하면, ChatGPT와 똑같은 타이핑 효과를 구현할 수 있습니다. 사용자는 즉시 반응을 보고, 원하지 않는 답변이면 중간에 취소할 수도 있습니다.
핵심 특징은: 1) 낮은 지연시간(첫 번째 토큰이 즉시 전송), 2) 메모리 효율(전체 응답을 메모리에 보관하지 않음), 3) 실시간 업데이트. 이런 특징들이 AI 애플리케이션의 사용자 경험을 극적으로 개선합니다.
코드 예제
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio
import json
app = FastAPI()
# AI 응답을 스트리밍하는 비동기 제너레이터
async def generate_ai_stream(prompt: str):
# 실제로는 OpenAI API 스트리밍 호출
# 여기서는 시뮬레이션
full_response = f"This is a response to: {prompt}"
for word in full_response.split():
# 각 단어를 SSE 형식으로 전송
data = {"token": word, "done": False}
yield f"data: {json.dumps(data)}\n\n"
await asyncio.sleep(0.1) # AI 생성 시뮬레이션
# 완료 신호
yield f"data: {json.dumps({'done': True})}\n\n"
@app.get("/stream")
async def stream_chat(prompt: str):
return StreamingResponse(
generate_ai_stream(prompt),
media_type="text/event-stream", # SSE 형식
headers={
"Cache-Control": "no-cache",
"Connection": "keep-alive",
}
)
설명
이것이 하는 일: AI가 생성하는 토큰을 즉시 클라이언트로 전송하여 ChatGPT 같은 타이핑 효과를 만듭니다. 첫 번째로, 비동기 제너레이터 함수를 정의합니다.
async def 안에서 yield를 사용하면 비동기 제너레이터가 됩니다. 이 함수는 값을 하나씩 생성해서 반환합니다.
일반 함수처럼 한 번에 모든 데이터를 메모리에 만들지 않고, 필요할 때마다 하나씩 생성합니다. AI 응답이 수천 토큰이어도 메모리 사용량은 일정합니다.
그 다음으로, Server-Sent Events 형식으로 데이터를 포맷합니다. SSE는 간단한 텍스트 프로토콜로, data: {내용}\n\n 형식으로 보냅니다.
각 토큰을 JSON으로 직렬화해서 보내면 클라이언트에서 파싱하기 쉽습니다. done 필드로 스트림이 끝났는지 알릴 수 있습니다.
실제 구현에서는 OpenAI의 스트리밍 응답을 받아서 그대로 클라이언트로 전달하면 됩니다. 마지막으로, StreamingResponse를 반환합니다.
media_type="text/event-stream"으로 SSE임을 명시하고, 캐싱을 비활성화하는 헤더를 추가합니다. 캐싱되면 스트리밍이 제대로 작동하지 않습니다.
Connection: keep-alive로 연결을 유지하면서 데이터를 계속 전송합니다. 클라이언트는 EventSource API로 이 스트림을 받아서 화면에 표시합니다.
여러분이 이 코드를 사용하면 사용자가 AI 응답을 기다리는 시간이 체감상 거의 사라집니다. 첫 번째 단어가 0.5초 안에 화면에 나타나고, 계속해서 단어가 추가됩니다.
전체 응답이 완성되기까지 10초가 걸려도, 사용자는 즉각적인 피드백을 받기 때문에 만족도가 높습니다. 취소 버튼을 누르면 스트림을 중단할 수도 있습니다.
실전 팁
💡 실제 OpenAI API를 사용할 때는 stream=True 옵션을 주고, 응답을 반복하면서 각 청크를 그대로 yield 하면 됩니다. 매우 간단합니다.
💡 에러 처리가 중요합니다. 스트리밍 중간에 에러가 발생하면 에러 메시지도 SSE 형식으로 보내야 클라이언트가 처리할 수 있습니다. yield f"data: {json.dumps({'error': str(e)})}\n\n"
💡 타임아웃 설정이 필요합니다. AI가 응답을 멈추면 스트림이 영원히 열려있을 수 있습니다. asyncio.wait_for()로 최대 시간을 제한하세요.
💡 클라이언트에서는 EventSource 대신 fetch API로도 받을 수 있습니다. response.body.getReader()로 ReadableStream을 얻어서 청크를 읽으면 됩니다. 더 세밀한 제어가 가능합니다.
7. 의존성_주입_시스템
시작하며
여러분이 여러 엔드포인트에서 똑같은 코드를 반복하고 있다면 뭔가 잘못된 겁니다. 예를 들어, 사용자 인증 확인, 데이터베이스 연결, 설정 값 읽기 같은 작업을 각 함수마다 복붙하고 계신가요?
코드 중복은 버그의 온상입니다. 한 곳을 수정하면 다른 곳도 다 수정해야 하고, 하나라도 빠뜨리면 버그가 됩니다.
특히 인증 로직 같은 보안 관련 코드는 일관성이 생명인데, 10군데 다른 방식으로 구현되어 있으면 큰 문제입니다. 바로 이럴 때 필요한 것이 의존성 주입(Dependency Injection)입니다.
공통 로직을 한 번 정의하고 필요한 곳에 자동으로 주입받습니다. 코드 재사용성도 높아지고 테스트도 쉬워집니다.
개요
간단히 말해서, 의존성 주입은 함수가 필요로 하는 것들을 FastAPI가 자동으로 제공해주는 메커니즘입니다. FastAPI의 Depends를 사용하면 함수나 클래스를 의존성으로 선언할 수 있습니다.
엔드포인트가 호출되기 전에 FastAPI가 자동으로 의존성을 실행하고 결과를 매개변수로 전달합니다. 여러 엔드포인트가 같은 의존성을 사용하면 코드 중복 없이 공통 로직을 공유할 수 있습니다.
예를 들어, JWT 토큰을 검증해서 현재 사용자를 가져오는 로직을 의존성으로 만들면, 모든 보호된 엔드포인트에서 current_user: User = Depends(get_current_user) 한 줄로 인증을 처리할 수 있습니다. 의존성은 다른 의존성에도 의존할 수 있어서 복잡한 계층 구조도 만들 수 있습니다.
핵심 특징은: 1) 코드 재사용, 2) 명확한 의존 관계, 3) 쉬운 테스트(의존성을 mock으로 교체 가능). 이런 특징들이 대규모 프로젝트에서 코드 품질을 유지하는 핵심입니다.
코드 예제
from fastapi import FastAPI, Depends, HTTPException, Header
from typing import Optional
import jwt
app = FastAPI()
# 설정 의존성
class Settings:
def __init__(self):
self.jwt_secret = "your-secret-key"
self.jwt_algorithm = "HS256"
def get_settings():
return Settings()
# 토큰 검증 의존성
async def verify_token(
authorization: Optional[str] = Header(None),
settings: Settings = Depends(get_settings)
):
if not authorization or not authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="Missing or invalid token")
token = authorization.split(" ")[1]
try:
payload = jwt.decode(token, settings.jwt_secret, algorithms=[settings.jwt_algorithm])
return payload["user_id"]
except jwt.InvalidTokenError:
raise HTTPException(status_code=401, detail="Invalid token")
# 사용자 정보 의존성 (다른 의존성에 의존)
async def get_current_user(user_id: str = Depends(verify_token)):
# 실제로는 데이터베이스에서 사용자 조회
return {"id": user_id, "name": "User"}
# 보호된 엔드포인트
@app.get("/profile")
async def get_profile(current_user: dict = Depends(get_current_user)):
return {"user": current_user}
설명
이것이 하는 일: 인증, 설정, 데이터베이스 연결 같은 공통 로직을 의존성으로 만들어 여러 엔드포인트에서 재사용합니다. 첫 번째로, 설정 의존성을 만듭니다.
Settings 클래스는 애플리케이션 설정을 담고 있고, get_settings() 함수가 이를 반환합니다. 실제로는 환경 변수나 설정 파일에서 읽어올 것입니다.
이 패턴의 장점은 테스트 시 get_settings를 mock으로 교체해서 테스트용 설정을 주입할 수 있다는 겁니다. 그 다음으로, 토큰 검증 의존성을 만듭니다.
verify_token 함수는 Header로 Authorization 헤더를 자동으로 추출하고, Depends(get_settings)로 설정을 주입받습니다. 토큰이 없거나 유효하지 않으면 즉시 401 에러를 발생시킵니다.
이 함수가 반환하는 user_id는 다음 단계로 전달됩니다. 마지막으로, 의존성 체인을 만듭니다.
get_current_user는 Depends(verify_token)으로 검증된 사용자 ID를 받습니다. 이미 토큰 검증은 끝났으므로, 여기서는 사용자 정보를 데이터베이스에서 가져오는 로직만 작성하면 됩니다.
엔드포인트 함수는 Depends(get_current_user)로 완전한 사용자 객체를 받습니다. 인증과 사용자 조회가 모두 자동으로 처리됩니다.
여러분이 이 코드를 사용하면 보호된 엔드포인트를 만들 때마다 토큰 검증 코드를 작성할 필요가 없습니다. current_user: dict = Depends(get_current_user) 한 줄이면 끝입니다.
인증 로직을 수정해야 하면 verify_token 함수 하나만 고치면 모든 엔드포인트에 자동으로 적용됩니다. 버그 가능성이 줄어들고 유지보수가 쉬워집니다.
실전 팁
💡 의존성을 클래스로 만들면 더 깔끔합니다. __call__ 메서드를 구현하면 클래스가 함수처럼 동작합니다. 상태를 유지해야 하는 의존성(데이터베이스 세션 풀 등)에 유용합니다.
💡 Depends에 use_cache=False를 지정하면 같은 요청 내에서도 의존성을 매번 실행합니다. 기본적으로는 같은 요청에서 같은 의존성을 여러 번 호출해도 한 번만 실행되고 결과가 캐시됩니다.
💡 전역 의존성을 설정할 수도 있습니다. app = FastAPI(dependencies=[Depends(verify_token)])처럼 하면 모든 엔드포인트에 자동으로 적용됩니다. 인증이 필요한 admin 라우터 같은 곳에 유용합니다.
💡 의존성에서 yield를 사용하면 요청 전후에 코드를 실행할 수 있습니다. 데이터베이스 세션을 열고, 요청 처리 후 자동으로 닫는 패턴에 완벽합니다. try-finally처럼 동작합니다.
8. 배경_작업_처리
시작하며
여러분의 AI 에이전트가 답변을 생성한 후에 이메일 알림을 보내거나, 로그를 데이터베이스에 저장하거나, 분석 데이터를 업데이트해야 한다면 어떻게 해야 할까요? 이런 작업들을 API 응답 전에 다 하면 사용자는 불필요하게 오래 기다려야 합니다.
사용자는 AI 답변만 받으면 되는데, 내부적인 로깅이나 알림 때문에 2-3초를 더 기다린다면 사용자 경험이 나빠집니다. 이런 부가 작업들은 응답 속도에 영향을 주지 않아야 합니다.
하지만 그렇다고 안 할 수도 없는 중요한 작업들입니다. 바로 이럴 때 필요한 것이 배경 작업(Background Tasks)입니다.
응답을 바로 반환하고, 그 후에 추가 작업을 처리합니다. 사용자는 빠른 응답을 받고, 서버는 필요한 작업을 모두 수행합니다.
개요
간단히 말해서, 배경 작업은 API 응답을 보낸 후에 실행되는 작업입니다. 사용자는 기다리지 않습니다.
FastAPI의 BackgroundTasks를 사용하면 응답 후에 실행할 함수를 예약할 수 있습니다. 클라이언트는 즉시 응답을 받고, 서버는 백그라운드에서 추가 작업을 계속 처리합니다.
간단한 작업(로깅, 알림, 캐시 업데이트)에 적합합니다. 예를 들어, 사용자가 AI에게 질문하면 즉시 답변을 받고, 동시에 서버는 질문과 답변을 데이터베이스에 저장하고 사용량 통계를 업데이트합니다.
사용자는 이런 과정을 전혀 느끼지 못합니다. 중요한 점은 배경 작업은 같은 프로세스에서 실행되므로, 시간이 오래 걸리거나 실패 가능성이 높은 작업은 Celery 같은 작업 큐를 사용해야 합니다.
핵심 특징은: 1) 빠른 API 응답, 2) 비동기/동기 함수 모두 지원, 3) 간단한 구현. 이런 특징들이 사용자 경험과 서버 효율성을 동시에 개선합니다.
코드 예제
from fastapi import FastAPI, BackgroundTasks
from datetime import datetime
import logging
app = FastAPI()
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
# 배경에서 실행될 작업 함수들
def log_request(user_id: str, prompt: str, response: str):
"""요청/응답을 데이터베이스에 로깅"""
logger.info(f"Logging request for user {user_id}")
# 실제로는 데이터베이스에 저장
# db.save_conversation(user_id, prompt, response, datetime.now())
def update_usage_stats(user_id: str, tokens_used: int):
"""사용자의 토큰 사용량 업데이트"""
logger.info(f"Updating usage stats: {user_id} used {tokens_used} tokens")
# 실제로는 사용량 테이블 업데이트
# db.increment_usage(user_id, tokens_used)
async def send_notification(user_id: str, message: str):
"""비동기 알림 전송"""
logger.info(f"Sending notification to {user_id}: {message}")
# 실제로는 이메일/SMS/푸시 알림
# await notification_service.send(user_id, message)
@app.post("/chat")
async def chat(
prompt: str,
user_id: str,
background_tasks: BackgroundTasks
):
# AI 응답 생성 (메인 로직)
response = f"AI response to: {prompt}"
tokens_used = len(response.split())
# 배경 작업 예약 (응답 후 실행)
background_tasks.add_task(log_request, user_id, prompt, response)
background_tasks.add_task(update_usage_stats, user_id, tokens_used)
background_tasks.add_task(send_notification, user_id, "Your response is ready")
# 즉시 응답 반환 - 배경 작업은 이후에 실행됨
return {"response": response, "tokens": tokens_used}
설명
이것이 하는 일: API 응답을 즉시 반환하고, 로깅이나 알림 같은 부가 작업은 백그라운드에서 처리합니다. 첫 번째로, 배경에서 실행될 함수들을 정의합니다.
이 함수들은 일반 Python 함수나 비동기 함수 모두 가능합니다. log_request는 대화 내용을 데이터베이스에 저장하고, update_usage_stats는 사용량을 집계하고, send_notification은 사용자에게 알림을 보냅니다.
이런 작업들은 중요하지만 사용자가 즉시 알 필요는 없는 것들입니다. 그 다음으로, 엔드포인트에 background_tasks: BackgroundTasks 매개변수를 추가합니다.
FastAPI가 자동으로 BackgroundTasks 인스턴스를 주입해줍니다. 메인 로직(AI 응답 생성)을 먼저 처리하고, background_tasks.add_task()로 배경 작업을 예약합니다.
함수와 인자를 전달하면 됩니다. 여러 작업을 추가할 수 있고, 추가한 순서대로 실행됩니다.
마지막으로, 응답을 즉시 반환합니다. 중요한 점은 return 문이 실행되자마자 클라이언트가 응답을 받는다는 겁니다.
그 후에 FastAPI가 예약된 배경 작업들을 순서대로 실행합니다. 배경 작업이 실패해도 이미 응답은 보내진 상태이므로 사용자에게 영향을 주지 않습니다.
물론 로깅을 통해 실패를 추적해야 합니다. 여러분이 이 코드를 사용하면 API 응답 시간이 크게 줄어듭니다.
예를 들어, 메인 로직이 1초, 로깅과 알림이 0.5초 걸린다면, 기존에는 1.5초를 기다려야 했지만 이제는 1초만 기다리면 됩니다. 사용자는 30% 빨라진 것처럼 느낍니다.
특히 모바일 환경에서 이런 차이가 사용자 만족도에 큰 영향을 줍니다.
실전 팁
💡 배경 작업에서 에러가 발생해도 응답은 이미 보내진 상태입니다. 중요한 작업은 반드시 try-except로 감싸고 에러를 로깅하세요. Sentry 같은 에러 추적 도구와 연동하는 것도 좋습니다.
💡 시간이 오래 걸리는 작업(1분 이상)은 배경 작업보다 Celery나 Redis Queue 같은 작업 큐를 사용하세요. 배경 작업은 같은 워커 프로세스에서 실행되므로 다른 요청 처리를 방해할 수 있습니다.
💡 배경 작업은 응답 후 실행되지만, 테스트 환경에서는 동기적으로 실행됩니다. 테스트할 때는 배경 작업도 완료될 때까지 기다리므로 실제 동작을 검증할 수 있습니다.
💡 데이터베이스 세션을 배경 작업에서 사용할 때 주의하세요. 의존성 주입으로 받은 세션은 응답 후 닫힐 수 있습니다. 배경 작업 안에서 새 세션을 생성하거나, 세션을 명시적으로 전달하세요.
9. API_문서_자동화
시작하며
여러분이 만든 API를 다른 개발자가 사용하려면 어떻게 해야 할까요? "이 엔드포인트는 이런 파라미터를 받고 저런 응답을 줍니다" 같은 문서를 Word나 Notion에 작성하고 계신가요?
코드가 바뀌면 문서도 업데이트해야 하는데 깜빡하기 쉽습니다. 문서가 없거나 오래된 문서는 큰 문제입니다.
API를 사용하려는 개발자가 코드를 뜯어봐야 하거나, 시행착오를 겪어야 합니다. "이 필드가 필수인가요?" "어떤 형식으로 보내야 하나요?" 같은 질문이 계속 들어옵니다.
시간 낭비가 심각합니다. 바로 이럴 때 필요한 것이 자동 API 문서입니다.
FastAPI는 코드에서 자동으로 OpenAPI 스펙을 생성하고, Swagger UI와 ReDoc 같은 대화형 문서를 제공합니다. 코드가 곧 문서입니다.
개요
간단히 말해서, FastAPI는 여러분의 코드를 분석해서 자동으로 API 문서를 생성하고 웹 인터페이스로 보여줍니다. Pydantic 모델, 타입 힌팅, docstring을 보고 FastAPI가 OpenAPI 스펙(구 Swagger)을 자동 생성합니다.
/docs에 가면 Swagger UI로 모든 엔드포인트를 볼 수 있고, 브라우저에서 직접 테스트할 수 있습니다. /redoc에는 더 깔끔한 ReDoc 스타일 문서가 있습니다.
별도의 작업 없이 항상 최신 상태를 유지합니다. 예를 들어, Pydantic 모델에 Field(description="사용자 ID")를 추가하면 자동으로 문서에 설명이 표시됩니다.
엔드포인트 함수에 docstring을 작성하면 그것도 문서에 나타납니다. 타입 힌팅만 제대로 해도 요청/응답 예시가 자동으로 생성됩니다.
핵심 특징은: 1) 코드와 문서의 자동 동기화, 2) 대화형 API 테스트, 3) OpenAPI 표준 준수. 이런 특징들이 API 품질과 개발자 경험을 크게 향상시킵니다.
코드 예제
from fastapi import FastAPI, Query, Path
from pydantic import BaseModel, Field
from typing import List, Optional
app = FastAPI(
title="AI Agent API",
description="FastAPI로 구축한 AI 에이전트 서비스",
version="1.0.0",
docs_url="/docs", # Swagger UI 경로
redoc_url="/redoc", # ReDoc 경로
)
class ChatMessage(BaseModel):
role: str = Field(..., description="메시지 역할: user, assistant, system")
content: str = Field(..., description="메시지 내용")
class Config:
schema_extra = {
"example": {
"role": "user",
"content": "Python으로 AI 에이전트 만드는 방법을 알려줘"
}
}
class ChatResponse(BaseModel):
response: str = Field(..., description="AI 생성 응답")
tokens_used: int = Field(..., description="사용된 토큰 수")
@app.post(
"/chat",
response_model=ChatResponse,
summary="AI 채팅",
description="AI 에이전트와 대화합니다. 질문을 보내면 AI가 답변을 생성합니다.",
response_description="AI의 답변과 토큰 사용량",
tags=["Chat"]
)
async def chat(
message: ChatMessage,
temperature: float = Query(0.7, ge=0, le=2, description="응답의 창의성 (0-2)"),
max_tokens: Optional[int] = Query(None, gt=0, description="최대 토큰 수")
) -> ChatResponse:
"""
AI 에이전트와 채팅합니다.
- **role**: 메시지 역할 (user, assistant, system)
- **content**: 사용자의 질문이나 지시사항
- **temperature**: 높을수록 창의적인 응답
- **max_tokens**: 응답 길이 제한
"""
return ChatResponse(response="AI 응답", tokens_used=42)
설명
이것이 하는 일: 타입 힌팅과 Pydantic 모델에서 자동으로 OpenAPI 문서를 생성하고 Swagger UI로 제공합니다. 첫 번째로, FastAPI 앱에 메타데이터를 추가합니다.
title, description, version은 문서의 헤더에 표시됩니다. API가 무엇을 하는지, 현재 버전이 무엇인지 한눈에 알 수 있습니다.
docs_url과 redoc_url로 문서 경로를 커스터마이징할 수 있고, production에서는 비활성화(docs_url=None)할 수도 있습니다. 그 다음으로, Pydantic 모델을 문서화합니다.
Field의 description 매개변수는 각 필드에 대한 설명을 제공합니다. Config.schema_extra에 예시를 추가하면 문서에서 "Try it out" 버튼을 누를 때 기본값으로 채워집니다.
사용자가 바로 테스트할 수 있어서 편리합니다. 요청과 응답 모델을 명확히 정의하면 API 계약이 자동으로 문서화됩니다.
마지막으로, 엔드포인트를 자세히 문서화합니다. @app.post() 데코레이터에 여러 매개변수를 추가할 수 있습니다.
summary는 짧은 제목, description은 자세한 설명, response_description은 응답에 대한 설명입니다. tags로 엔드포인트를 그룹화하면 문서가 카테고리별로 정리됩니다.
함수의 docstring도 문서에 표시되므로, 자세한 파라미터 설명을 적을 수 있습니다. 여러분이 이 코드를 사용하면 프론트엔드 개발자나 다른 팀원이 여러분의 API를 쉽게 이해하고 사용할 수 있습니다.
/docs에 가서 엔드포인트를 클릭하고, 값을 입력하고, "Execute"를 누르면 실제 응답을 볼 수 있습니다. Postman이 필요 없습니다.
코드를 수정하면 문서도 자동으로 업데이트되므로, 문서와 실제 동작이 어긋날 일이 없습니다.
실전 팁
💡 response_model을 반드시 지정하세요. 이게 있어야 응답 스키마가 문서에 표시되고, 민감한 필드를 자동으로 제외할 수도 있습니다. response_model_exclude={"password"}처럼 사용합니다.
💡 여러 응답 코드를 문서화할 수 있습니다. @app.post(..., responses={404: {"description": "User not found"}})로 각 상황에 대한 설명을 추가하세요. 에러 케이스도 문서화하면 클라이언트 개발자가 에러 처리를 제대로 할 수 있습니다.
💡 tags를 일관되게 사용하세요. 같은 기능의 엔드포인트는 같은 태그로 그룹화하면 문서를 탐색하기 쉽습니다. 예: "Authentication", "Chat", "Admin"
💡 OpenAPI JSON을 내보낼 수도 있습니다. /openapi.json에 접속하면 전체 스펙을 JSON으로 받을 수 있습니다. 이를 사용해 클라이언트 SDK를 자동 생성하거나, Postman에 import 할 수 있습니다.
10. 환경_변수_관리
시작하며
여러분의 코드에 API 키, 데이터베이스 비밀번호, JWT 시크릿 같은 민감한 정보가 하드코딩되어 있나요? 코드를 GitHub에 올리는 순간 전 세계에 공개되고, 해커들이 그 정보를 악용할 수 있습니다.
실제로 이런 사고가 매일 발생합니다. 또한 개발/스테이징/production 환경마다 다른 설정이 필요합니다.
같은 코드로 세 환경을 운영하려면 설정을 외부에서 주입받아야 합니다. 코드를 수정하지 않고 환경만 바꿔서 배포할 수 있어야 합니다.
그래야 실수로 개발용 API 키를 production에 배포하는 일이 없습니다. 바로 이럴 때 필요한 것이 환경 변수 관리입니다.
설정을 코드에서 분리하고, 안전하게 관리하고, 환경별로 다르게 적용합니다. 12-Factor App의 핵심 원칙입니다.
개요
간단히 말해서, 환경 변수는 운영 체제나 배포 환경에서 제공하는 키-값 쌍으로, 민감한 설정을 코드 밖에서 관리하게 해줍니다. Python의 python-dotenv와 Pydantic의 BaseSettings를 사용하면 환경 변수를 타입 안전하게 로드할 수 있습니다.
.env 파일에 개발 환경 설정을 넣고 .gitignore에 추가하면 GitHub에 올라가지 않습니다. production에서는 Kubernetes Secret, AWS Parameter Store, 또는 호스팅 플랫폼의 환경 변수 설정을 사용합니다.
예를 들어, OpenAI API 키를 환경 변수 OPENAI_API_KEY로 저장하고, 코드에서는 settings.openai_api_key로 접근합니다. 개발자마다 자신의 API 키를 사용할 수 있고, production에서는 팀 계정의 키를 사용합니다.
코드는 전혀 수정하지 않습니다. 핵심 특징은: 1) 민감 정보 보호, 2) 환경별 설정 분리, 3) 타입 검증.
이런 특징들이 보안과 배포 안정성을 크게 향상시킵니다.
코드 예제
from pydantic_settings import BaseSettings
from pydantic import Field
from functools import lru_cache
import os
# 설정 클래스
class Settings(BaseSettings):
# API 설정
openai_api_key: str = Field(..., env="OPENAI_API_KEY")
app_name: str = Field(default="AI Agent API", env="APP_NAME")
environment: str = Field(default="development", env="ENVIRONMENT")
# 서버 설정
host: str = Field(default="0.0.0.0", env="HOST")
port: int = Field(default=8000, env="PORT")
debug: bool = Field(default=False, env="DEBUG")
# 데이터베이스 설정
database_url: str = Field(..., env="DATABASE_URL")
# JWT 설정
jwt_secret: str = Field(..., env="JWT_SECRET")
jwt_algorithm: str = Field(default="HS256", env="JWT_ALGORITHM")
class Config:
# .env 파일에서 로드
env_file = ".env"
env_file_encoding = "utf-8"
case_sensitive = False
# 싱글톤 패턴으로 설정 캐싱
@lru_cache()
def get_settings() -> Settings:
return Settings()
# 사용 예시
settings = get_settings()
print(f"Running {settings.app_name} in {settings.environment} mode")
설명
이것이 하는 일: Pydantic으로 환경 변수를 타입 안전하게 로드하고, 필수 설정이 누락되면 에러를 발생시킵니다. 첫 번째로, BaseSettings를 상속받아 설정 클래스를 만듭니다.
각 필드는 환경 변수와 매핑됩니다. env 매개변수로 환경 변수 이름을 명시적으로 지정할 수 있습니다.
Field(...)는 필수 필드를 의미하고, 환경 변수가 없으면 프로그램이 시작되지 않습니다. 이렇게 하면 production에서 중요한 설정이 누락되어 런타임 에러가 발생하는 것을 방지할 수 있습니다.
그 다음으로, 기본값과 타입을 지정합니다. port: int는 환경 변수의 문자열을 자동으로 정수로 변환합니다.
변환할 수 없는 값이 들어오면 에러가 발생합니다. debug: bool도 마찬가지로, "true", "1", "yes" 같은 값을 자동으로 boolean으로 파싱합니다.
타입 안전성 덕분에 실수로 잘못된 값을 설정해도 조기에 발견할 수 있습니다. 마지막으로, @lru_cache()로 설정을 캐싱합니다.
Settings() 인스턴스 생성은 .env 파일을 읽고 환경 변수를 파싱하는 작업이므로, 매번 하면 비효율적입니다. 한 번 생성하면 그 결과를 메모리에 캐싱해서 재사용합니다.
FastAPI의 의존성 주입과 함께 사용하면 완벽합니다. settings: Settings = Depends(get_settings)처럼 주입받으면 됩니다.
여러분이 이 코드를 사용하면 민감한 정보를 안전하게 관리할 수 있습니다. .env 파일은 절대 git에 올리지 않고, .env.example 파일로 필요한 환경 변수 목록만 공유합니다.
새로운 개발자가 합류하면 .env.example을 복사해서 자신의 값으로 채웁니다. production 배포 시에는 호스팅 플랫폼의 환경 변수 설정에 값을 입력하면 됩니다.
코드는 동일하게 유지됩니다.
실전 팁
💡 .env.example 파일을 git에 포함시키세요. 실제 값 대신 OPENAI_API_KEY=your_key_here 같은 플레이스홀더를 넣어서, 어떤 환경 변수가 필요한지 문서화합니다.
💡 환경별로 다른 파일을 사용할 수 있습니다. .env.development, .env.production처럼 만들고, env_file 매개변수를 환경에 따라 바꿉니다. 또는 ENVIRONMENT 변수로 분기합니다.
💡 Kubernetes나 Docker를 사용한다면 환경 변수를 컨테이너에 주입하세요. .env 파일을 이미지에 포함시키지 말고, 런타임에 주입하는 게 안전합니다. docker run -e OPENAI_API_KEY=...
💡 AWS에서는 Parameter Store나 Secrets Manager를 사용하세요. boto3로 런타임에 시크릿을 가져와서 환경 변수로 설정하면, 코드에 아무것도 노출되지 않습니다. 자동 로테이션도 가능합니다.