# Stream Send (/sdks/python/how-to/events-store/stream-send)



## Overview [#overview]

<Callout type="info" title="Which to use">
  This page covers high-throughput **Events Store** (persistent, replayable) streaming via `send_event_store`. For the fire-and-forget equivalent, see [Events Stream Send](/sdks/python/how-to/events/stream-send).
</Callout>

**Stream send** covers publishing a batch of persistent events back-to-back, without pausing your producer between messages. The single-shot `send_event_store` call is fine for one-off writes, but if you're bulk-loading history, replicating a firehose of records, or backfilling an Events Store channel, sending events in rapid succession — while a subscriber consumes them concurrently — turns network latency into your throughput ceiling instead of an app-level bottleneck.

Each `send_event_store` call still confirms storage synchronously, returning an `EventStoreResult` with the broker-assigned `id` and a `sent` flag, and each persisted event gets a monotonically increasing `sequence` number. Running the sends in a loop while a concurrent `subscribe_to_events_store` task drains the channel pipelines production and consumption instead of serializing them. &#x2A;*Gotchas:** each call still blocks on its own confirmation, so very high fan-out workloads benefit from concurrent sends rather than one tight loop; the subscriber must be running *before* you publish if you want every event, since Events Store subscriptions don't retroactively grab messages sent before they connected; and don't rely on wall-clock sleeps to guarantee delivery — use the returned `sequence` instead.

## Prerequisites [#prerequisites]

* KubeMQ server running on `localhost:50000`
* Python SDK installed (`pip install kubemq`)

## Code [#code]

```python title="stream_send.py"
"""Example: Stream send — send events store messages via the bidirectional stream path."""

from __future__ import annotations

import asyncio

from kubemq import (
    AsyncCancellationToken,
    AsyncPubSubClient,
    EventStoreMessage,
    EventsStoreSubscription,
)
from kubemq.pubsub import EventStoreStartPosition


async def main() -> None:
    async with AsyncPubSubClient(
        address="localhost:50000",
        client_id="python-events-store-stream-send-client",
    ) as client:
        token = AsyncCancellationToken()
        received: list[str] = []

        async def subscriber() -> None:
            async for event in client.subscribe_to_events_store(
                subscription=EventsStoreSubscription(
                    channel="python-events-store.stream-send",
                    on_receive_event_callback=lambda e: None,
                    on_error_callback=lambda e: print(f"Error: {e}"),
                    events_store_type=EventStoreStartPosition.StartFromNew,
                ),
                cancellation_token=token,
            ):
                received.append(event.body.decode("utf-8"))
                print(f"Received seq={event.sequence}: {event.body.decode('utf-8')}")

        task = asyncio.create_task(subscriber())
        await asyncio.sleep(1)

        for i in range(10):
            result = await client.send_event_store(
                EventStoreMessage(
                    channel="python-events-store.stream-send",
                    body=f"StoreEvent-{i + 1}".encode(),
                )
            )
            print(f"Sent: id={result.id}, sent={result.sent}")

        await asyncio.sleep(3)
        print(f"Received {len(received)} events store messages")
        token.cancel()
        task.cancel()
        try:
            await task
        except asyncio.CancelledError:
            pass


if __name__ == "__main__":
    asyncio.run(main())

```

## How It Works [#how-it-works]

* Each `send_event_store` call returns an `EventStoreResult` with `id` and `sent=True`, confirming persistence on the broker.
* The subscriber accumulates received events in `received`; the final count verifies delivery of all 10 persisted messages.
* Messages are sent sequentially in the loop; the broker persists each one atomically and assigns an increasing `sequence` number.
* The 3-second sleep after sending allows the broker to forward all queued events to the subscriber before the token is cancelled.

## Related [#related]

* [Pattern overview](/learn/events-store/getting-started)
* [Python SDK Reference](/sdks/python/reference/events-store)
* [Persistent Pub/Sub](/sdks/python/tutorials/persistent-pubsub)
* [Cancel Subscription](/sdks/python/how-to/events-store/cancel-subscription)