FastAPI SSE Events

pip install fastapi-sse-events

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"
)

@app.post("/tasks")
@publish_event(topic="tasks", event="task:created")
async def create_task(request: Request, task: dict):
    # ... save task to database ...
    return {"id": 1, **task}  # auto-published to SSE clients

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

// Subscribe to topic from query param
const eventSource = new EventSource('/events?topic=tasks');

eventSource.addEventListener('task:created', async (event) => {
    const data = JSON.parse(event.data);
    console.log('Task event:', data);

    // Notify-then-fetch pattern (recommended)
    const response = await fetch(`/tasks/${data.id}`);
    const task = await response.json();
    renderTask(task);
});

eventSource.addEventListener('open', () => {
    console.log('✅ Connected to SSE');
});

from fastapi import Request
from fastapi_sse_events import SSEApp, sse_event, sse_endpoint

# One-line setup with auto-configuration!
app = SSEApp(
    title="My API",
    redis_url="redis://localhost:6379"  # or use REDIS_URL env var
)

# That's it! No manual broker setup, no lifecycle management needed!

@app.post("/comments")
@publish_event(topic="comments", event="comment_created")
async def create_comment(request: Request, comment: dict):
    # Your business logic here
    new_comment = {"id": 1, "content": comment["content"]}
    
    # Just return it - automatically published to SSE clients!
    return new_comment

@app.get("/events")
@subscribe_to_events()
async def events(request: Request):
    pass  # Decorator handles all streaming logic!

from fastapi_sse_events import SSEApp

app = SSEApp(
    title="My API",
    version="1.0.0",
    
    # Redis connection (choose one)
    redis_url="redis://localhost:6379",       # Option 1: URL
    # redis_host="localhost",                  # Option 2: Host/Port
    # redis_port=6379,
    # redis_db=0,
    
    # Optional settings
    topic_prefix="myapp:",                    # Prefix all topics
    enable_cors=True,                         # Auto-enable CORS
    cors_origins=["*"],                       # Allowed origins
)

from fastapi_sse_events import publish_event

# Explicit topic and event name
@app.post("/comments")
@publish_event(topic="comments", event="comment_created")
async def create_comment(request: Request, comment: dict):
    return {"id": 1, "content": comment["content"]}

# Auto-infer topic from route path
@app.post("/threads/{thread_id}/comments")
@publish_event()  # Auto-infers topic: "threads.comments"
async def add_comment(request: Request, thread_id: int, comment: dict):
    return {"id": 1, "thread_id": thread_id, **comment}

# Custom data extraction
@app.post("/users")
@publish_event(
    topic="users",
    extract_data=lambda resp: {"user_id": resp["id"]}  # Only publish ID
)
async def create_user(request: Request, user: dict):
    full_user = {"id": 1, "password": "secret", **user}
    return full_user  # Only {"user_id": 1} published to SSE

from fastapi_sse_events import sse_endpoint

# Simple SSE endpoint with static topics
@app.get("/events")
@sse_endpoint(topics=["comments", "users"])
async def events(request: Request):
    pass  # Decorator handles everything!

# Dynamic topics from query params
@app.get("/events/dynamic")
@sse_endpoint()  # Topics from ?topic=comments,users
async def events_dynamic(request: Request):
    pass

# With authorization
async def check_auth(request: Request, topic: str) -> bool:
    user = await get_current_user(request)
    return user.can_access(topic)

@app.get("/events/secure")
@sse_endpoint(topics=["comments"], authorize=check_auth)
async def events_secure(request: Request):
    pass

# Custom heartbeat
@app.get("/events/fast")
@sse_endpoint(topics=["live-data"], heartbeat=10)  # 10 second heartbeat
async def events_fast(request: Request):
    pass

# Before
app = FastAPI()
config = RealtimeConfig(redis_url="...")
# ... manual setup ...

# After
app = SSEApp(redis_url="...")

# Before
@app.post("/comments")
async def create_comment(comment: dict):
    new_comment = save(comment)
    await broker.publish("comments", "created", new_comment)
    return new_comment

# After
@app.post("/comments")
@sse_event(topic="comments", event="created")
async def create_comment(request: Request, comment: dict):
    new_comment = save(comment)
    return new_comment  # Auto-published!

# Before
mount_sse(app, broker, authorize_fn)
# Or manual endpoint creation...

# After
@app.get("/events")
@sse_endpoint(topics=["comments"], authorize=authorize_fn)
async def events(request: Request):
    pass

"""Simplified CRM Comments with SSE"""
from fastapi import Request, HTTPException
from pydantic import BaseModel
from fastapi_sse_events import SSEApp, sse_event, sse_endpoint

# One-line app setup
app = SSEApp(
    title="CRM Comments",
    redis_url="redis://localhost:6379",
    enable_cors=True
)

# In-memory storage
comments_db = {}
comment_id = 0

class CommentCreate(BaseModel):
    thread_id: int
    author: str
    content: str

# CRUD endpoints with auto-publishing
@app.post("/comments")
@sse_event(topic="comments", event="comment_created")
async def create_comment(request: Request, comment: CommentCreate):
    global comment_id
    comment_id += 1
    
    new_comment = {
        "id": comment_id,
        "thread_id": comment.thread_id,
        "author": comment.author,
        "content": comment.content
    }
    
    comments_db[comment_id] = new_comment
    return new_comment  # Auto-published to SSE!

@app.put("/comments/{comment_id}")
@sse_event(topic="comments", event="comment_updated")
async def update_comment(
    request: Request,
    comment_id: int,
    content: str
):
    if comment_id not in comments_db:
        raise HTTPException(404, "Not found")
    
    comments_db[comment_id]["content"] = content
    return comments_db[comment_id]  # Auto-published!

@app.delete("/comments/{comment_id}")
@sse_event(topic="comments", event="comment_deleted")
async def delete_comment(request: Request, comment_id: int):
    if comment_id not in comments_db:
        raise HTTPException(404, "Not found")
    
    deleted = comments_db.pop(comment_id)
    return {"id": comment_id}  # Auto-published!

@app.get("/comments")
async def list_comments():
    return list(comments_db.values())

# SSE endpoint with auto-streaming
@app.get("/events")
@sse_endpoint(topics=["comments"])
async def events(request: Request):
    pass  # All automatic!

# Run with: uvicorn app_simple:app --reload

┌─────────────┐         ┌─────────────┐         ┌─────────────┐
│   Browser   │  SSE    │  FastAPI    │  Pub    │    Redis    │
│   Client    │◄────────│   Instance  │◄────────│   Pub/Sub   │
└─────────────┘         └─────────────┘         └─────────────┘
      │                       │                        ▲
      │                       │                        │
      │  POST /api/task       │  Publish Event         │
      └──────────────────────►│───────────────────────►│
                              │                        │
                              │   Subscribe Topics     │
                              └───────────────────────►│

User A                  Server                  User B
  │                         │                      │
  │  1. POST /comments      │                      │
  │────────────────────────►│                      │
  │                         │                      │
  │                         │  2. SSE Event        │
  │                         │────────────────────►│
  │                         │   "comment:created"  │
  │                         │                      │
  │                         │  3. GET /comments    │
  │                         │◄────────────────────│
  │                         │                      │
  │                         │  4. Fresh Data       │
  │                         │────────────────────►│

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"
)

@app.post("/comments")
@publish_event(topic="comments", event="comment:created")
async def create_comment(request: Request, comment: dict):
    # Save to database
    saved_comment = {"id": 1, **comment}
    return saved_comment  # auto-published

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

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

// Listen for specific events
sse.addEventListener('comment:created', (event) => {
  const data = JSON.parse(event.data);
  console.log('New comment:', data);

  // Refresh data from REST API
    fetch(`/comments/${data.id}`).then(r => r.json()).then(renderComment);
});

// Connection status
sse.addEventListener('open', () => {
  console.log('Connected to SSE');
});

sse.addEventListener('error', (error) => {
  console.error('SSE error:', error);
  // EventSource auto-reconnects
});

from fastapi_sse_events import publish_event

@app.post("/tickets/{ticket_id}")
@publish_event(topic="tickets", event="ticket:updated")
async def update_ticket(request: Request, ticket_id: int, status: str):
    await db.update(ticket_id, status=status)
    return {"id": ticket_id, "status": status}  # Auto-published!

await app.state.event_broker.publish(
    topic="resource:123",      # Topic (routing key)
    event="resource:updated",  # Event type
    data={                     # Payload
        "id": "123",
        "changed_fields": ["status"]
    }
)

# Resource Created
await app.state.event_broker.publish(
    topic="tasks",
    event="task:created",
    data={"id": task_id, "title": "New Task"}
)

# Resource Updated
await app.state.event_broker.publish(
    topic=f"task:{task_id}",
    event="task:updated",
    data={"id": task_id, "status": "completed"}
)

# Resource Deleted
await app.state.event_broker.publish(
    topic=f"task:{task_id}",
    event="task:deleted",
    data={"id": task_id}
)

# Broadcast to All
await app.state.event_broker.publish(
    topic="global",
    event="system:maintenance",
    data={"message": "Scheduled maintenance in 5 minutes"}
)

from fastapi import Request
from fastapi_sse_events import SSEApp

async def authorize_subscription(request: Request, topic: str) -> bool:
    """
    Authorization callback - return True to allow, False to deny.
    Called for each topic subscription attempt.
    """
    # 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)
    
    # 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
)

# Simple topics
await app.state.event_broker.publish(
    topic="comments",  # Broadcast to all subscribers
    event="comment:created",
    data={"id": 1, "text": "Hello"}
)

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

# Workspace-specific
await app.state.event_broker.publish(
    topic="workspace:acme-corp",
    event="task:created",
    data={"id": 456, "title": "New Task"}
)

from fastapi_sse_events import TopicBuilder

# Built-in patterns (all optional)
TopicBuilder.comment("c123")        # → "comment:c123"
TopicBuilder.task("t456")           # → "task:t456"
TopicBuilder.workspace("ws789")     # → "workspace:ws789"
TopicBuilder.user("john")           # → "user:john"

# Custom topics
TopicBuilder.custom("invoice", "inv001")  # → "invoice:inv001"

// Subscribe to broadcast topic
new EventSource('/events?topic=comments')

// Subscribe to specific resource
new EventSource('/events?topic=comment:123')

// Subscribe to workspace
new EventSource('/events?topic=workspace:acme-corp')

// Subscribe to multiple topics (comma-separated)
new EventSource('/events?topic=comment:123,workspace:acme-corp')

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)
)

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

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

from dotenv import load_dotenv
import os
from fastapi_sse_events import SSEApp

load_dotenv()  # Load .env file

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

// Create SSE connection
const eventSource = new EventSource('/events?topic=comments');

// Listen for connection events
eventSource.addEventListener('open', () => {
  console.log('✅ Connected to SSE');
  updateStatus('Connected');
});

eventSource.addEventListener('error', (error) => {
  console.error('❌ SSE Error:', error);
  updateStatus('Disconnected');
  // EventSource automatically reconnects
});

// Listen for specific events
eventSource.addEventListener('comment:created', (event) => {
  const data = JSON.parse(event.data);
  console.log('New comment:', data);
  
  // Fetch and display the new comment
  fetch(`/comments/${data.id}`)
    .then(r => r.json())
    .then(comment => addCommentToDOM(comment));
});

eventSource.addEventListener('comment:updated', (event) => {
  const data = JSON.parse(event.data);
  updateCommentInDOM(data.id, data);
});

eventSource.addEventListener('heartbeat', () => {
  console.log('💓 Heartbeat');
});

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

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

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

interface UseSSEOptions {
  topic: string;
  onEvent?: (eventType: string, data: any) => void;
}

export function useSSE({ topic, onEvent }: UseSSEOptions) {
  const [connected, setConnected] = useState(false);
  const [error, setError] = useState(null);

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

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

    eventSource.addEventListener('error', () => {
      setConnected(false);
      setError(new Error('SSE connection error'));
    });

    // 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, error };
}

// Usage example
export function CommentThread({ threadId }: { threadId: number }) {
  const [comments, setComments] = useState([]);

  const handleEvent = useCallback((eventType: string, data: any) => {
    if (['comment:created', 'comment:updated'].includes(eventType)) {
      // Refresh comments from API
      fetchComments();
    }
  }, []);

  const { connected } = useSSE(
    { topic: `comment_thread:${threadId}`, onEvent: handleEvent }
  );

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

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

  return (
    

      
        {connected ? '🟢 Live' : '🔴 Offline'}
      
      {comments.map(c => (
        
{c.content}
      ))}
    
  );
}

import httpx
import json

async def listen_to_sse():
    """Listen to SSE events using httpx."""
    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}")
                    except json.JSONDecodeError:
                        pass

from sseclient import SSEClient
import json

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

# Run it
listen_to_events()

# Memory management
maxmemory 2gb
maxmemory-policy allkeys-lru

# Persistence (optional for pub/sub)
save ""
appendonly no

# Network
bind 0.0.0.0
protected-mode yes
requirepass your-secure-password

# Pub/Sub specific
timeout 0
tcp-keepalive 300

# Redis
SSE_REDIS_URL=redis://:password@redis-host:6379/0

# SSE Config
SSE_HEARTBEAT_SECONDS=30
SSE_PATH=/sse
SSE_TOPIC_PREFIX=production

# FastAPI
APP_ENV=production
LOG_LEVEL=info

uvicorn app:app \
  --host 0.0.0.0 \
  --port 8000 \
  --workers 4 \
  --loop uvloop \
  --log-level info \
  --proxy-headers \
  --forwarded-allow-ips='*'

upstream fastapi_backend {
    server 127.0.0.1:8000;
    server 127.0.0.1:8001;
    server 127.0.0.1:8002;
    keepalive 32;
}

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

    # Standard API endpoints
    location / {
        proxy_pass http://fastapi_backend;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    }

    # SSE endpoint - special configuration
    location /sse {
        proxy_pass http://fastapi_backend;
        proxy_http_version 1.1;
        
        # SSE specific headers
        proxy_set_header Connection '';
        proxy_set_header Cache-Control 'no-cache';
        
        # Disable buffering
        proxy_buffering off;
        proxy_cache off;
        
        # Long timeouts for SSE
        proxy_read_timeout 86400s;
        proxy_send_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;
        
        # Chunked transfer encoding
        chunked_transfer_encoding on;
    }
}

version: '3.8'

services:
  redis:
    image: redis:7-alpine
    command: redis-server --requirepass ${REDIS_PASSWORD}
    ports:
      - "6379:6379"
    volumes:
      - redis_data:/data
    restart: unless-stopped

  api:
    build: .
    command: uvicorn app:app --host 0.0.0.0 --port 8000
    environment:
      SSE_REDIS_URL: redis://:${REDIS_PASSWORD}@redis:6379/0
      SSE_HEARTBEAT_SECONDS: 30
    ports:
      - "8000:8000"
    depends_on:
      - redis
    restart: unless-stopped
    deploy:
      replicas: 3
      
volumes:
  redis_data:

FROM python:3.11-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Copy application
COPY . .

# Run
CMD ["uvicorn", "app:app", "--host", "0.0.0.0", "--port", "8000"]

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

@publish_event(
    topic: str,
    event: str
)

@subscribe_to_events()
async def events(request: Request):
    pass

async def publish(
    self,
    topic: str,
    event: str,
    data: dict | EventData
) -> None

class TopicBuilder:
    @staticmethod
    def comment(comment_id: str) -> str       # "comment:id"
    
    @staticmethod
    def task(task_id: str) -> str             # "task:id"
    
    @staticmethod
    def workspace(workspace_id: str) -> str   # "workspace:id"
    
    @staticmethod
    def user(user_id: str) -> str             # "user:id"
    
    @staticmethod
    def custom(resource: str, id: str) -> str # "resource:id"

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

# 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

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

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=auth_fn)

# Now use: app.state.event_broker

cd examples/quickstart
pip install -r requirements.txt  # Or: pip install fastapi uvicorn fastapi-sse-events redis
redis-server                      # Start Redis (or: docker run -d -p 6379:6379 redis:7-alpine)
uvicorn app:app --reload         # Start FastAPI

cd examples/crm_comments
./start.sh

# Or manually:
pip install -r requirements.txt
uvicorn app:app --reload

cd examples/production_scale
./start.sh          # Starts all services via docker-compose

# Run load tests
pip install -r load_test_requirements.txt
python load_test.py

// Enable verbose logging
const eventSource = new EventSource('/sse?topics=debug');

eventSource.onmessage = (event) => {
  console.log('Raw SSE:', event);
};

// Monitor all event types
['open', 'error', 'message'].forEach(type => {
  eventSource.addEventListener(type, (e) => {
    console.log(`[${type}]`, e);
  });
});