# Stream Receive (/sdks/python/how-to/queues/stream-receive)



## Overview [#overview]

A **downstream receiver** is the persistent-connection way to pull queue messages: instead of opening and tearing down a request for every batch, you open one gRPC stream and reuse it across many receive cycles. That matters for any consumer that runs continuously — a worker loop, a background processor — where reconnecting per batch would add latency and churn on both the client and the broker.

`receive_queue_messages` fetches a batch under manual settlement — no `auto_ack` — so nothing is removed from the queue until you explicitly settle it. Each message is settled on its own: `async_ack()` permanently removes it, `async_nack()` returns it for redelivery, and leaving it unsettled returns it once the visibility timeout expires.

**Gotchas:** a crash between receiving and settling redelivers the whole batch, so processing must be idempotent; forgetting to settle a message just delays its redelivery until the timeout, it doesn't drop it; and mixing ack/nack per message, as shown here, only makes sense for per-item failures — a batch-wide failure is usually better handled by nacking everything and retrying later.

## Prerequisites [#prerequisites]

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

## Code [#code]

```python title="stream_receive.py"
"""Example: Stream receive — receive messages with ack/reject/requeue options."""

from __future__ import annotations

import asyncio

from kubemq import AsyncQueuesClient
from kubemq import QueueMessage


async def main() -> None:
    async with AsyncQueuesClient(
        address="localhost:50000",
        client_id="python-queues-stream-stream-receive-client",
    ) as client:
        # Send some test messages
        for i in range(3):
            await client.send_queue_message(
                QueueMessage(
                    channel="python-queues-stream.stream-receive",
                    body=f"Message #{i + 1}".encode(),
                )
            )
        print("Sent 3 test messages")

        # Receive messages via streaming
        response = await client.receive_queue_messages(
            channel="python-queues-stream.stream-receive",
            max_messages=10,
            wait_timeout_seconds=5,
        )

        for msg in response.messages:
            body = msg.body.decode("utf-8")
            print(f"Received: {body}")

            # Demonstrate different acknowledgment strategies
            if "1" in body:
                await msg.async_ack()
                print(f"  -> Acknowledged: {body}")
            elif "2" in body:
                await msg.async_nack()
                print(f"  -> Rejected: {body}")
            else:
                await msg.async_ack()
                print(f"  -> Acknowledged: {body}")


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

```

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

* `receive_queue_messages` polls all available messages in a single call (up to `max_messages`), returning them as a list without `auto_ack`.
* Each message is settled individually: `async_ack()` permanently removes it from the queue; `async_nack()` returns it for redelivery.
* The example demonstrates inline decision logic inside the settlement loop, which is the typical pattern for content-based routing to ack or nack.
* Without `auto_ack`, the broker holds messages in a visibility-timeout state until settled or until the timeout expires and returns them.

## Related [#related]

* [Pattern overview](/learn/queues/getting-started)
* [Python SDK Reference](/sdks/python/reference/queues)
* [Send & Receive](/sdks/python/tutorials/send-receive)
* [Ack All](/sdks/python/how-to/queues/ack-all)