# Query Group (/sdks/python/how-to/rpc/query-group)



## Overview [#overview]

A **consumer group** scales query handling horizontally without touching the caller's side. Instead of one process answering every query on a channel, you run several identical handler instances under the same group name, and the broker routes each query to exactly one member — never to all of them. That turns a single responder into a pool you can grow or shrink to match load, which matters for anything RPC-shaped: a lookup service, a cache-fill handler, a synchronous read path behind an API.

It works by tying group membership to the subscription: `QueriesSubscription(channel=..., group=...)` passed to `client.subscribe_to_queries` load-balances across every subscriber sharing that channel and group. The sender calls `client.send_query` exactly as it would against a single handler — it never knows how many members exist or which one answered.

**Gotchas:** channel and group name must match exactly, or a typo quietly creates a second, empty group instead of erroring. Omit `group` and every subscriber reverts to broadcast, each answering independently. A stuck group member isn't bypassed — the caller just sees a timeout.

## Prerequisites [#prerequisites]

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

## Code [#code]

```python title="consumer_group.py"
"""Example: Consumer group — load-balance queries across multiple responders."""

from __future__ import annotations

import asyncio

from kubemq import (
    AsyncCancellationToken,
    AsyncCQClient,
    QueriesSubscription,
    QueryMessage,
    QueryResponse,
)


async def main() -> None:
    async with AsyncCQClient(
        address="localhost:50000",
        client_id="python-queries-consumer-group-client",
    ) as client:
        token = AsyncCancellationToken()

        async def make_responder(name: str) -> None:
            async for query in client.subscribe_to_queries(
                subscription=QueriesSubscription(
                    channel="python-queries.consumer-group",
                    group="responders",
                    on_receive_query_callback=lambda q: None,
                    on_error_callback=lambda e: print(f"Error: {e}"),
                ),
                cancellation_token=token,
            ):
                print(f"[{name}] Received query: {query.body.decode('utf-8')}")
                await client.send_response(
                    QueryResponse(
                        query_received=query,
                        is_executed=True,
                        body=f"Response from {name}".encode(),
                    )
                )

        task1 = asyncio.create_task(make_responder("Responder-1"))
        task2 = asyncio.create_task(make_responder("Responder-2"))
        await asyncio.sleep(1)

        for i in range(4):
            response = await client.send_query(
                QueryMessage(
                    channel="python-queries.consumer-group",
                    body=f"Query #{i + 1}".encode(),
                    timeout_in_seconds=10,
                )
            )
            print(f"Query #{i + 1} response: {response.body}")

        token.cancel()
        for t in [task1, task2]:
            t.cancel()
            try:
                await t
            except asyncio.CancelledError:
                pass


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

```

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

Both responders subscribe to the same channel with `group="responders"`. KubeMQ routes each query to exactly one responder in the group (round-robin or least-loaded, depending on server config). Each responder returns `body=f"Response from {name}".encode()` so you can see in the output which instance handled each query. The load-balancing happens entirely on the broker side — the sender always calls `send_query` the same way regardless of how many responders are running.

## Related [#related]

* [Pattern overview](/learn/rpc/getting-started)
* [Python SDK Reference](/sdks/python/reference/rpc)
* [Send Query](/sdks/python/tutorials/query-send)
* [Handle Query](/sdks/python/how-to/rpc/query-handle)