3 SERP Data Streaming Strategies Compared: SSE / WebSocket / Polling

Background

Frontends need SERP data to update "in realtime", three mainstream options: SSE / WebSocket / Polling. Wrong choice means high implementation complexity + poor latency.

1. Polling

import time
import requests

def poll_serp():
    while True:
        try:
            r = requests.post(
                "https://api.serpbase.dev/google/search",
                headers={"X-API-Key": "sk_xxx"},
                json={"q": "best serp api", "gl": "us", "num": 5},
                timeout=10,
            )
            data = r.json()
            handle_change(data)
        except Exception as e:
            print(f"Error: {e}")

        time.sleep(60)  # poll every minute

Pros: Simple, HTTP standard, easy to debug
Cons: High latency (avg 30s), wastes requests (90% no change)

2. SSE (Server-Sent Events)

from flask import Flask, Response, stream_with_context
import requests
import json
import time

app = Flask(__name__)

def stream_serp():
    last_data = None
    while True:
        r = requests.post(
            "https://api.serpbase.dev/google/search",
            headers={"X-API-Key": "sk_xxx"},
            json={"q": "best serp api", "gl": "us", "num": 5},
            timeout=10,
        )
        data = r.json()

        if data != last_data:
            yield f"data: {json.dumps(data)}\n\n"
            last_data = data

        time.sleep(5)

@app.route("/stream")
def stream():
    return Response(stream_with_context(stream_serp()), mimetype="text/event-stream")

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=5000)

Pros: Low latency (second-level), HTTP standard, auto-reconnect
Cons: One-way (server → client), no bidirectional communication

3. WebSocket

from flask_socketio import SocketIO, emit
import requests
import json

app = Flask(__name__)
socketio = SocketIO(app, cors_allowed_origins="*")

@socketio.on("connect")
def handle_connect():
    print("Client connected")

@socketio.on("disconnect")
def handle_disconnect():
    print("Client disconnected")

@socketio.on("subscribe")
def handle_subscribe(data):
    query = data.get("q", "best serp api")

    while True:
        r = requests.post(
            "https://api.serpbase.dev/google/search",
            headers={"X-API-Key": "sk_xxx"},
            json={"q": query, "gl": "us", "num": 5},
            timeout=10,
        )
        data = r.json()

        emit("serp_update", data)
        socketio.sleep(5)

if __name__ == "__main__":
    socketio.run(app, host="0.0.0.0", port=5000)

Pros: Bidirectional, realtime, state-preserving
Cons: Complex, needs WebSocket server, hard to debug

4. 3 Strategies Compared

Dimension	Polling	SSE	WebSocket
Realtime	30s+ avg	Second-level	Realtime
Complexity	★	★★	★★★
HTTP standard	✓	✓	✗
Bidirectional	✗	✗	✓
Browser support	✓	✓ (EventSource)	✓
Firewall	✓ (HTTP)	✓ (HTTP)	△ (ws://)
Server pressure	High (polling)	Low	Medium
Best for	Simple data	Server push	Real-time collaboration

5. Selection

Scenario	Recommended
Simple data (tolerate 30s delay)	Polling
Server push (realtime rank)	SSE
Client frequent query + server push	WebSocket
AI Agent tool streaming	SSE
Real-time collaboration	WebSocket

6. Real Code (SSE + Client Integration)

Server (FastAPI)

from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import requests
import json
import asyncio

app = FastAPI()

async def serp_event_stream(query):
    last_data = None
    while True:
        r = requests.post(
            "https://api.serpbase.dev/google/search",
            headers={"X-API-Key": "sk_xxx"},
            json={"q": query, "gl": "us", "num": 5},
            timeout=10,
        )
        data = r.json()

        if data != last_data:
            yield f"event: change\ndata: {json.dumps(data)}\n\n"
            last_data = data

        await asyncio.sleep(5)

@app.get("/stream")
async def stream(query: str = "best serp api"):
    return StreamingResponse(
        serp_event_stream(query),
        media_type="text/event-stream",
    )

Client (JavaScript)

const source = new EventSource(`/stream?q=${query}`);
source.addEventListener("change", (e) => {
  const data = JSON.parse(e.data);
  console.log("Updated SERP:", data);
});

7. 5 Engineering Details

Detail 1: Auto-Reconnect (SSE)

// EventSource auto-reconnect
const source = new EventSource("/stream");
source.addEventListener("open", () => console.log("Connected"));
source.addEventListener("error", () => console.log("Reconnecting..."));

Detail 2: Heartbeat

async def stream_with_heartbeat():
    while True:
        yield ": heartbeat\n\n"  # comment line, client ignores
        await asyncio.sleep(15)

Detail 3: Last-Event-ID Resume

async def stream_with_id():
    event_id = 0
    while True:
        event_id += 1
        yield f"id: {event_id}\ndata: {data}\n\n"
        await asyncio.sleep(5)

Detail 4: WebSocket Auth

@socketio.on("connect")
def handle_connect(auth):
    token = auth.get("token")
    if not verify_token(token):
        return False

Detail 5: Rate Limiting

@socketio.on("subscribe")
@rate_limit(per_minute=10)
def handle_subscribe(data):
    pass

8. serpbase 集成优势

serpbase 1.4s P50 + auto-refund, all 3 streaming options work:

Option	Use Case	Latency
Polling	Simple monitoring	30s avg
SSE	AI agent realtime push	1s + 5s polling
WebSocket	Collaboration tools	1s realtime

9. Real Data (My Project)

Option	Users	Monthly requests	Monthly cost
Polling	Monitoring	100,000	$3.5
SSE	AI assistant	50,000	$1.5
WebSocket	Collaboration tool	80,000	$2.5

SSE is the most common option, low latency + low complexity.

Summary

3 streaming strategy selection:

• Polling: Simple, low cost
• SSE: Realtime push, preferred
• WebSocket: Bidirectional, complex scenarios

serpbase + SSE 5-step realtime SERP push system, 5-minute deploy, monthly cost $1.5, suits 80% of scenarios.