Chạy FastAPI và Socket.IO chung một process ASGI với socketio.ASGIApp

Chạy FastAPI và Socket.IO chung một process ASGI với socketio.ASGIApp

Khi xây dựng hệ thống helpdesk có tính năng realtime — notification khi có ticket mới, trạng thái reply cập nhật ngay không cần F5 — câu hỏi đầu tiên thường là: FastAPI và Socket.IO chạy chung hay riêng?

Câu trả lời ngắn: chạy chung được, và đây là cách làm đúng.

Vấn đề với 2 server riêng

Cách phổ biến ban đầu là chạy FastAPI trên port 8001 và Socket.IO server riêng trên port 8002. Về lý thuyết thì hoạt động, nhưng trên thực tế sinh ra một loạt rắc rối:

CORS nhân đôi. Frontend phải kết nối 2 origin khác nhau. Mỗi khi thêm domain hay thay đổi cấu hình, phải sửa ở 2 chỗ — dễ bị lệch nhau gây lỗi khó debug.

Deploy phức tạp hơn. Docker Compose phải quản lý 2 container riêng (hoặc 2 process trong 1 container). Nginx phải proxy 2 upstream. CI/CD phải build và restart 2 service. Nhân đôi điểm có thể fail.

Không giao tiếp được trực tiếp. Khi REST handler xử lý xong một action — ví dụ lưu reply vào DB — nó muốn push notification realtime ngay lập tức đến client liên quan. Nếu Socket.IO là process riêng, REST handler phải gọi qua Redis Pub/Sub, RabbitMQ, hoặc HTTP nội bộ. Thêm một tầng trung gian, thêm độ trễ, thêm thứ có thể hỏng.

Tất cả những vấn đề này biến mất nếu chạy chung một process.

Giải pháp: ASGI composability

ASGI (Asynchronous Server Gateway Interface) — chuẩn interface giữa Python async web framework và server như uvicorn — được thiết kế để có thể compose: ghép nhiều app ASGI vào nhau trong cùng một process.

Thư viện python-socketio cung cấp socketio.ASGIApp, một ASGI wrapper thực hiện đúng việc này: nhận FastAPI app làm other_asgi_app, tự route /socket.io/* đến Socket.IO engine, còn lại chuyển thẳng đến FastAPI. Một uvicorn process phục vụ cả hai.

Code

Cài thư viện cần thiết:

pip install fastapi uvicorn python-socketio

Cấu trúc main.py:

import socketio
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

# 1. Tạo Socket.IO server — async_mode='asgi' là bắt buộc
sio = socketio.AsyncServer(
    async_mode='asgi',
    cors_allowed_origins=['http://localhost:5173', 'https://yourdomain.com'],
)

# 2. FastAPI app bình thường — thêm CORS riêng cho REST
fastapi_app = FastAPI()
fastapi_app.add_middleware(
    CORSMiddleware,
    allow_origins=['http://localhost:5173', 'https://yourdomain.com'],
    allow_methods=['*'],
    allow_headers=['*'],
)

# 3. Wrap FastAPI trong ASGIApp — đây là app uvicorn sẽ chạy
app = socketio.ASGIApp(sio, other_asgi_app=fastapi_app)

Chạy server:

uvicorn main:app --host 0.0.0.0 --port 8001
# Trỏ vào `app` (ASGIApp), KHÔNG phải `fastapi_app`

Một lỗi hay gặp là trỏ uvicorn vào fastapi_app — lúc đó Socket.IO không hoạt động vì không có gì xử lý /socket.io/ requests.

Lợi ích thực tế: REST gọi Socket.IO trực tiếp

Đây là điểm khác biệt lớn nhất. Khi REST handler và Socket.IO cùng process, chúng dùng chung object sio — gọi trực tiếp không cần trung gian:

@fastapi_app.post("/tickets/{ticket_id}/reply")
async def post_reply(ticket_id: int, body: ReplyCreate, db: Session = Depends(get_db)):
    # Lưu reply vào database
    reply = save_reply_to_db(db, ticket_id, body)

    # Push realtime ngay lập tức — cùng process, không cần queue
    await sio.emit(
        "new_reply",
        {"ticket_id": ticket_id, "reply": reply.dict()},
        room=f"ticket_{ticket_id}",
    )

    return {"ok": True, "reply_id": reply.id}

Client tham gia room khi mở ticket:

@sio.event
async def join_ticket(sid, data):
    ticket_id = data["ticket_id"]
    await sio.enter_room(sid, f"ticket_{ticket_id}")

Khi có reply mới, tất cả client đang xem ticket đó nhận được event ngay — không độ trễ từ message queue, không polling.

Lưu ý khi triển khai

async_mode='asgi' là bắt buộc. Nếu dùng gevent hoặc eventlet thì không tương thích với uvicorn async. Luôn dùng asgi khi kết hợp với FastAPI.

CORS phải set ở cả hai chỗ. Socket.IO và FastAPI quản lý CORS riêng biệt — cors_allowed_origins trong AsyncServer cho WebSocket, CORSMiddleware trong FastAPI cho REST. Thiếu một trong hai sẽ bị block ở browser.

Rooms để push đúng người. Đừng emit broadcast đến toàn bộ client. Dùng room theo user_id hoặc ticket_id để chỉ push đến người liên quan:

# Push cho user cụ thể
await sio.emit("notification", data, room=f"user_{user_id}")

# Push cho tất cả người đang xem ticket
await sio.emit("new_reply", data, room=f"ticket_{ticket_id}")

Authentication Socket.IO. Dùng connect event để xác thực JWT trước khi cho phép kết nối:

@sio.event
async def connect(sid, environ, auth):
    token = (auth or {}).get("token")
    if not verify_jwt(token):
        raise socketio.exceptions.ConnectionRefusedError("Unauthorized")

Tóm lại: socketio.ASGIApp là giải pháp gọn nhất để kết hợp FastAPI và Socket.IO — một process, một port, một deployment unit. REST handler push realtime trực tiếp không qua trung gian. Cấu hình thêm chút CORS và nhớ trỏ uvicorn đúng vào app là chạy được ngay.