如何用Python WebSocket构建问答式Web聊天室？
摘要：0 序 在开发 MCP Tool 时，了解到 基于 HTTP SSE 通信模式下 MCP ServerTool；而与 HTTP SSE 通信模式较为接近的是 WebSocket 通信模式。 为此，对 WebSocket 进行一个简单的试验
0 序
在开发 MCP Tool 时，了解到 基于 HTTP SSE 通信模式下 MCP Server/Tool；而与 HTTP SSE 通信模式较为接近的是 WebSocket 通信模式。
为此，对 WebSocket 进行一个简单的试验、实践。
1 基于 python + websocket 实现简易的Web版聊天室
创建一个基于 WebSocket + Python 的简易 Web 版聊天室。这个实现将包含完整的后端和前端代码，支持多人实时聊天、用户加入/离开通知等功能。
1 源码实现
chat.html - WebSocket 聊天室
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>WebSocket 聊天室</title>
<style>
* {
margin: 0;
padding: 0;
box-sizing: border-box;
}
body {
font-family: 'Segoe UI', Tahoma, Geneva, Verdana, sans-serif;
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
height: 100vh;
display: flex;
justify-content: center;
align-items: center;
}
.chat-container {
width: 90%;
max-width: 800px;
height: 90vh;
background: rgba(255, 255, 255, 0.95);
border-radius: 20px;
box-shadow: 0 20px 60px rgba(0,0,0,0.3);
display: flex;
flex-direction: column;
overflow: hidden;
}
.chat-header {
background: linear-gradient(90deg, #667eea 0%, #764ba2 100%);
color: white;
padding: 20px;
text-align: center;
position: relative;
}
.chat-header h1 {
font-size: 24px;
margin-bottom: 5px;
}
.online-count {
font-size: 14px;
opacity: 0.9;
}
.connection-status {
position: absolute;
right: 20px;
top: 50%;
transform: translateY(-50%);
width: 12px;
height: 12px;
border-radius: 50%;
background: #ff4444;
transition: background 0.3s;
}
.connection-status.connected {
background: #00ff88;
box-shadow: 0 0 10px #00ff88;
}
.chat-messages {
flex: 1;
overflow-y: auto;
padding: 20px;
background: #f8f9fa;
}
.message {
margin-bottom: 15px;
animation: slideIn 0.3s ease-out;
}
@keyframes slideIn {
from {
opacity: 0;
transform: translateY(10px);
}
to {
opacity: 1;
transform: translateY(0);
}
}
.message-header {
display: flex;
align-items: center;
margin-bottom: 5px;
}
.username {
font-weight: bold;
color: #667eea;
margin-right: 10px;
}
.timestamp {
font-size: 12px;
color: #999;
}
.message-content {
background: white;
padding: 12px 16px;
border-radius: 15px;
box-shadow: 0 2px 5px rgba(0,0,0,0.05);
display: inline-block;
max-width: 80%;
word-wrap: break-word;
}
.message.own .message-header {
flex-direction: row-reverse;
}
.message.own .username {
color: #764ba2;
margin-right: 0;
margin-left: 10px;
}
.message.own .message-content {
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
color: white;
margin-left: auto;
}
.system-message {
text-align: center;
color: #888;
font-style: italic;
margin: 10px 0;
font-size: 14px;
}
.typing-indicator {
display: none;
padding: 10px 20px;
color: #999;
font-style: italic;
font-size: 14px;
}
.typing-indicator.active {
display: block;
}
.chat-input-area {
padding: 20px;
background: white;
border-top: 1px solid #eee;
}
.input-wrapper {
display: flex;
gap: 10px;
}
#messageInput {
flex: 1;
padding: 12px 20px;
border: 2px solid #e0e0e0;
border-radius: 25px;
outline: none;
font-size: 16px;
transition: border-color 0.3s;
}
#messageInput:focus {
border-color: #667eea;
}
#sendBtn {
padding: 12px 30px;
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
color: white;
border: none;
border-radius: 25px;
cursor: pointer;
font-size: 16px;
transition: transform 0.2s, box-shadow 0.2s;
}
#sendBtn:hover {
transform: translateY(-2px);
box-shadow: 0 5px 15px rgba(102, 126, 234, 0.4);
}
#sendBtn:active {
transform: translateY(0);
}
.login-modal {
position: fixed;
top: 0;
left: 0;
width: 100%;
height: 100%;
background: rgba(0,0,0,0.8);
display: flex;
justify-content: center;
align-items: center;
z-index: 1000;
}
.login-box {
background: white;
padding: 40px;
border-radius: 20px;
text-align: center;
animation: popIn 0.3s ease-out;
}
@keyframes popIn {
from {
transform: scale(0.8);
opacity: 0;
}
to {
transform: scale(1);
opacity: 1;
}
}
.login-box h2 {
margin-bottom: 20px;
color: #333;
}
#usernameInput {
width: 100%;
padding: 15px;
margin-bottom: 20px;
border: 2px solid #e0e0e0;
border-radius: 10px;
font-size: 16px;
outline: none;
}
#usernameInput:focus {
border-color: #667eea;
}
#joinBtn {
width: 100%;
padding: 15px;
background: linear-gradient(135deg, #667eea 0%, #764ba2 100%);
color: white;
border: none;
border-radius: 10px;
font-size: 16px;
cursor: pointer;
transition: transform 0.2s;
}
#joinBtn:hover {
transform: scale(1.05);
}
.hidden {
display: none !important;
}
/* 滚动条样式 */
.chat-messages::-webkit-scrollbar {
width: 8px;
}
.chat-messages::-webkit-scrollbar-track {
background: #f1f1f1;
}
.chat-messages::-webkit-scrollbar-thumb {
background: #888;
border-radius: 4px;
}
.chat-messages::-webkit-scrollbar-thumb:hover {
background: #555;
}
</style>
</head>
<body>
<!-- 登录界面 -->
<div id="loginModal" class="login-modal">
<div class="login-box">
<h2>🚀 加入聊天室</h2>
<input type="text" id="usernameInput" placeholder="请输入你的昵称" maxlength="20">
<button id="joinBtn">进入聊天室</button>
</div>
</div>
<!-- 聊天界面 -->
<div class="chat-container hidden" id="chatContainer">
<div class="chat-header">
<h1>💬 WebSocket 聊天室</h1>
<div class="online-count">在线人数: <span id="onlineCount">0</span></div>
<div class="connection-status" id="connectionStatus"></div>
</div>
<div class="chat-messages" id="chatMessages">
<!-- 消息将在这里动态添加 -->
</div>
<div class="typing-indicator" id="typingIndicator">
有人正在输入...
</div>
<div class="chat-input-area">
<div class="input-wrapper">
<input type="text" id="messageInput" placeholder="输入消息..." maxlength="500">
<button id="sendBtn">发送</button>
</div>
</div>
</div>
<script>
let ws;
let username;
let reconnectAttempts = 0;
const maxReconnectAttempts = 5;
// DOM 元素
const loginModal = document.getElementById('loginModal');
const chatContainer = document.getElementById('chatContainer');
const usernameInput = document.getElementById('usernameInput');
const joinBtn = document.getElementById('joinBtn');
const messageInput = document.getElementById('messageInput');
const sendBtn = document.getElementById('sendBtn');
const chatMessages = document.getElementById('chatMessages');
const onlineCount = document.getElementById('onlineCount');
const connectionStatus = document.getElementById('connectionStatus');
// 加入聊天室
joinBtn.addEventListener('click', joinChat);
usernameInput.addEventListener('keypress', (e) => {
if (e.key === 'Enter') joinChat();
});
function joinChat() {
username = usernameInput.value.trim();
if (!username) {
alert('请输入昵称');
return;
}
loginModal.classList.add('hidden');
chatContainer.classList.remove('hidden');
connectWebSocket();
}
// 连接 WebSocket
function connectWebSocket() {
// 使用当前主机和端口，或指定服务器地址
const wsUrl = `ws://${window.location.hostname}:8765`;
ws = new WebSocket(wsUrl);
ws.onopen = () => {
console.log('WebSocket 连接成功');
connectionStatus.classList.add('connected');
reconnectAttempts = 0;
// 发送加入消息
ws.send(JSON.stringify({
type: 'join',
username: username,
timestamp: new Date().toISOString()
}));
};
ws.onmessage = (event) => {
const data = JSON.parse(event.data);
handleMessage(data);
};
ws.onclose = () => {
console.log('WebSocket 连接关闭');
connectionStatus.classList.remove('connected');
attemptReconnect();
};
ws.onerror = (error) => {
console.error('WebSocket 错误:', error);
};
}
// 重连机制
function attemptReconnect() {
if (reconnectAttempts < maxReconnectAttempts) {
reconnectAttempts++;
console.log(`尝试重连... (${reconnectAttempts}/${maxReconnectAttempts})`);
setTimeout(connectWebSocket, 3000);
} else {
addSystemMessage('连接已断开，请刷新页面重试');
}
}
// 处理收到的消息
function handleMessage(data) {
switch(data.type) {
case 'message':
addMessage(data.username, data.content, data.timestamp, data.username === username);
break;
case 'system':
addSystemMessage(data.content);
break;
case 'user_count':
onlineCount.textContent = data.count;
break;
case 'history':
// 加载历史消息
data.messages.forEach(msg => {
addMessage(msg.username, msg.content, msg.timestamp, msg.username === username);
});
break;
}
}
// 添加普通消息
function addMessage(user, content, timestamp, isOwn = false) {
const messageDiv = document.createElement('div');
messageDiv.className = `message ${isOwn ? 'own' : ''}`;
const time = new Date(timestamp).toLocaleTimeString('zh-CN', {
hour: '2-digit',
minute: '2-digit'
});
messageDiv.innerHTML = `
<div class="message-header">
<span class="username">${escapeHtml(user)}</span>
<span class="timestamp">${time}</span>
</div>
<div class="message-content">${escapeHtml(content)}</div>
`;
chatMessages.appendChild(messageDiv);
scrollToBottom();
}
// 添加系统消息
function addSystemMessage(content) {
const div = document.createElement('div');
div.className = 'system-message';
div.textContent = content;
chatMessages.appendChild(div);
scrollToBottom();
}
// 滚动到底部
function scrollToBottom() {
chatMessages.scrollTop = chatMessages.scrollHeight;
}
// HTML 转义防止 XSS
function escapeHtml(text) {
const div = document.createElement('div');
div.textContent = text;
return div.innerHTML;
}
// 发送消息
function sendMessage() {
const content = messageInput.value.trim();
if (!content || !ws || ws.readyState !== WebSocket.OPEN) return;
ws.send(JSON.stringify({
type: 'message',
username: username,
content: content,
timestamp: new Date().toISOString()
}));
messageInput.value = '';
messageInput.focus();
}
sendBtn.addEventListener('click', sendMessage);
messageInput.addEventListener('keypress', (e) => {
if (e.key === 'Enter') sendMessage();
});
// 页面关闭时发送离开消息
window.addEventListener('beforeunload', () => {
if (ws && ws.readyState === WebSocket.OPEN) {
ws.send(JSON.stringify({
type: 'leave',
username: username
}));
}
});
</script>
</body>
</html>
server.py - WebSocket 聊天室服务端
# server.py - WebSocket 聊天室服务端
import asyncio
import websockets
import json
import logging
from datetime import datetime
from collections import defaultdict
# 配置日志
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class ChatServer:
def __init__(self):
# 存储所有连接的客户端 {websocket: username}
self.clients = {}
# 消息历史记录（保留最近100条）
self.message_history = []
self.max_history = 100
async def register(self, websocket, username):
"""注册新客户端"""
self.clients[websocket] = username
logger.info(f"用户 {username} 加入聊天室，当前在线: {len(self.clients)}人")
# 发送历史消息给新用户
await self.send_history(websocket)
# 广播用户加入消息
await self.broadcast({
"type": "system",
"content": f"👋 {username} 加入了聊天室"
})
# 更新在线人数
await self.broadcast_user_count()
async def unregister(self, websocket):
"""注销客户端"""
if websocket in self.clients:
username = self.clients[websocket]
del self.clients[websocket]
logger.info(f"用户 {username} 离开聊天室，当前在线: {len(self.clients)}人")
# 广播用户离开消息
await self.broadcast({
"type": "system",
"content": f"👋 {username} 离开了聊天室"
})
# 更新在线人数
await self.broadcast_user_count()
async def send_history(self, websocket):
"""发送历史消息给新连接的用户"""
if self.message_history:
await websocket.send(json.dumps({
"type": "history",
"messages": self.message_history
}))
async def broadcast_user_count(self):
"""广播当前在线人数"""
await self.broadcast({
"type": "user_count",
"count": len(self.clients)
})
async def broadcast(self, message, exclude=None):
"""广播消息给所有客户端"""
if self.clients:
message_str = json.dumps(message)
# 创建发送任务列表
tasks = []
for client in self.clients:
if client != exclude and client.open:
tasks.append(asyncio.create_task(self.safe_send(client, message_str)))
if tasks:
await asyncio.gather(*tasks, return_exceptions=True)
async def safe_send(self, websocket, message):
"""安全发送消息（捕获异常）"""
try:
await websocket.send(message)
except Exception as e:
logger.error(f"发送消息失败: {e}")
async def handle_message(self, websocket, data):
"""处理收到的消息"""
msg_type = data.get("type")
if msg_type == "join":
username = data.get("username")
await self.register(websocket, username)
elif msg_type == "message":
username = data.get("username")
content = data.get("content", "").strip()
if not content:
return
# 保存到历史记录
message_data = {
"username": username,
"content": content,
"timestamp": data.get("timestamp", datetime.now().isoformat())
}
self.message_history.append(message_data)
# 限制历史记录长度
if len(self.message_history) > self.max_history:
self.message_history.pop(0)
# 广播消息
await self.broadcast({
"type": "message",
**message_data
})
elif msg_type == "leave":
await self.unregister(websocket)
async def handler(self, websocket, path):
"""WebSocket 连接处理器"""
try:
async for message in websocket:
try:
data = json.loads(message)
await self.handle_message(websocket, data)
except json.JSONDecodeError:
logger.error("收到无效的 JSON 数据")
except Exception as e:
logger.error(f"处理消息时出错: {e}")
except websockets.exceptions.ConnectionClosed:
logger.info("客户端连接关闭")
finally:
await self.unregister(websocket)
# 创建服务器实例
chat_server = ChatServer()
async def main():
"""启动 WebSocket 服务器"""
host = "0.0.0.0"
port = 8765
logger.info(f"启动聊天室服务器于 ws://{host}:{port}")
async with websockets.serve(chat_server.handler, host, port):
await asyncio.Future() # 永久运行
if __name__ == "__main__":
try:
asyncio.run(main())
except KeyboardInterrupt:
logger.info("服务器已停止")
2 使用说明
2.1 安装依赖
pip install websockets
2.2 启动服务端
python server.py
启动运行日志
(ai-env) PS D:\Workspace\xxx\xxx\websocket_chat> python server.py
2026-03-27 09:31:39,782 - INFO - 启动聊天室服务器于 ws://0.0.0.0:8765
2026-03-27 09:31:39,876 - INFO - server listening on 0.0.0.0:8765
2026-03-27 09:33:49,209 - INFO - connection open
2026-03-27 09:33:49,212 - INFO - 用户 johnny 加入聊天室，当前在线: 1人
2026-03-27 09:34:30,394 - INFO - connection open
2026-03-27 09:34:30,399 - INFO - 用户 jack 加入聊天室，当前在线: 2人
2026-03-27 09:41:05,575 - INFO - server closing
2026-03-27 09:41:05,577 - INFO - 用户 jack 离开聊天室，当前在线: 1人
2026-03-27 09:41:05,578 - INFO - connection closed
2026-03-27 09:41:05,578 - INFO - 用户 johnny 离开聊天室，当前在线: 0人
2026-03-27 09:41:05,579 - INFO - connection closed
2026-03-27 09:41:05,579 - INFO - server closed
2026-03-27 09:41:05,580 - INFO - 服务器已停止
2.3 运行客户端
直接在浏览器中打开 HTML 文件，或 使用 Python 简单 HTTP 服务器：
# 在 HTML 文件所在目录运行
python -m http.server 8080
# 然后访问 http://localhost:8080
启动/运行日志:
$ python -m http.server 8080
Serving HTTP on :: port 8080 (http://[::]:8080/) ...
::1 - - [27/Mar/2026 09:33:27] "GET / HTTP/1.1" 200 -
::1 - - [27/Mar/2026 09:33:27] code 404, message File not found
::1 - - [27/Mar/2026 09:33:27] "GET /favicon.ico HTTP/1.1" 404 -
::1 - - [27/Mar/2026 09:33:30] "GET /chat.html HTTP/1.1" 200 -
::1 - - [27/Mar/2026 09:34:08] "GET /chat.html HTTP/1.1" 304 -
2.4 体验/使用
用户们可通过 http://localhost:8080/chat.html 先后进入聊天室
3 功能特性
功能
说明
实时通信
基于 WebSocket 的全双工通信
多人聊天
支持多个客户端同时在线
用户系统
昵称登录，加入/离开通知
消息历史
新用户加入可看到最近100条消息
在线人数
实时显示当前在线用户数量
重连机制
客户端自动重连（最多5次）
XSS 防护
自动转义 HTML 特殊字符
响应式设计
适配移动端和桌面端
4 架构说明
┌─────────────┐ WebSocket ┌──────────────┐
│ 浏览器 │ ◄──────────────────────► │ Python 服务端 │
│ (HTML/JS) │ ws://localhost:8765 │ websockets │
└─────────────┘ └──────────────┘
▲ │
│ │
└────────────── 广播消息 ◄────────────────────┘
这个实现非常轻量，核心代码不到 200 行，但包含了生产环境所需的基本功能如异常处理、心跳检测（通过自动重连）、历史消息等。你可以基于此扩展更多功能如私聊、表情包、文件传输等。
Z FAQ for WebSocket/聊天室
Y 推荐文献
MCP Server/Tool 开发指南 - 博客园/数据知音
基于 HTTP SSE 模式的 MCP Server/Tool
SSE(Server Send Events) ：服务器 => 浏览器的消息推送解决方案 - 博客园/千千寰宇
HTTP SSE 通信模式（半双工） vs WebSocket 通信模式（全双工）
X 参考文献