विषय पर बढ़ें

Server-Sent Events (SSE)

🌐 एआई और मनुष्यों द्वारा किया गया अनुवाद

यह अनुवाद मनुष्यों के मार्गदर्शन में एआई द्वारा किया गया है। 🤝

इसमें मूल अर्थ को गलत समझने या अप्राकृतिक लगने आदि जैसी गलतियाँ हो सकती हैं। 🤖

आप हमें एआई LLM को बेहतर मार्गदर्शन करने में मदद करके इस अनुवाद को बेहतर बना सकते हैं।

अंग्रेज़ी संस्करण

आप Server-Sent Events (SSE) का उपयोग करके client को data stream कर सकते हैं।

यह Stream JSON Lines जैसा है, लेकिन text/event-stream format का उपयोग करता है, जिसे browsers EventSource API के साथ natively support करते हैं।

नोट

FastAPI 0.135.0 में जोड़ा गया।

Server-Sent Events क्या हैं?

SSE, HTTP के माध्यम से server से client तक data stream करने के लिए एक standard है।

हर event एक छोटा text block होता है जिसमें data, event, id, और retry जैसे "fields" होते हैं, जिन्हें खाली lines से अलग किया जाता है।

यह ऐसा दिखता है:

data: {"name": "Portal Gun", "price": 999.99}

data: {"name": "Plumbus", "price": 32.99}

SSE का उपयोग आमतौर पर AI chat streaming, live notifications, logs और observability, और अन्य मामलों में किया जाता है जहाँ server client को updates push करता है।

सुझाव

अगर आप binary data stream करना चाहते हैं, जैसे video या audio, तो advanced guide देखें: Data Stream करें.

FastAPI के साथ SSE Stream करें

FastAPI के साथ SSE stream करने के लिए, अपनी path operation function में yield का उपयोग करें और response_class=EventSourceResponse set करें।

EventSourceResponse को fastapi.sse से import करें:

from collections.abc import AsyncIterable, Iterable

from fastapi import FastAPI
from fastapi.sse import EventSourceResponse
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str | None


items = [
    Item(name="Plumbus", description="A multi-purpose household device."),
    Item(name="Portal Gun", description="A portal opening device."),
    Item(name="Meeseeks Box", description="A box that summons a Meeseeks."),
]


@app.get("/items/stream", response_class=EventSourceResponse)
async def sse_items() -> AsyncIterable[Item]:
    for item in items:
        yield item

# Code below omitted 👇
👀 Full file preview 
from collections.abc import AsyncIterable, Iterable

from fastapi import FastAPI
from fastapi.sse import EventSourceResponse
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str | None


items = [
    Item(name="Plumbus", description="A multi-purpose household device."),
    Item(name="Portal Gun", description="A portal opening device."),
    Item(name="Meeseeks Box", description="A box that summons a Meeseeks."),
]


@app.get("/items/stream", response_class=EventSourceResponse)
async def sse_items() -> AsyncIterable[Item]:
    for item in items:
        yield item


@app.get("/items/stream-no-async", response_class=EventSourceResponse)
def sse_items_no_async() -> Iterable[Item]:
    for item in items:
        yield item


@app.get("/items/stream-no-annotation", response_class=EventSourceResponse)
async def sse_items_no_annotation():
    for item in items:
        yield item


@app.get("/items/stream-no-async-no-annotation", response_class=EventSourceResponse)
def sse_items_no_async_no_annotation():
    for item in items:
        yield item

हर yielded item को JSON के रूप में encode किया जाता है और SSE event के data: field में भेजा जाता है।

अगर आप return type को AsyncIterable[Item] के रूप में declare करते हैं, तो FastAPI इसका उपयोग Pydantic के साथ data को validate, document, और serialize करने के लिए करेगा।

from collections.abc import AsyncIterable, Iterable

from fastapi import FastAPI
from fastapi.sse import EventSourceResponse
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str | None


items = [
    Item(name="Plumbus", description="A multi-purpose household device."),
    Item(name="Portal Gun", description="A portal opening device."),
    Item(name="Meeseeks Box", description="A box that summons a Meeseeks."),
]


@app.get("/items/stream", response_class=EventSourceResponse)
async def sse_items() -> AsyncIterable[Item]:
    for item in items:
        yield item

# Code below omitted 👇
👀 Full file preview 
from collections.abc import AsyncIterable, Iterable

from fastapi import FastAPI
from fastapi.sse import EventSourceResponse
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str | None


items = [
    Item(name="Plumbus", description="A multi-purpose household device."),
    Item(name="Portal Gun", description="A portal opening device."),
    Item(name="Meeseeks Box", description="A box that summons a Meeseeks."),
]


@app.get("/items/stream", response_class=EventSourceResponse)
async def sse_items() -> AsyncIterable[Item]:
    for item in items:
        yield item


@app.get("/items/stream-no-async", response_class=EventSourceResponse)
def sse_items_no_async() -> Iterable[Item]:
    for item in items:
        yield item


@app.get("/items/stream-no-annotation", response_class=EventSourceResponse)
async def sse_items_no_annotation():
    for item in items:
        yield item


@app.get("/items/stream-no-async-no-annotation", response_class=EventSourceResponse)
def sse_items_no_async_no_annotation():
    for item in items:
        yield item

सुझाव

क्योंकि Pydantic इसे Rust side में serialize करेगा, आपको return type declare न करने की तुलना में कहीं बेहतर performance मिलेगी।

Non-async path operation functions

आप नियमित def functions (बिना async) का भी उपयोग कर सकते हैं, और उसी तरह yield का उपयोग कर सकते हैं।

FastAPI सुनिश्चित करेगा कि यह सही तरीके से run हो ताकि यह event loop को block न करे।

क्योंकि इस मामले में function async नहीं है, सही return type Iterable[Item] होगा:

# Code above omitted 👆

@app.get("/items/stream-no-async", response_class=EventSourceResponse)
def sse_items_no_async() -> Iterable[Item]:
    for item in items:
        yield item

# Code below omitted 👇
👀 Full file preview 
from collections.abc import AsyncIterable, Iterable

from fastapi import FastAPI
from fastapi.sse import EventSourceResponse
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str | None


items = [
    Item(name="Plumbus", description="A multi-purpose household device."),
    Item(name="Portal Gun", description="A portal opening device."),
    Item(name="Meeseeks Box", description="A box that summons a Meeseeks."),
]


@app.get("/items/stream", response_class=EventSourceResponse)
async def sse_items() -> AsyncIterable[Item]:
    for item in items:
        yield item


@app.get("/items/stream-no-async", response_class=EventSourceResponse)
def sse_items_no_async() -> Iterable[Item]:
    for item in items:
        yield item


@app.get("/items/stream-no-annotation", response_class=EventSourceResponse)
async def sse_items_no_annotation():
    for item in items:
        yield item


@app.get("/items/stream-no-async-no-annotation", response_class=EventSourceResponse)
def sse_items_no_async_no_annotation():
    for item in items:
        yield item

कोई Return Type नहीं

आप return type को छोड़ भी सकते हैं। FastAPI data को convert करने और भेजने के लिए jsonable_encoder का उपयोग करेगा।

# Code above omitted 👆

@app.get("/items/stream-no-annotation", response_class=EventSourceResponse)
async def sse_items_no_annotation():
    for item in items:
        yield item

# Code below omitted 👇
👀 Full file preview 
from collections.abc import AsyncIterable, Iterable

from fastapi import FastAPI
from fastapi.sse import EventSourceResponse
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    description: str | None


items = [
    Item(name="Plumbus", description="A multi-purpose household device."),
    Item(name="Portal Gun", description="A portal opening device."),
    Item(name="Meeseeks Box", description="A box that summons a Meeseeks."),
]


@app.get("/items/stream", response_class=EventSourceResponse)
async def sse_items() -> AsyncIterable[Item]:
    for item in items:
        yield item


@app.get("/items/stream-no-async", response_class=EventSourceResponse)
def sse_items_no_async() -> Iterable[Item]:
    for item in items:
        yield item


@app.get("/items/stream-no-annotation", response_class=EventSourceResponse)
async def sse_items_no_annotation():
    for item in items:
        yield item


@app.get("/items/stream-no-async-no-annotation", response_class=EventSourceResponse)
def sse_items_no_async_no_annotation():
    for item in items:
        yield item

ServerSentEvent

अगर आपको event, id, retry, या comment जैसे SSE fields set करने की ज़रूरत है, तो आप plain data के बजाय ServerSentEvent objects yield कर सकते हैं।

ServerSentEvent को fastapi.sse से import करें:

from collections.abc import AsyncIterable

from fastapi import FastAPI
from fastapi.sse import EventSourceResponse, ServerSentEvent
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    price: float


items = [
    Item(name="Plumbus", price=32.99),
    Item(name="Portal Gun", price=999.99),
    Item(name="Meeseeks Box", price=49.99),
]


@app.get("/items/stream", response_class=EventSourceResponse)
async def stream_items() -> AsyncIterable[ServerSentEvent]:
    yield ServerSentEvent(comment="stream of item updates")
    for i, item in enumerate(items):
        yield ServerSentEvent(data=item, event="item_update", id=str(i + 1), retry=5000)

data field हमेशा JSON के रूप में encode किया जाता है। आप कोई भी value pass कर सकते हैं जिसे JSON के रूप में serialize किया जा सकता हो, जिसमें Pydantic models भी शामिल हैं।

Raw Data

अगर आपको JSON encoding के बिना data भेजना है, तो data के बजाय raw_data का उपयोग करें।

यह pre-formatted text, log lines, या [DONE] जैसे विशेष "sentinel" values भेजने के लिए उपयोगी है।

from collections.abc import AsyncIterable

from fastapi import FastAPI
from fastapi.sse import EventSourceResponse, ServerSentEvent

app = FastAPI()


@app.get("/logs/stream", response_class=EventSourceResponse)
async def stream_logs() -> AsyncIterable[ServerSentEvent]:
    logs = [
        "2025-01-01 INFO  Application started",
        "2025-01-01 DEBUG Connected to database",
        "2025-01-01 WARN  High memory usage detected",
    ]
    for log_line in logs:
        yield ServerSentEvent(raw_data=log_line)

नोट

data और raw_data mutually exclusive हैं। आप हर ServerSentEvent पर उनमें से केवल एक ही set कर सकते हैं।

Last-Event-ID के साथ फिर से शुरू करना

जब कोई browser connection drop होने के बाद reconnect करता है, तो वह अंतिम प्राप्त id को Last-Event-ID header में भेजता है।

आप इसे header parameter के रूप में read कर सकते हैं और stream को वहाँ से resume करने के लिए उपयोग कर सकते हैं जहाँ client ने छोड़ा था:

from collections.abc import AsyncIterable
from typing import Annotated

from fastapi import FastAPI, Header
from fastapi.sse import EventSourceResponse, ServerSentEvent
from pydantic import BaseModel

app = FastAPI()


class Item(BaseModel):
    name: str
    price: float


items = [
    Item(name="Plumbus", price=32.99),
    Item(name="Portal Gun", price=999.99),
    Item(name="Meeseeks Box", price=49.99),
]


@app.get("/items/stream", response_class=EventSourceResponse)
async def stream_items(
    last_event_id: Annotated[int | None, Header()] = None,
) -> AsyncIterable[ServerSentEvent]:
    start = last_event_id + 1 if last_event_id is not None else 0
    for i, item in enumerate(items):
        if i < start:
            continue
        yield ServerSentEvent(data=item, id=str(i))

POST के साथ SSE

SSE किसी भी HTTP method के साथ काम करता है, केवल GET के साथ नहीं।

यह MCP जैसे protocols के लिए उपयोगी है, जो POST पर SSE stream करते हैं:

from collections.abc import AsyncIterable

from fastapi import FastAPI
from fastapi.sse import EventSourceResponse, ServerSentEvent
from pydantic import BaseModel

app = FastAPI()


class Prompt(BaseModel):
    text: str


@app.post("/chat/stream", response_class=EventSourceResponse)
async def stream_chat(prompt: Prompt) -> AsyncIterable[ServerSentEvent]:
    words = prompt.text.split()
    for word in words:
        yield ServerSentEvent(data=word, event="token")
    yield ServerSentEvent(raw_data="[DONE]", event="done")

तकनीकी विवरण

FastAPI कुछ SSE best practices को out of the box implement करता है।

  • जब कोई message नहीं आया हो, तो हर 15 सेकंड में "keep alive" ping comment भेजें, ताकि कुछ proxies connection को close न कर दें, जैसा कि HTML specification: Server-Sent Events में सुझाया गया है।
  • stream की caching रोकने के लिए Cache-Control: no-cache header set करें।
  • Nginx जैसे कुछ proxies में buffering रोकने के लिए एक special header X-Accel-Buffering: no set करें।

आपको इसके लिए कुछ भी करने की ज़रूरत नहीं है, यह out of the box काम करता है। 🤓