FastAPI SSE Events

Server-Sent Events (SSE) notifications for FastAPI using Redis Pub/Sub

Python 3.10+ PyPI version License: MIT Tests codecov

Add real-time "refresh-less" updates to your FastAPI REST API in minutes.

Perfect for collaborative tools (CRMs, project management, dashboards) where multiple users need to see updates instantly without manual page refreshes.

📚 Documentation

Table of Contents

Overview

FastAPI SSE Events enables near real-time notifications for REST-based applications without the complexity of WebSockets. Clients subscribe to Server-Sent Event streams for specific topics, receive lightweight notifications when data changes, then refresh data via existing REST endpoints.

Key Features

  • 🚀 Simple Integration - Add SSE with 3 lines of code, minimal changes to existing endpoints
  • 📡 Server-Sent Events - Lightweight, one-way communication (server → client)
  • 🔄 Redis Pub/Sub - Horizontal scaling across multiple API instances
  • 🔐 Authorization Hooks - Secure topic subscriptions with custom auth logic
  • 💓 Heartbeat Support - Automatic connection keepalive (configurable interval)
  • 🎯 Topic-based Routing - Fine-grained subscription control per resource
  • 🔧 Type Safe - Full type hints and mypy compliance
  • ✅ Well Tested - Comprehensive test suite with 80%+ coverage

Why SSE?

Problem: Traditional REST APIs require manual refresh to see updates from other users. This creates poor collaboration experiences and inefficient workflows.

Why not WebSockets? WebSockets are powerful but complex:

  • Require separate infrastructure
  • More difficult to debug
  • Overkill for one-way notifications
  • Don't work well with existing REST patterns

SSE Solution:

  • Built on HTTP (works with existing infrastructure)
  • Native browser support (EventSource)
  • Automatic reconnection
  • Perfect for "notify then fetch" pattern
  • Keeps REST as source of truth

Installation

pip install fastapi-sse-events

Requirements:

  • Python 3.10+
  • FastAPI
  • Redis server

Quick Start

1. Install and Start Redis

# Using Docker
docker run -d -p 6379:6379 redis:alpine

# Or use local Redis
redis-server

2. Add to Your FastAPI App

from fastapi import Request
from fastapi_sse_events import SSEApp, publish_event, subscribe_to_events

app = SSEApp(
  title="My API",
  redis_url="redis://localhost:6379/0",
)

@app.get("/events")
@subscribe_to_events()
async def events(request: Request):
  pass

3. Publish Events from Your Endpoints

@app.post("/comments")
@publish_event(topic="comments", event="comment_created")
async def create_comment(request: Request, comment: Comment):
  # Save comment to database
  saved_comment = await db.save(comment)
  return saved_comment  # Auto-published to SSE clients

4. Subscribe from Client

// Connect to SSE stream
const eventSource = new EventSource('http://localhost:8000/events?topic=comments');

// Handle events
eventSource.addEventListener('comment_created', (e) => {
  const data = JSON.parse(e.data);
  console.log('Comment event:', data);

  // Notify-then-fetch pattern (recommended)
  fetch(`http://localhost:8000/comments/${data.id}`)
    .then(r => r.json())
    .then(renderComment);
});

That's it! Your app now has real-time updates.

Core Concepts

Architecture

┌─────────────┐         ┌─────────────┐         ┌─────────────┐
│  Client 1   │         │  Client 2   │         │  Client N   │
│ EventSource │         │ EventSource │         │ EventSource │
└──────┬──────┘         └──────┬──────┘         └──────┬──────┘
       │                       │                       │
       │        SSE Stream (/events?topic=...)        │
       └───────────────────────┼───────────────────────┘
                               │
                    ┌──────────▼──────────┐
                    │  FastAPI Instance   │
                    │                     │
                    │  REST API + SSE     │
                    └──────────┬──────────┘
                               │
                               │ Pub/Sub
                    ┌──────────▼──────────┐
                    │   Redis Server      │
                    │                     │
                    │  Topic Channels     │
                    └─────────────────────┘
                               ▲
                               │ Pub/Sub
                    ┌──────────┴──────────┐
                    │  FastAPI Instance   │
                    │                     │
                    │  REST API + SSE     │
                    └─────────────────────┘

How It Works

  1. Client opens SSE connection: GET /events?topic=comment_thread:123
  2. Server subscribes to Redis channel for that topic
  3. Another client posts data via REST API
  4. Server saves data and publishes event to Redis
  5. Redis broadcasts to all subscribed FastAPI instances
  6. All clients receive lightweight notification
  7. Clients fetch fresh data via existing REST endpoints

Event Flow

User Action → REST Endpoint → Save to DB → Publish Event → Redis Pub/Sub
                                                                ↓
Client ← SSE Stream ← FastAPI ← Redis Subscribe ← Redis Pub/Sub
  ↓
Fetch Updated Data ← REST Endpoint

Usage Guide

Basic Integration (Recommended)

from fastapi import Request
from fastapi_sse_events import SSEApp, publish_event, subscribe_to_events

app = SSEApp(redis_url="redis://localhost:6379/0")

@app.post("/tickets/{ticket_id}/status")
@publish_event(topic="tickets", event="ticket_status_changed")
async def update_ticket_status(request: Request, ticket_id: int, status: str):
  await db.update_ticket(ticket_id, status=status)
  return {"id": ticket_id, "status": status}

@app.get("/events")
@subscribe_to_events()  # topics via ?topic=tickets
async def events(request: Request):
  pass

Use SSEApp + decorators for minimal boilerplate. Keep mount_sse() for advanced/manual integration.

Publishing Events

There are several approaches to publishing events, ranging from the simple (recommended) to the advanced (manual control).

Method 1: Decorator-Based Publish (Recommended)

The simplest and most Pythonic approach. Just add @publish_event() to any endpoint, and the decorator automatically publishes the response:

from fastapi import Request
from fastapi_sse_events import publish_event

@app.post("/tickets/{ticket_id}/status")
@publish_event(topic="tickets", event="ticket_status_changed")
async def update_ticket_status(request: Request, ticket_id: int, status: str):
    await db.update_ticket(ticket_id, status=status)
    return {"id": ticket_id, "status": status}  # Auto-published to all subscribers

Features:

  • ✅ No boilerplate - just decorate your endpoint
  • ✅ Response automatically published as event data
  • ✅ Automatic HTTP and SSE notification
  • ✅ Works with any endpoint (GET, POST, PUT, DELETE)

Method 2: Direct Publish (Advanced)

For fine-grained control over event data, publish manually using app.state.event_broker:

from fastapi_sse_events import publish_event

@app.post("/tickets/{ticket_id}/status")
async def update_ticket_status(request: Request, ticket_id: int, status: str):
    # Update database
    await db.update_ticket(ticket_id, status=status)
    
    # Manual publish with custom data
    await app.state.event_broker.publish(
        topic=f"ticket:{ticket_id}",
        event="ticket_status_changed",
        data={
            "ticket_id": ticket_id,
            "status": status,
            "changed_at": datetime.utcnow().isoformat(),
            "changed_by": request.user.id
        }
    )
    
    return {"id": ticket_id, "status": status}

Use this when you need:

  • Custom/computed event data not in the HTTP response
  • Different data for SSE vs HTTP clients
  • Multiple events from a single endpoint

Method 3: Service Layer Pattern (Advanced)

Encapsulate publishing logic in service classes for clean separation of concerns:

from fastapi_sse_events import publish_event

class TaskService:
    def __init__(self, broker):
        self.broker = broker
    
    async def create_task(self, data: dict, user_id: int):
        # Business logic
        task = await db.save_task(data, user_id)
        
        # Publish to project subscribers
        await self.broker.publish(
            topic=f"project:{data['project_id']}",
            event="task:created",
            data={"id": task.id, "title": task.title, "assignee": user_id}
        )
        
        return task

# In endpoint
@app.post("/tasks")
async def create_task_endpoint(request: Request, data: TaskCreate):
    service = TaskService(app.state.event_broker)
    return await service.create_task(data.dict(), request.user.id)

Authorization

Protect SSE subscriptions with custom authorization callbacks. Control which users can subscribe to which topics:

from fastapi import Request, HTTPException
from fastapi_sse_events import SSEApp

async def authorize_subscription(request: Request, topic: str) -> bool:
    """
    Authorization callback - return True to allow subscription, False to deny.
    Called for each topic subscription attempt.
    
    Args:
        request: FastAPI request (contains user info via session, JWT, etc.)
        topic: Topic being subscribed to
    
    Returns:
        True if authorized, False otherwise
    """
    # Get current user (implement based on your auth method)
    user = await get_current_user(request)
    if not user:
        return False
    
    # Allow user to subscribe to their own topic
    if topic.startswith("user:"):
        user_id = topic.split(":")[1]
        return str(user.id) == user_id
    
    # Check workspace membership
    if topic.startswith("workspace:"):
        workspace_id = topic.split(":")[1]
        return await user_has_workspace_access(user.id, workspace_id)
    
    # Check organization access
    if topic.startswith("org:"):
        org_id = topic.split(":")[1]
        return await user_in_organization(user.id, org_id)
    
    # Broadcast topics - anyone can subscribe
    if topic in ["announcements", "system"]:
        return True
    
    # Deny by default
    return False

# Create app with authorization
app = SSEApp(
    redis_url="redis://localhost:6379/0",
    authorize=authorize_subscription  # Pass authorization callback
)

JWT-Based Authorization Example:

from fastapi.security import HTTPBearer, HTTPAuthCredentialDetails
from jose import jwt, JWTError

security = HTTPBearer()

async def authorize_subscription(request: Request, topic: str) -> bool:
    # Get token from header
    credentials = await security(request)
    if not credentials:
        return False
    
    try:
        # Decode JWT token
        payload = jwt.decode(
            credentials.credentials,
            settings.SECRET_KEY,
            algorithms=["HS256"]
        )
        user_id = payload.get("sub")
        user_roles = payload.get("roles", [])
    except JWTError:
        return False
    
    # Check topic permissions based on roles
    if topic.startswith("admin:"):
        return "admin" in user_roles
    
    if topic.startswith("user:"):
        return user_id == topic.split(":")[1]
    
    return True

Security Best Practices:

✅ Always:

  • Validate user authentication (JWT, sessions, etc.)
  • Check permissions for resource-specific topics
  • Log authorization failures for security monitoring
  • Use granular topic patterns with IDs (e.g., workspace:123 not just workspace)

❌ Never:

  • Allow wildcard subscriptions without explicit auth
  • Trust client-supplied topic names
  • Skip authorization for "internal" topics
  • Cache authorization results without invalidation

Topic Patterns

Topics are simple strings that route events to subscribers. Choose a consistent naming convention for your application:

Simple Topic Strings:

# Just use strings directly!
await app.state.event_broker.publish(
    topic="comments",
    event="comment:created",
    data={"id": 1, "text": "Hello"}
)

# Or with resource IDs for more granularity
await app.state.event_broker.publish(
    topic="comment:123",
    event="updated",
    data={"text": "Edited"}
)

Recommended Topic Conventions:

Pattern Use Case Example
resource Broadcast updates to all comments, tasks, notifications
resource:id Updates to specific resource comment:123, task:456, invite:789
resource:id:action Specific action on resource comment:123:deleted, task:456:assigned
workspace:id All updates in a workspace workspace:acme-corp
user:id User-specific notifications user:john_doe
broadcast or global System-wide announcements broadcast

Using TopicBuilder (Optional):

For consistency, you can use the optional TopicBuilder helper:

from fastapi_sse_events import TopicBuilder

# Built-in helper methods
TopicBuilder.comment("c123")           # → "comment:c123"
TopicBuilder.task("t456")              # → "task:t456"
TopicBuilder.workspace("ws789")        # → "workspace:ws789"
TopicBuilder.user("john")              # → "user:john"

# Or custom topics
TopicBuilder.custom("invoice", "inv123")  # → "invoice:inv123"

However, TopicBuilder is optional - plain strings work equally well and are often simpler.

Best Practices:

  • ✅ Use colons for namespacing: resource:id
  • ✅ Keep topic names lowercase and short
  • ✅ Include IDs for resource-specific topics (enables authorization)
  • ✅ Use consistent separators (colon recommended)
  • ✅ Document topic names your app uses
  • ❌ Avoid dynamic topic names without validation
  • ❌ Avoid spaces or special characters
  • ❌ Don't use topics to store data (topics are routing keys only)

Configuration

SSEApp Configuration

The SSEApp class provides simple one-line configuration:

from fastapi_sse_events import SSEApp
import os

# Minimal setup
app = SSEApp(redis_url="redis://localhost:6379/0")

# Full configuration
app = SSEApp(
    title="My API",
    redis_url=os.getenv("REDIS_URL", "redis://localhost:6379/0"),
    heartbeat_seconds=30,              # SSE keepalive interval (5-60s)
    authorize=authorize_subscription,   # Authorization callback (optional)
)

Configuration Parameters:

Parameter Type Default Description
title str "FastAPI" App title (optional)
redis_url str ⚠️ Required Redis connection URL
heartbeat_seconds int 30 SSE keepalive interval (5-60)
authorize callable None Authorization callback function

Environment Variables

Load configuration from environment for easier deployment:

import os
from fastapi_sse_events import SSEApp

app = SSEApp(
    redis_url=os.getenv("REDIS_URL", "redis://localhost:6379/0"),
    heartbeat_seconds=int(os.getenv("SSE_HEARTBEAT_SECONDS", "30")),
)

Standard Environment Variables:

# Development
export REDIS_URL="redis://localhost:6379/0"
export SSE_HEARTBEAT_SECONDS=30

# Production (with authentication)
export REDIS_URL="redis://:password@redis-prod.example.com:6379/0"
export SSE_HEARTBEAT_SECONDS=60

Load from .env file using python-dotenv:

from dotenv import load_dotenv
import os
from fastapi_sse_events import SSEApp

load_dotenv()

app = SSEApp(
    redis_url=os.getenv("REDIS_URL"),
    heartbeat_seconds=int(os.getenv("SSE_HEARTBEAT_SECONDS", "30")),
)

Advanced: Manual Broker Setup

For advanced use cases, you can configure the broker manually (not recommended for most applications):

from fastapi import FastAPI
from fastapi_sse_events import EventBroker, RealtimeConfig, mount_sse

app = FastAPI()

config = RealtimeConfig(
    redis_url="redis://localhost:6379/0",
    heartbeat_seconds=30,
)

broker = mount_sse(app, config, authorize=authorize_subscription)

Prefer SSEApp for new projects - it handles all the setup automatically.

Client Integration

JavaScript / Browser

// Connect to SSE endpoint
const eventSource = new EventSource('/events?topic=comments');

// Handle connection opened
eventSource.addEventListener('open', () => {
  console.log('✅ Connected to SSE');
  document.getElementById('status').textContent = 'Connected';
});

// Handle connection errors
eventSource.addEventListener('error', (error) => {
  console.error('❌ SSE Error:', error);
  document.getElementById('status').textContent = 'Disconnected';
  // EventSource automatically reconnects
});

// Listen to specific events
eventSource.addEventListener('comment:created', (event) => {
  const data = JSON.parse(event.data);
  console.log('New comment:', data);
  
  // Fetch fresh data from API
  fetch(`/comments/${data.id}`)
    .then(r => r.json())
    .then(comment => renderComment(comment));
});

eventSource.addEventListener('comment:updated', (event) => {
  const data = JSON.parse(event.data);
  console.log('Updated comment:', data);
  
  // Refresh specific comment
  fetch(`/comments/${data.id}`)
    .then(r => r.json())
    .then(comment => updateComment(comment));
});

// Listen to heartbeat (optional)
eventSource.addEventListener('heartbeat', (event) => {
  console.log('💓 Heartbeat');
});

// Subscribe to multiple topics
const multiTopics = ['comments', 'tasks', 'notifications'];
const eventSource2 = new EventSource(
  `/events?topic=${multiTopics.join(',')}`
);

// Cleanup
function closeSSE() {
  eventSource.close();
}
window.addEventListener('beforeunload', closeSSE);

React Hook Example

import { useEffect, useState, useCallback } from 'react';

// Custom hook for SSE
function useSSE(topic, onEvent) {
  const [connected, setConnected] = useState(false);

  useEffect(() => {
    const eventSource = new EventSource(`/events?topic=${topic}`);

    eventSource.addEventListener('open', () => {
      setConnected(true);
    });

    eventSource.addEventListener('error', () => {
      setConnected(false);
    });

    // Forward all events to callback
    eventSource.onmessage = (event) => {
      const data = JSON.parse(event.data);
      onEvent?.(event.type, data);
    };

    return () => eventSource.close();
  }, [topic, onEvent]);

  return connected;
}

// Component using the hook
function CommentList({ threadId }) {
  const [comments, setComments] = useState([]);

  // Handle SSE events
  const handleSSEEvent = useCallback((eventType, data) => {
    if (eventType === 'comment:created' || eventType === 'comment:updated') {
      // Refresh data from API
      fetchComments();
    }
  }, []);

  const connected = useSSE(`comment_thread:${threadId}`, handleSSEEvent);

  useEffect(() => {
    // Initial load
    fetchComments();
  }, [threadId]);

  const fetchComments = async () => {
    const response = await fetch(`/comments?thread_id=${threadId}`);
    const data = await response.json();
    setComments(data);
  };

  return (
    <div>
      <div style={{ color: connected ? 'green' : 'red' }}>
        {connected ? '🟢 Live' : '🔴 Offline'}
      </div>
      {comments.map(comment => (
        <div key={comment.id} className="comment">
          {comment.content}
        </div>
      ))}
    </div>
  );
}

Python Client

import httpx
import json

async def listen_to_events():
    """Listen to SSE events from Python."""
    async with httpx.AsyncClient() as client:
        async with client.stream(
            "GET",
            "http://localhost:8000/events?topic=comments",
            headers={"Accept": "text/event-stream"}
        ) as response:
            async for line in response.aiter_lines():
                if line.startswith("event:"):
                    event_type = line[7:].strip()
                elif line.startswith("data:"):
                    data_str = line[5:].strip()
                    try:
                        data = json.loads(data_str)
                        print(f"Event: {event_type}, Data: {data}")
                    except json.JSONDecodeError:
                        pass

# Using sseclient-py library (simpler)
from sseclient import SSEClient

def listen_simple():
    """Simple SSE listening with sseclient-py."""
    url = "http://localhost:8000/events?topic=comments"
    
    for event in SSEClient(url):
        if event.event != "heartbeat":
            print(f"Event: {event.event}")
            print(f"Data: {event.data}")

Deployment

Nginx Configuration

Disable buffering for SSE endpoints:

server {
    listen 80;
    server_name api.example.com;

    location /events {
        proxy_pass http://fastapi_backend;
        proxy_http_version 1.1;
        proxy_set_header Connection '';
        
        # Critical: Disable buffering for SSE
        proxy_buffering off;
        proxy_cache off;
        
        # Keep connection alive
        proxy_read_timeout 86400s;
        
        # Forward headers
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
    }

    location / {
        proxy_pass http://fastapi_backend;
        # Normal proxy settings
    }
}

Traefik Configuration

http:
  routers:
    sse-router:
      rule: "PathPrefix(`/events`)"
      service: sse-service
      
  services:
    sse-service:
      loadBalancer:
        servers:
          - url: "http://fastapi:8000"
        # Disable buffering
        responseForwarding:
          flushInterval: "1ms"

Docker Compose

version: '3.8'

services:
  redis:
    image: redis:alpine
    ports:
      - "6379:6379"
    
  api:
    build: .
    ports:
      - "8000:8000"
    environment:
      - SSE_REDIS_URL=redis://redis:6379/0
    depends_on:
      - redis
    # Scale horizontally
    deploy:
      replicas: 3

Horizontal Scaling

SSE Events works seamlessly across multiple instances:

  1. All instances connect to same Redis
  2. Events published by any instance reach all SSE clients
  3. No sticky sessions required
  4. Load balance normally
# Start multiple instances
uvicorn app:app --port 8000 &
uvicorn app:app --port 8001 &
uvicorn app:app --port 8002 &

# All instances share events via Redis

Production Deployment

FastAPI SSE Events is designed to scale from prototype to 100,000+ concurrent connections.

Architecture for 100K Users

                    ┌─────────────────┐
                    │  Nginx/ALB      │
                    │  Load Balancer  │
                    └────────┬────────┘
                             │
         ┌──────────────────┼──────────────────┐
         │                  │                  │
    ┌────▼───┐         ┌────▼───┐       ┌────▼────┐
    │FastAPI │         │FastAPI │  ...  │ FastAPI │
    │  #1    │         │  #2    │       │  #10    │
    │10K conn│         │10K conn│       │ 10K conn│
    └────┬───┘         └────┬───┘       └────┬────┘
         │                  │                  │
         └──────────────────┼──────────────────┘
                            │
                    ┌───────▼────────┐
                    │ Redis Cluster  │
                    │  (3-5 nodes)   │
                    └────────────────┘

Key Metrics

  • 10,000 connections per FastAPI instance
  • ~100KB memory per connection
  • < 10ms message latency
  • 100K messages/sec throughput
  • 99.9%+ uptime

Quick Start

cd examples/production_scale
./start.sh

This starts:

  • 10 FastAPI instances (100K capacity)
  • Redis Cluster (3 nodes)
  • Nginx load balancer
  • Prometheus + Grafana monitoring

Configuration for Scale

# .env.100k_users
SSE_REDIS_URL=redis://redis-1:7001,redis-2:7002,redis-3:7003/0
SSE_MAX_CONNECTIONS=10000      # Per instance
SSE_MAX_QUEUE_SIZE=50          # Memory optimization
SSE_MAX_MESSAGE_SIZE=32768     # 32KB limit
SSE_HEARTBEAT_SECONDS=30       # Efficient keepalive

Monitoring & Health Checks

Built-in endpoints for production monitoring:

GET /health          # Basic health check
GET /health/ready    # Readiness probe (load balancers)
GET /health/live     # Liveness probe (Kubernetes)
GET /metrics         # Detailed metrics (JSON)
GET /metrics/prometheus  # Prometheus format

Key Metrics to Monitor:

# Concurrent connections
sum(sse_connections_current)

# Connection rejection rate (scale up if > 0)
rate(sse_connections_rejected[5m])

# Message drop rate (should be < 0.1%)
rate(sse_messages_dropped[5m]) / rate(sse_messages_delivered[5m])

# Publish latency
avg(sse_publish_latency_ms)

Production Checklist

Before deploying to production:

  • Load Testing - Test with expected peak load × 2
  • Monitoring - Set up Prometheus + Grafana dashboards
  • Alerting - Configure alerts for connection rejections, high latency
  • Auto-scaling - Configure triggers (CPU > 70%, connections > 8K)
  • Security - Enable Redis auth, use SSL, restrict CORS
  • Persistence - Configure Redis persistence (AOF + RDB)
  • Failover - Test Redis cluster failover scenarios
  • Backpressure - Verify slow clients don't crash servers
  • Rate Limiting - Implement API rate limits
  • Logging - Set up centralized logging (ELK/Loki)

Performance Tuning

Redis Optimization:

# Increase max clients
redis-cli CONFIG SET maxclients 20000

# Optimize memory
redis-cli CONFIG SET maxmemory-policy allkeys-lru
redis-cli CONFIG SET maxmemory 4gb

# Disable persistence for pure cache (optional)
redis-cli CONFIG SET save ""

Linux Kernel Tuning:

# Increase file descriptors
ulimit -n 100000

# Increase network buffers
sysctl -w net.core.somaxconn=4096
sysctl -w net.ipv4.tcp_max_syn_backlog=8096

FastAPI Instance Sizing:

  • CPU: 1-2 cores per instance
  • Memory: 2GB base + 1GB per 10K connections
  • Network: 100Mbps per 10K connections

Complete Guide

For detailed deployment instructions, architecture decisions, and troubleshooting:

📚 Complete 100K User Deployment Guide

Topics covered:

  • Multi-region deployment
  • Kubernetes manifests
  • Cost optimization strategies
  • Disaster recovery
  • Performance benchmarking
  • Security hardening

Examples

See the examples/ directory for complete working examples:

CRM Comment System

Full-featured collaborative comment system with:

  • REST API for CRUD operations
  • Real-time SSE updates
  • Beautiful web interface
  • Multi-client synchronization
cd examples/crm_comments
uvicorn app:app --reload
# Open client.html in multiple browsers

More examples coming soon:

  • Ticket tracking system
  • Live dashboard with metrics
  • Collaborative document editing
  • Multi-tenant workspace notifications

API Reference

SSEApp (Recommended)

Simplified FastAPI subclass with built-in SSE support.

from fastapi_sse_events import SSEApp

class SSEApp(FastAPI):
    """FastAPI app with SSE support pre-configured."""
    
    def __init__(
        self,
        title: str = "API",
        redis_url: str = "redis://localhost:6379/0",
        heartbeat_seconds: int = 30,
        authorize: Optional[AuthorizeFn] = None,
        **kwargs
    ):
        pass

Usage:

app = SSEApp(
    title="My API",
    redis_url="redis://localhost:6379/0",
    heartbeat_seconds=30,
    authorize=authorize_callback  # Optional
)

Automatically Provides:

  • ✅ EventBroker at app.state.event_broker
  • ✅ SSE endpoint at /events (with ?topic= query param)
  • ✅ Health checks at /health*
  • ✅ Metrics at /metrics

Decorators

@publish_event()

Automatically publishes endpoint response to subscribers.

from fastapi_sse_events import publish_event

@app.post("/comments")
@publish_event(topic="comments", event="comment:created")
async def create_comment(request: Request, text: str):
    # Save to database
    comment = await db.save_comment(text)
    # Response automatically published to all subscribers!
    return comment

Parameters:

  • topic (str): Topic to publish to
  • event (str): Event type identifier

@subscribe_to_events()

Converts endpoint to SSE stream. Query param: ?topic=...

from fastapi_sse_events import subscribe_to_events

@app.get("/events")
@subscribe_to_events()
async def events(request: Request):
    # Endpoint body can be empty
    pass

Query Parameters:

  • topic (str, required): Topic to subscribe to
  • topic=topic1,topic2: Multiple topics (comma-separated)

EventBroker

Manual event publishing (use decorators when possible).

class EventBroker:
    async def publish(
        self,
        topic: str,
        event: str,
        data: dict[str, Any] | EventData
    ) -> None:
        """Publish event to all subscribers on topic."""

Example:

await app.state.event_broker.publish(
    topic="comments",
    event="comment:created",
    data={"id": 1, "text": "Hello"}
)

Configuration (SSEApp Parameters)

Parameter Type Default Description
title str "API" Application title
redis_url str "redis://localhost:6379/0" Redis connection URL
heartbeat_seconds int 30 SSE keepalive interval (5-60)
authorize AuthorizeFn None Authorization callback (optional)

Advanced: Manual Setup

For complex use cases, manually configure the broker:

from fastapi import FastAPI
from fastapi_sse_events import mount_sse, RealtimeConfig

app = FastAPI()

config = RealtimeConfig(
    redis_url="redis://localhost:6379/0",
    heartbeat_seconds=30,
)

broker = mount_sse(app, config, authorize=authorize_fn)

TopicBuilder (Optional)

Helper for consistent topic naming (optional).

from fastapi_sse_events import TopicBuilder

# All optional - plain strings work too
TopicBuilder.comment("c123")        # → "comment:c123"
TopicBuilder.task("t456")           # → "task:t456"
TopicBuilder.workspace("ws789")     # → "workspace:ws789"
TopicBuilder.user("john")           # → "user:john"
TopicBuilder.custom("invoice", "i001")  # → "invoice:i001"

Type Definitions

# Authorization callback type
AuthorizeFn = Callable[[Request, str], Awaitable[bool]]

# Event data
EventData = dict[str, Any]  # Any JSON-serializable dict

# Example authorization
async def authorize(request: Request, topic: str) -> bool:
    """Return True to allow subscription, False to deny."""
    user = await get_current_user(request)
    return user is not None and topic.startswith(f"user:{user.id}")

Troubleshooting

SSE Connection Fails

Symptom: Browser can't connect to /events

Solutions:

  1. Check Redis is running: redis-cli ping
  2. Verify endpoint is registered: check FastAPI docs at /docs
  3. Check CORS settings if cross-origin
  4. Look for errors in server logs

Updates Not Appearing

Symptom: Events published but clients don't receive them

Solutions:

  1. Verify topic names match exactly
  2. Check Redis pub/sub: redis-cli SUBSCRIBE topic_name
  3. Ensure all instances connect to same Redis
  4. Check authorization function isn't denying access
  5. Look for JavaScript errors in browser console

High Memory Usage

Symptom: Redis or app memory grows over time

Solutions:

  1. Monitor Redis: redis-cli INFO memory
  2. Implement idle connection timeouts
  3. Reduce heartbeat frequency if many connections
  4. Set connection limits in load balancer
  5. Monitor disconnected clients

Proxy Buffering Issues

Symptom: Events delayed or not arriving

Solution: Disable buffering in proxy:

# Nginx
proxy_buffering off;
proxy_cache off;

# Apache
SetEnv proxy-nokeepalive 1

"Redis not connected" Error

Symptom: RuntimeError: Redis client not connected

Solutions:

  1. Check Redis is running and accessible
  2. Verify redis_url in config
  3. Check network/firewall rules
  4. Ensure mount_sse() called before app starts

Performance Tips

  1. Use topic scoping - Narrow topics reduce fan-out
  2. Keep payloads small - Send IDs, not full objects
  3. Implement idle timeouts - Clean up dead connections
  4. Monitor Redis - Watch memory and connection count
  5. Connection pooling - Reuse Redis connections
  6. Event batching - For high-frequency updates, debounce on client

Comparison

Feature SSE (this library) WebSockets Polling
Real-time ✅ Near real-time ✅ Real-time ❌ Delayed
Infrastructure ✅ HTTP (simple) ❌ Complex ✅ HTTP
Browser support ✅ Native ✅ Native ✅ Native
Bidirectional ❌ Server→Client ✅ Both ❌ Client→Server
Auto-reconnect ✅ Built-in ❌ Manual N/A
Horizontal scaling ✅ Via Redis ❌ Sticky sessions ✅ Stateless
Debugging ✅ Easy ❌ Harder ✅ Easy
Use case Notifications Chat, games Legacy

🤝 Contributing

Contributions are welcome! Here's how you can help:

Development Setup

# Clone repository
git clone https://github.com/bhadri01/fastapi_sse_events.git
cd fastapi_sse_events

# Install with dev dependencies
pip install -e ".[dev]"

# Run tests
pytest tests/ -v

# Run linting
ruff check fastapi_sse_events/ tests/ examples/
ruff format fastapi_sse_events/ tests/ examples/

# Run type checking
mypy fastapi_sse_events/

Contribution Guidelines

  1. Fork the repository
  2. Create a feature branch (git checkout -b feature/amazing-feature)
  3. Add tests for new features
  4. Ensure all tests pass (pytest)
  5. Run linting and formatting (ruff)
  6. Commit your changes (git commit -m 'Add amazing feature')
  7. Push to the branch (git push origin feature/amazing-feature)
  8. Open a Pull Request

Reporting Issues

  • Use GitHub Issues
  • Include Python version, FastAPI version, and Redis version
  • Provide minimal reproducible example
  • Check existing issues first

📄 License

This project is licensed under the MIT License - see the LICENSE file for details.

Copyright © 2025 bhadri01

🙏 Acknowledgments

🔗 Related Projects

⭐ Star History

If you find this project helpful, please consider giving it a star on GitHub!

Star History Chart

Made with ❤️ by bhadri01

Support

Built with ❤️ for the FastAPI community

Project links

Key dates

PyPI data
Data sourced directly from PyPI's database.
  • Released:
Latest release

1 maintainer

PyPI data
Data sourced directly from PyPI's database.
Avatar for bhadri01 from gravatar.com bhadri01

Credits

Author/Maintainer: bhadri01

License

MIT License (MIT)

Requires

Python >=3.10

Provides Extra

server dev all

Tags

fastapi sse server-sent-events redis pubsub realtime notifications websocket-alternative push-notification event-streaming

Classifiers

Report project as malware

Metadata

Project links

Key dates

PyPI data
Data sourced directly from PyPI's database.
  • Released:
Latest release

1 maintainer

PyPI data
Data sourced directly from PyPI's database.
Avatar for bhadri01 from gravatar.com bhadri01

Credits

Author/Maintainer: bhadri01

License

MIT License (MIT)

Requires

Python >=3.10

Provides Extra

server dev all

Tags

fastapi sse server-sent-events redis pubsub realtime notifications websocket-alternative push-notification event-streaming

Classifiers

Report project as malware

Release files for fastapi-sse-events 0.2.0

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for fastapi-sse-events 0.2.0
File Size Uploaded
fastapi_sse_events-0.2.0.tar.gz 59.8 kB Details

Built distribution (wheel)

Table of built distributions (wheels) for fastapi-sse-events 0.2.0
File Interpreter ABI Platform
fastapi_sse_events-0.2.0-py3-none-any.whl 37.3 kB Python 3 none any Details

Total release size: 97.1 kB

Release files / fastapi_sse_events-0.2.0.tar.gz

Download URL fastapi_sse_events-0.2.0.tar.gz
Size 59.8 kB
Tags Source
SHA-256 checksum
How to use checksums 		e1aff539ce1567da5dda7600c3ff353e08da19efd62e836d379e0ce6b70ccc30
BLAKE2b-256 checksum
How to use checksums 		1ca0e44abbdfc1f8725e31f400a99fec8fce73e0865cc955fa798626d7e74c5d
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing? 		No
Uploaded via poetry/2.3.2 CPython/3.10.19 Linux/6.14.0-1017-azure

Release files / fastapi_sse_events-0.2.0-py3-none-any.whl

Download URL fastapi_sse_events-0.2.0-py3-none-any.whl
Size 37.3 kB
Tags Python 3
SHA-256 checksum
How to use checksums 		8ac7418e55fd513456c0e1710dc785342924135d6b971828e860fc0f837d0db7
BLAKE2b-256 checksum
How to use checksums 		921cab4295ddceb684f6bf5a44b1c6e4d5ce96aeac987c7c89f62d820652080f
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing? 		No
Uploaded via poetry/2.3.2 CPython/3.10.19 Linux/6.14.0-1017-azure

Release history Release notifications | RSS feed

This release

0.2.0 This release

2 release files

Anthropic, PBC Visionary sponsor Bloomberg Visionary sponsor Hudson River Trading Visionary sponsor Meta Visionary sponsor NVIDIA Visionary sponsor Microsoft Sustainability sponsor Depot Continuous Integration AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page