WebSockets مقابل الأحداث المرسلة من الخادم 2026: شرح الويب في الوقت الفعلي

# FastAPI WebSocket server
from fastapi import FastAPI, WebSocket

app = FastAPI()
connected = []

@app.websocket('/ws')
async def websocket_endpoint(ws: WebSocket):
    await ws.accept()
    connected.append(ws)
    try:
        while True:
            data = await ws.receive_text()
            # Broadcast to all clients
            for client in connected:
                await client.send_text(f'Message: {data}')
    except:
        connected.remove(ws)

// Browser WebSocket client
const ws = new WebSocket('wss://example.com/ws');

ws.onopen = () => console.log('Connected');
ws.onmessage = (e) => console.log('Received:', e.data);
ws.onerror  = (e) => console.error('Error:', e);
ws.onclose  = () => console.log('Disconnected');

// Send message
ws.send('Hello server!');

# FastAPI SSE server
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import asyncio

app = FastAPI()

@app.get('/stream')
async def stream():
    async def event_generator():
        for i in range(10):
            yield f'data: Message {i}\n\n'
            await asyncio.sleep(1)
    return StreamingResponse(
        event_generator(),
        media_type='text/event-stream'
    )

// Browser SSE client
const evtSource = new EventSource('/stream');

evtSource.onmessage = (e) => {
  console.log('Received:', e.data);
};

evtSource.onerror = (e) => {
  console.error('Error:', e);
  // EventSource auto-reconnects after error
};

// Close when done
evtSource.close();

import anthropic
from fastapi.responses import StreamingResponse

client = anthropic.Anthropic()

@app.post('/ask')
async def ask(question: str):
    async def generate():
        with client.messages.stream(
            model='claude-sonnet-4-5',
            max_tokens=1024,
            messages=[{'role': 'user', 'content': question}]
        ) as stream:
            for text in stream.text_stream:
                yield f'data: {text}\n\n'
        yield 'data: [DONE]\n\n'

    return StreamingResponse(generate(), media_type='text/event-stream')