WebSocket 실시간 채팅: FastAPI + SvelteKit 구현 - 코드픽 블로그
CodePick
로그인
WebSocket 실시간 채팅: FastAPI + SvelteKit 구현
기술 가이드

WebSocket 실시간 채팅: FastAPI + SvelteKit 구현

2026년 2월 28일 35 views by 코드벤터
#WebSocket#실시간#SvelteKit

WebSocket 실시간 채팅: FastAPI + SvelteKit 완벽 구현 가이드

실시간 채팅은 현대 웹 애플리케이션에서 빠질 수 없는 핵심 기능입니다. 전통적인 HTTP 요청-응답 방식으로는 실시간성을 구현하기 어렵고, 폴링(Polling) 방식은 서버 부하를 높입니다. 이 문제를 해결하는 것이 바로 WebSocket 프로토콜입니다.

이 글에서는 Python 비동기 프레임워크인 FastAPI를 백엔드로, 모던 풀스택 프레임워크인 SvelteKit을 프론트엔드로 사용하여 실시간 채팅 애플리케이션을 단계별로 구현하는 방법을 알아봅니다.

WebSocket이란?

WebSocket은 단일 TCP 연결을 통해 전이중(full-duplex) 통신 채널을 제공하는 프로토콜입니다. HTTP 핸드셰이크를 통해 연결을 업그레이드한 후, 서버와 클라이언트가 서로 자유롭게 데이터를 주고받을 수 있습니다.

HTTP vs WebSocket 비교

항목HTTPWebSocket
연결 방식요청마다 새로 연결한 번 연결 후 유지
통신 방향단방향 (클라이언트 -> 서버)양방향
오버헤드헤더 반복 전송최초 핸드셰이크 이후 최소화
실시간성폴링 필요즉시 푸시 가능
사용 사례일반 웹 요청채팅, 게임, 주식 시세

개발 환경 준비

필수 패키지 설치

bash 
# 백엔드 패키지 설치
pip3 install fastapi uvicorn websockets

# 프론트엔드 SvelteKit 프로젝트 생성
npx sv create my-chat-app
cd my-chat-app
npm install

FastAPI 백엔드 구현

ConnectionManager 클래스

여러 클라이언트의 WebSocket 연결을 효율적으로 관리하기 위해 ConnectionManager 클래스를 설계합니다.

python 
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from fastapi.middleware.cors import CORSMiddleware
from typing import List, Dict
import json

app = FastAPI(title="실시간 채팅 서버")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:5173"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

class ConnectionManager:
    def __init__(self):
        self.active_connections: Dict[str, WebSocket] = {}

    async def connect(self, websocket: WebSocket, client_id: str):
        await websocket.accept()
        self.active_connections[client_id] = websocket
        await self.broadcast_system(f"{client_id}님이 입장하셨습니다.")

    def disconnect(self, client_id: str):
        if client_id in self.active_connections:
            del self.active_connections[client_id]

    async def broadcast(self, message: str):
        for connection in self.active_connections.values():
            await connection.send_text(message)

    async def broadcast_system(self, text: str):
        payload = json.dumps({"type": "system", "message": text}, ensure_ascii=False)
        await self.broadcast(payload)

manager = ConnectionManager()

@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: str):
    await manager.connect(websocket, client_id)
    try:
        while True:
            data = await websocket.receive_text()
            payload = json.dumps({
                "type": "message",
                "sender": client_id,
                "message": data
            }, ensure_ascii=False)
            await manager.broadcast(payload)
    except WebSocketDisconnect:
        manager.disconnect(client_id)
        await manager.broadcast_system(f"{client_id}님이 퇴장하셨습니다.")

서버 실행

bash 
uvicorn main:app --reload --port 8000

SvelteKit 프론트엔드 구현

채팅 페이지 컴포넌트

src/routes/+page.svelte 파일에 채팅 UI와 WebSocket 로직을 구현합니다.

svelte 
<script lang="ts">
  import { onMount, onDestroy } from "svelte";

  let ws: WebSocket | null = null;
  let messageInput = "";
  let messages: { type: string; sender?: string; message: string }[] = [];
  let username = "User" + Math.floor(Math.random() * 1000);
  let isConnected = false;

  onMount(() => {
    ws = new WebSocket(`ws://localhost:8000/ws/${username}`);

    ws.onopen = () => {
      isConnected = true;
    };

    ws.onmessage = (event) => {
      const data = JSON.parse(event.data);
      messages = [...messages, data];
    };

    ws.onclose = () => {
      isConnected = false;
    };
  });

  onDestroy(() => {
    if (ws) ws.close();
  });

  function sendMessage() {
    if (ws && ws.readyState === WebSocket.OPEN && messageInput.trim()) {
      ws.send(messageInput);
      messageInput = "";
    }
  }

  function handleKeydown(e: KeyboardEvent) {
    if (e.key === "Enter") sendMessage();
  }
</script>

<div class="chat-wrapper">
  <div class="chat-header">
    <h1>실시간 채팅</h1>
    <span class="status" class:connected={isConnected}>
      {isConnected ? "연결됨" : "연결 끊김"}
    </span>
  </div>

  <div class="message-list">
    {#each messages as msg}
      {#if msg.type === "system"}
        <div class="system-msg">{msg.message}</div>
      {:else}
        <div class="chat-msg" class:mine={msg.sender === username}>
          <strong>{msg.sender}</strong>
          <p>{msg.message}</p>
        </div>
      {/if}
    {/each}
  </div>

  <div class="input-row">
    <input
      type="text"
      bind:value={messageInput}
      on:keydown={handleKeydown}
      placeholder="메시지를 입력하세요..."
    />
    <button on:click={sendMessage}>전송</button>
  </div>
</div>

고급 기능 확장

채팅방(Room) 기능 추가

실제 서비스에서는 여러 채팅방이 필요합니다. 백엔드를 확장하여 룸 기반 브로드캐스트를 구현할 수 있습니다.

python 
class RoomManager:
    def __init__(self):
        self.rooms: Dict[str, Dict[str, WebSocket]] = {}

    async def join_room(self, room_id: str, client_id: str, websocket: WebSocket):
        await websocket.accept()
        if room_id not in self.rooms:
            self.rooms[room_id] = {}
        self.rooms[room_id][client_id] = websocket

    async def broadcast_to_room(self, room_id: str, message: str):
        if room_id in self.rooms:
            for ws in self.rooms[room_id].values():
                await ws.send_text(message)

room_manager = RoomManager()

@app.websocket("/ws/{room_id}/{client_id}")
async def room_websocket(websocket: WebSocket, room_id: str, client_id: str):
    await room_manager.join_room(room_id, client_id, websocket)
    try:
        while True:
            data = await websocket.receive_text()
            payload = json.dumps({"room": room_id, "sender": client_id, "message": data})
            await room_manager.broadcast_to_room(room_id, payload)
    except WebSocketDisconnect:
        pass

배포 시 고려사항

프로덕션 환경에서는 다음 사항을 반드시 챙겨야 합니다.

  • Nginx 설정: proxy_pass, Upgrade, Connection 헤더를 올바르게 설정해야 WebSocket 연결이 유지됩니다.
  • SSL/TLS: wss:// 프로토콜을 사용하여 보안 연결을 유지하세요.
  • 인증: JWT 토큰을 WebSocket 연결 쿼리 파라미터나 첫 메시지로 전달하여 인증을 처리합니다.
  • 스케일 아웃: 여러 서버 인스턴스를 운영할 경우 Redis Pub/Sub를 활용하여 인스턴스 간 메시지를 동기화해야 합니다.

마치며

WebSocket과 FastAPI, SvelteKit의 조합은 실시간 채팅 구현에 매우 강력한 선택지입니다. FastAPI의 비동기 처리 덕분에 수천 개의 동시 연결도 효율적으로 처리할 수 있으며, SvelteKit의 반응형 상태 관리는 UI 업데이트를 간결하게 만들어줍니다.

코드벤터는 이처럼 최신 기술 스택을 활용하여 실용적인 솔루션을 만들어가는 개발 문화를 지향합니다. WebSocket 기반 실시간 기능이 필요한 프로젝트가 있다면 언제든지 코드벤처와 함께 설계하고 구현해보세요. 더 나은 사용자 경험은 올바른 기술 선택에서 시작됩니다.

개발 의뢰 상담

AI 서비스나 플랫폼 개발을
고민 중이신가요?

CodePick에서는
기획 → 개발 → 운영까지 함께합니다.
아이디어만 있어도 상담 가능합니다.

무료 개발 상담 신청 개발 비용 확인하기

CodeVenter 개발팀이 직접 담당 · 1~2 영업일 내 회신

✓ 스타트업 MVP 개발✓ AI 서비스 개발✓ 웹 플랫폼 개발✓ 기업 시스템 구축✓ 모바일 앱 개발
블로그 목록 프로젝트 문의

AI Development Studio

코드픽 by 코드벤터

  • 대표: 윤승환 · 사업자등록번호: 121-57-64983
  • 대구광역시 중구 국채보상로 586, 16층 · info@codeventer.com
회사소개 무료 상담 →

© 2025 코드벤터. All rights reserved.