# Voicebox Async Generation with SSE Streaming: Implementation Guide

> Implement Voicebox async generation with SSE streaming. Learn how to stream real-time TTS progress from FastAPI to React for seamless audio playback. Get the guide now.

- Repository: [Jamie Pine/voicebox](https://github.com/jamiepine/voicebox)
- Tags: how-to-guide
- Published: 2026-04-14

---

**Voicebox implements asynchronous text-to-speech generation using Server-Sent Events (SSE) to stream real-time progress updates from the FastAPI backend to the React frontend, enabling fire-and-forget generation with automatic audio playback upon completion.**

Voicebox by jamiepine delivers a fully asynchronous text-to-speech pipeline that eliminates polling overhead through persistent SSE connections. When users initiate generation, the backend spawns background TTS workers while the frontend subscribes to event streams that report granular progress states including `loading_model`, `generating`, `completed`, `failed`, and `not_found`. This architecture enables real-time UI updates without page reloads and supports automatic audio playback the moment synthesis finishes.

## Architecture Overview

The async generation system spans both frontend and backend layers, with clean separation between connection management, business logic, and data persistence.

| Component | Responsibility | Primary Files |
|-----------|----------------|---------------|
| **React UI** | Initiates generation, manages pending IDs, consumes SSE streams, handles auto-playback logic | [`app/src/lib/hooks/useGenerationProgress.ts`](https://github.com/jamiepine/voicebox/blob/main/app/src/lib/hooks/useGenerationProgress.ts) |
| **API Client** | Constructs SSE endpoint URLs and audio download paths | [`app/src/lib/api/client.ts`](https://github.com/jamiepine/voicebox/blob/main/app/src/lib/api/client.ts) |
| **FastAPI Backend** | Accepts requests, schedules background jobs, exposes SSE status streams and direct audio streaming | [`backend/routes/generations.py`](https://github.com/jamiepine/voicebox/blob/main/backend/routes/generations.py) |
| **Task Workers** | Execute heavy-weight TTS computation via `generate_chunked` and update database records | [`backend/utils/chunked_tts.py`](https://github.com/jamiepine/voicebox/blob/main/backend/utils/chunked_tts.py) |
| **Database** | Persists `DBGeneration` rows with `status`, `duration`, and `error` fields | [`backend/database/models.py`](https://github.com/jamiepine/voicebox/blob/main/backend/database/models.py) |

The **SSE stream** maintains a lightweight, unidirectional HTTP connection that pushes JSON payloads whenever generation status changes. The UI tracks active connections using a `Map<string, EventSource>` keyed by generation ID, ensuring resources are released promptly when jobs complete or fail.

## SSE Data Flow

The generation lifecycle follows a strict event-driven protocol:

1. **Generation Request**: The UI POSTs a `GenerationRequest` to `/generate`. The server creates a `DBGeneration` row with status `loading_model` and returns the UUID.
2. **Pending Tracking**: The client adds the ID to a Zustand store (`useGenerationStore.pendingGenerationIds`).
3. **SSE Connection**: The `useGenerationProgress` hook opens an `EventSource` to `GET /generate/{id}/status`.
4. **Server Loop**: The `event_stream` coroutine in [`backend/routes/generations.py`](https://github.com/jamiepine/voicebox/blob/main/backend/routes/generations.py) queries the database every second, yields JSON payloads, and sleeps via `await asyncio.sleep(1)` until status reaches `completed` or `failed`.
5. **Completion Handling**: Upon receiving the final event, the hook closes the connection, invalidates React Query caches, optionally adds the audio to stories, and triggers auto-playback if enabled.

This **fire-and-forget** pattern ensures the UI never blocks waiting for generation while providing real-time feedback.

## Client-Side Implementation

### The useGenerationProgress Hook

The core of the frontend implementation resides in [`app/src/lib/hooks/useGenerationProgress.ts`](https://github.com/jamiepine/voicebox/blob/main/app/src/lib/hooks/useGenerationProgress.ts). This hook manages the lifecycle of SSE connections and coordinates state updates across the application.

```typescript
// app/src/lib/hooks/useGenerationProgress.ts
export function useGenerationProgress() {
  const queryClient = useQueryClient();
  const { toast } = useToast();
  
  const pendingIds = useGenerationStore(s => s.pendingGenerationIds);
  const removePendingGeneration = useGenerationStore(s => s.removePendingGeneration);
  const removePendingStoryAdd = useGenerationStore(s => s.removePendingStoryAdd);
  const isPlaying = usePlayerStore(s => s.isPlaying);
  const setAudioWithAutoPlay = usePlayerStore(s => s.setAudioWithAutoPlay);
  const autoplayOnGenerate = useServerStore(s => s.autoplayOnGenerate);

  const isPlayingRef = useRef(isPlaying);
  const autoplayRef = useRef(autoplayOnGenerate);
  isPlayingRef.current = isPlaying;
  autoplayRef.current = autoplayOnGenerate;

  const eventSourcesRef = useRef<Map<string, EventSource>>(new Map());

  // Cleanup all connections on unmount
  useEffect(() => {
    const sources = eventSourcesRef.current;
    return () => {
      for (const source of sources.values()) source.close();
      sources.clear();
    };
  }, []);

  // Manage SSE connections based on pending IDs
  useEffect(() => {
    const currentSources = eventSourcesRef.current;

    // Close connections for IDs no longer pending
    for (const [id, source] of currentSources.entries()) {
      if (!pendingIds.has(id)) {
        source.close();
        currentSources.delete(id);
      }
    }

    // Open connections for newly pending IDs
    for (const id of pendingIds) {
      if (currentSources.has(id)) continue;

      const url = apiClient.getGenerationStatusUrl(id);
      const source = new EventSource(url);

      source.onmessage = event => {
        try {
          const data: GenerationStatusEvent = JSON.parse(event.data);
          
          if (data.status === 'completed') {
            source.close();
            currentSources.delete(id);
            removePendingGeneration(id);
            queryClient.refetchQueries({ queryKey: ['history'] });

            const storyId = removePendingStoryAdd(id);
            if (storyId) {
              apiClient.addStoryItem(storyId, { generation_id: id })
                .then(() => {
                  queryClient.invalidateQueries({ queryKey: ['stories'] });
                  toast({ title: 'Added to story', description: `Audio generated (${data.duration?.toFixed(2)}s)` });
                });
            }

            if (autoplayRef.current && !isPlayingRef.current) {
              const genAudioUrl = apiClient.getAudioUrl(id);
              setAudioWithAutoPlay(genAudioUrl, id, '', '');
            }
          } else if (data.status === 'failed' || data.status === 'not_found') {
            source.close();
            currentSources.delete(id);
            removePendingGeneration(id);
            removePendingStoryAdd(id);
            queryClient.refetchQueries({ queryKey: ['history'] });
            toast({
              title: data.status === 'not_found' ? 'Generation not found' : 'Generation failed',
              description: data.error ?? 'An error occurred',
              variant: 'destructive',
            });
          }
        } catch {
          // Ignore malformed messages
        }
      };

      source.onerror = () => {
        source.close();
        currentSources.delete(id);
        removePendingGeneration(id);
        queryClient.refetchQueries({ queryKey: ['history'] });
      };

      currentSources.set(id, source);
    }
  }, [
    pendingIds,
    removePendingGeneration,
    removePendingStoryAdd,
    queryClient,
    toast,
    setAudioWithAutoPlay,
  ]);
}

```

The hook uses `eventSourcesRef` to prevent duplicate connections for the same generation ID. Connection cleanup occurs both when IDs leave the pending set and when the component unmounts, preventing memory leaks.

### API Client Configuration

The `ApiClient` class centralizes URL construction for SSE endpoints and audio retrieval.

```typescript
// app/src/lib/api/client.ts
class ApiClient {
  getGenerationStatusUrl(generationId: string): string {
    return `${this.getBaseUrl()}/generate/${generationId}/status`;
  }

  getAudioUrl(audioId: string): string {
    return `${this.getBaseUrl()}/audio/${audioId}`;
  }
  
  // ... additional methods
}

```

This abstraction ensures consistent base URL handling across environments and simplifies endpoint management within React components.

## Server-Side Implementation

### Status Streaming Endpoint

The FastAPI route in [`backend/routes/generations.py`](https://github.com/jamiepine/voicebox/blob/main/backend/routes/generations.py) implements the SSE protocol using Python's async generators and FastAPI's `StreamingResponse`.

```python

# backend/routes/generations.py

@router.get("/generate/{generation_id}/status")
async def get_generation_status(generation_id: str, db: Session = Depends(get_db)):
    """SSE endpoint that streams generation status updates."""
    import json

    async def event_stream():
        try:
            while True:
                db.expire_all()
                gen = db.query(DBGeneration).filter_by(id=generation_id).first()
                
                if not gen:
                    yield f"data: {json.dumps({'status': 'not_found', 'id': generation_id})}\n\n"
                    return

                payload = {
                    "id": gen.id,
                    "status": gen.status or "completed",
                    "duration": gen.duration,
                    "error": gen.error,
                }
                yield f"data: {json.dumps(payload)}\n\n"

                if (gen.status or "completed") in ("completed", "failed"):
                    return

                await asyncio.sleep(1)
        except (BrokenPipeError, ConnectionResetError, asyncio.CancelledError):
            logger.debug("SSE client disconnected for generation %s", generation_id)

    return StreamingResponse(
        event_stream(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",
        },
    )

```

The coroutine queries the database every second, yielding properly formatted SSE data lines (`data: {...}\n\n`). The `X-Accel-Buffering: no` header prevents reverse proxies from buffering the stream, ensuring immediate delivery to clients. Disconnection exceptions are caught silently to prevent server-side errors when users close browser tabs.

### Direct Audio Streaming

For scenarios requiring immediate playback without disk persistence, Voicebox exposes a secondary streaming endpoint that transmits raw WAV bytes in chunks.

```python

# backend/routes/generations.py

@router.post("/generate/stream")
async def stream_speech(
    data: models.GenerationRequest,
    db: Session = Depends(get_db),
):
    """Generate speech and stream the WAV audio directly without saving to disk."""
    
    # ... model loading and audio generation logic ...

    wav_bytes = tts.audio_to_wav_bytes(audio, sample_rate)

    async def _wav_stream():
        try:
            chunk_size = 64 * 1024
            for i in range(0, len(wav_bytes), chunk_size):
                yield wav_bytes[i: i + chunk_size]
        except (BrokenPipeError, ConnectionResetError, asyncio.CancelledError):
            logger.debug("Client disconnected during audio stream")

    return StreamingResponse(
        _wav_stream(),
        media_type="audio/wav",
        headers={"Content-Disposition": 'attachment; filename="speech.wav"'},
    )

```

The 64 KB chunk size balances memory efficiency with streaming performance, allowing the client to begin playback while the server is still transmitting data.

## Complete Implementation Example

Integrating async generation into a React component requires coordinating the generation request with the progress hook.

```tsx
import { useGenerationStore } from '@/stores/generationStore';
import { apiClient } from '@/lib/api/client';
import { useGenerationProgress } from '@/hooks/useGenerationProgress';

export function GenerateButton({ text }: { text: string }) {
  const addPending = useGenerationStore(s => s.addPendingGeneration);
  
  useGenerationProgress(); // Registers SSE listeners automatically

  const handleGenerate = async () => {
    const resp = await apiClient.postGeneration({ 
      text, 
      profile_id: "default" 
    });
    addPending(resp.id); // Triggers SSE connection
  };

  return <button onClick={handleGenerate}>Generate Speech</button>;
}

```

When the user clicks the button, the component posts to the generation endpoint and immediately adds the returned ID to the pending store. The `useGenerationProgress` hook detects this new ID, opens an SSE connection, and manages all subsequent state updates including toast notifications, cache invalidation, and automatic audio playback.

## Summary

- **Voicebox uses native EventSource APIs** to maintain persistent connections for real-time generation progress, eliminating the need for client-side polling.
- **The `useGenerationProgress` hook** maintains a Map of active connections, ensuring one SSE stream per generation ID with automatic cleanup on completion or failure.
- **FastAPI's StreamingResponse** powers the backend, emitting JSON status updates every second via the `/generate/{id}/status` endpoint until reaching terminal states.
- **Two streaming modes exist**: SSE for progress metadata and direct binary streaming via `/generate/stream` for immediate audio playback without disk writes.
- **Robust error handling** catches disconnections at both network and application layers, preventing resource leaks and ensuring clean UI state transitions.

## Frequently Asked Questions

### How does Voicebox handle SSE connection failures?

The implementation includes multiple safeguards against connection instability. In [`useGenerationProgress.ts`](https://github.com/jamiepine/voicebox/blob/main/useGenerationProgress.ts), the `source.onerror` handler closes the EventSource, removes the ID from pending generations, and refetches the history query to sync state. On the backend, the `event_stream` coroutine catches `BrokenPipeError`, `ConnectionResetError`, and `asyncio.CancelledError` to handle abrupt client disconnections without crashing the server.

### What is the polling interval for generation status updates?

The backend explicitly throttles database queries using `await asyncio.sleep(1)` in the event loop. This one-second interval balances real-time responsiveness with database load, as implemented in [`backend/routes/generations.py`](https://github.com/jamiepine/voicebox/blob/main/backend/routes/generations.py). The client receives updates as Server-Sent Events rather than through active polling, reducing network overhead compared to traditional REST polling architectures.

### Can Voicebox stream audio before generation completes?

The current implementation supports two distinct streaming patterns. The `/generate/stream` endpoint streams raw WAV bytes in 64 KB chunks immediately after synthesis completes, but requires the full audio buffer to be ready first. True progressive streaming during generation would require chunked TTS output and WebSocket connections, which are not implemented in the current architecture according to the source code.

### How does the frontend know when to play audio automatically?

Auto-playback relies on the `autoplayOnGenerate` setting stored in the server configuration store (`useServerStore`). When the SSE stream reports `status: "completed"`, the `useGenerationProgress` hook checks `autoplayRef.current` and `isPlayingRef.current` before calling `setAudioWithAutoPlay` with the audio URL constructed via `apiClient.getAudioUrl(id)`. This ensures audio only plays automatically if the user enabled the setting and no other audio is currently playing.