# Commands & Queries (RPC) (/integrations/faststream/how-to/commands-queries)



## Overview [#overview]

Commands and Queries are KubeMQ's two **request-reply** patterns → see [RPC](/learn/rpc)
for the broker-level model of commands (fire-and-wait) and queries (ask-and-receive with
optional caching).

In `kubemq-faststream`, both build on the same FastStream primitive,
`broker.request(...)`, which sends a message and blocks until a matching subscriber
replies. You register a handler with `@broker.subscriber(commands=...)` or
`@broker.subscriber(queries=...)`, and the handler's return value is sent back to the
caller. The difference is intent: a command confirms an action ran (the natural response
is an execution **status**), a query returns **data** and can be cached server-side.

<Mermaid
  chart="sequenceDiagram
    participant C as Caller
    participant B as KubeMQBroker
    participant K as KubeMQ
    participant H as Handler
    C->>B: await broker.request(payload, commands=/queries=...)
    B->>K: dispatch on AsyncCQClient
    K->>H: deliver request
    H-->>K: return value (status or data)
    K-->>B: response
    B-->>C: response (caller unblocks)"
/>

## Commands [#commands]

A command handler subscribes with `commands=<channel>`. The caller uses `broker.request(...)` with the same `commands=` keyword and blocks until the handler returns. Pass a `timeout` (in seconds) to bound how long the caller waits.

```python title="basic_command.py"
import asyncio
import logging

from faststream import FastStream

from kubemq_faststream import KubeMQBroker

logging.basicConfig(level=logging.INFO)

broker = KubeMQBroker("kubemq://localhost:50000")
app = FastStream(broker)

CHANNEL = "example.commands.basic"


@broker.subscriber(commands=CHANNEL)
async def handle_command(msg: dict) -> dict:
    """Execute the command and return a result."""
    action = msg.get("action", "unknown")
    print(f"Executing command: {action}")
    return {"status": "executed", "action": action, "success": True}


@app.after_startup
async def run_demo() -> None:
    """Send a command and inspect the response."""
    print("Sending command...")
    response = await broker.request(
        {"action": "create_user", "name": "Alice"},
        commands=CHANNEL,
    )
    print(f"Command response: {response}")

    await asyncio.sleep(2)
    print("Demo complete")
    await app.stop()


if __name__ == "__main__":
    asyncio.run(app.run())
```

The README's canonical example uses a `device.restart` command — the sender blocks on `broker.request(...)` until the handler confirms the restart completed:

```python title="device_restart.py"
@broker.subscriber(commands="device.restart")
async def restart_device(msg: dict) -> None:
    device_id = msg["device_id"]
    # ... perform restart ...
    print(f"Device {device_id} restarted")


# Send command and wait for completion
await broker.request(
    {"device_id": "sensor-01"},
    commands="device.restart",
    timeout=10,
)
```

A command handler may return `None` (a pure side-effect with no payload) or a `dict` carrying execution details. Either way, `broker.request(...)` only unblocks once the handler has run, giving you a hard delivery-and-execution confirmation.

### Inspecting the Command Response [#inspecting-the-command-response]

When the handler returns a structured `dict`, the caller can inspect the execution outcome — for example, distinguishing an `executed` status from an `error`. The handler below simulates a deployment and returns rich detail the caller then unpacks:

```python title="command_response.py"
CHANNEL = "example.commands.response"


@broker.subscriber(commands=CHANNEL)
async def handle_deploy(msg: dict) -> dict:
    """Simulate a deployment command with detailed response."""
    service = msg.get("service", "unknown")
    version = msg.get("version", "0.0.0")
    print(f"Deploying {service} v{version}...")

    return {
        "status": "deployed",
        "service": service,
        "version": version,
        "replicas": 3,
        "healthy": True,
    }


@app.after_startup
async def run_demo() -> None:
    """Send a deploy command and inspect the full response."""
    print("Sending deploy command...")
    response = await broker.request(
        {"service": "payment-api", "version": "2.1.0"},
        commands=CHANNEL,
    )
    print(f"Raw response: {response}")

    print("\nParsed response details:")
    if isinstance(response, dict):
        print(f"  Status:   {response.get('status')}")
        print(f"  Service:  {response.get('service')}")
        print(f"  Version:  {response.get('version')}")
        print(f"  Replicas: {response.get('replicas')}")
        print(f"  Healthy:  {response.get('healthy')}")

    await asyncio.sleep(2)
    print("\nDemo complete")
    await app.stop()
```

### Fire-and-Forget Commands [#fire-and-forget-commands]

If you do not need the execution confirmation, declare the subscriber with `no_reply=True`. The handler still runs, but it does **not** send a response. The caller then uses `broker.publish(commands=...)` instead of `broker.request(...)`, since there is nothing to wait for.

```python title="no_reply_command.py"
CHANNEL = "example.commands.noreply"


@broker.subscriber(commands=CHANNEL, no_reply=True)
async def handle_command(msg: dict) -> None:
    """Execute the command without sending a response."""
    action = msg.get("action", "unknown")
    print(f"[handler] Executing fire-and-forget command: {action}")


@app.after_startup
async def run_demo() -> None:
    """Publish commands without waiting for a response."""
    commands_to_send = [
        {"action": "restart_service", "target": "web-server-01"},
        {"action": "clear_cache", "target": "cache-node-03"},
    ]

    for cmd in commands_to_send:
        await broker.publish(cmd, commands=CHANNEL)
        print("[caller] Command published (no response expected)")

    await asyncio.sleep(2)
    print("Demo complete")
    await app.stop()
```

<Callout type="warn">
  With `no_reply=True` you must publish, not request. Calling `broker.request(...)` against a `no_reply` handler would block until the caller's timeout, because the handler never sends a reply.
</Callout>

## Queries [#queries]

A query handler subscribes with `queries=<channel>` and returns a `dict` of data. The caller invokes it with `broker.request(..., queries=...)` and receives that data back as the response.

```python title="basic_query.py"
import asyncio
import logging

from faststream import FastStream

from kubemq_faststream import KubeMQBroker

logging.basicConfig(level=logging.INFO)

broker = KubeMQBroker("kubemq://localhost:50000")
app = FastStream(broker)

CHANNEL = "example.queries.basic"


@broker.subscriber(queries=CHANNEL)
async def handle_query(msg: dict) -> dict:
    """Process the query and return data."""
    user_id = msg.get("user_id")
    print(f"Looking up user {user_id}...")
    return {
        "user_id": user_id,
        "name": "Alice",
        "email": "alice@example.com",
        "active": True,
    }


@app.after_startup
async def run_demo() -> None:
    """Send a query and print the response."""
    print("Sending query...")
    response = await broker.request(
        {"user_id": 42},
        queries=CHANNEL,
    )
    print(f"Query response: {response}")

    await asyncio.sleep(2)
    print("Demo complete")
    await app.stop()


if __name__ == "__main__":
    asyncio.run(app.run())
```

The README's `product.lookup` example is the same shape — a handler returning a data `dict`, invoked with a 10-second timeout:

```python title="product_lookup.py"
@broker.subscriber(queries="product.lookup")
async def lookup_product(msg: dict) -> dict:
    return {"name": "Widget", "price": 29.99}


# Send query and get response
result = await broker.request(
    {"product_id": "SKU-100"},
    queries="product.lookup",
    timeout=10,
)
```

Like commands, queries support `no_reply=True` for endpoints that ingest data without returning anything — analytics or audit logging, for instance. In that mode the caller publishes with `broker.publish(queries=...)` rather than requesting.

## Server-Side Response Caching [#server-side-response-caching]

Queries can be cached on the broker. Pass `cache_key` and `cache_ttl` (seconds) on the request: the **first** call with a given key invokes the handler and the broker stores the response; subsequent calls with the same key within the TTL window are served from cache **without invoking the handler**.

```python title="cache_ttl.py"
CHANNEL = "example.queries.cache_ttl"

call_count = 0


@broker.subscriber(queries=CHANNEL)
async def handle_query(msg: dict) -> dict:
    """Handler that tracks invocation count."""
    global call_count  # noqa: PLW0603
    call_count += 1
    print(f"Handler invoked (call #{call_count})")
    return {
        "data": msg.get("key", "default"),
        "computed_at": time.time(),
        "call_count": call_count,
    }


@app.after_startup
async def run_demo() -> None:
    """Send queries with caching enabled."""
    print("--- First request (cache miss, handler invoked) ---")
    r1 = await broker.request(
        {"key": "config-v1"},
        queries=CHANNEL,
        cache_key="config-lookup",
        cache_ttl=30,
    )
    print(f"Response 1: {r1}")

    print("\n--- Second request (cache hit, handler NOT invoked) ---")
    r2 = await broker.request(
        {"key": "config-v1"},
        queries=CHANNEL,
        cache_key="config-lookup",
        cache_ttl=30,
    )
    print(f"Response 2: {r2}")

    print(f"\nHandler was invoked {call_count} time(s) — second call served from cache")
```

### Detecting Cache Hits vs Misses [#detecting-cache-hits-vs-misses]

A practical way to confirm caching is to count handler invocations. A cache **miss** runs the handler; a cache **hit** does not. Reusing the same `cache_key` returns the cached response, while a different key forces a fresh miss:

```python title="cache_hit_miss.py"
CHANNEL = "example.queries.cache_hit"

handler_calls = 0


@broker.subscriber(queries=CHANNEL)
async def lookup_product(msg: dict) -> dict:
    """Product lookup that should be cached."""
    global handler_calls  # noqa: PLW0603
    handler_calls += 1
    product_id = msg.get("product_id")
    print(f"[Handler] Looking up product {product_id} (call #{handler_calls})")
    return {
        "product_id": product_id,
        "name": "Widget Pro",
        "price": 29.99,
        "in_stock": True,
    }


@app.after_startup
async def run_demo() -> None:
    """Demonstrate cache hit vs miss."""
    cache_key = "product-42"
    cache_ttl = 60

    print("=== Request 1: Cache MISS (handler will be invoked) ===")
    r1 = await broker.request(
        {"product_id": 42},
        queries=CHANNEL,
        cache_key=cache_key,
        cache_ttl=cache_ttl,
    )
    print(f"Response: {r1}")
    print(f"Handler invocations so far: {handler_calls}")

    print("\n=== Request 2: Cache HIT (handler NOT invoked) ===")
    r2 = await broker.request(
        {"product_id": 42},
        queries=CHANNEL,
        cache_key=cache_key,
        cache_ttl=cache_ttl,
    )
    print(f"Response: {r2}")
    print(f"Handler invocations so far: {handler_calls}")

    print("\n=== Request 3: Different key — Cache MISS ===")
    r3 = await broker.request(
        {"product_id": 99},
        queries=CHANNEL,
        cache_key="product-99",
        cache_ttl=cache_ttl,
    )
    print(f"Response: {r3}")
    print(f"Handler invocations so far: {handler_calls}")
```

Running this prints `1`, `1`, then `2` invocations: request 2 is served from cache (the counter stays flat), and request 3 — with a new `cache_key` — misses and runs the handler again.

<Callout type="info">
  Caching is keyed entirely by `cache_key`, not by the request payload. Two requests with the same `cache_key` return the same cached response even if their payloads differ — choose keys that uniquely identify the data you are caching.
</Callout>

## Timeout Handling [#timeout-handling]

Both `broker.request(...)` calls accept a `timeout` (seconds). If the handler does not respond before the deadline, the request raises an exception on the caller side. This protects callers from hanging on a slow or unavailable handler.

The command example below sends one fast request that succeeds within a 10-second timeout, then a slow one that exceeds a 2-second timeout and raises:

```python title="timeout_handling.py"
CHANNEL = "example.commands.timeout"


@broker.subscriber(commands=CHANNEL)
async def slow_handler(msg: dict) -> dict:
    """Simulate a handler that may exceed the timeout."""
    delay = msg.get("delay_seconds", 0)
    print(f"Handler working for {delay}s...")
    await asyncio.sleep(delay)
    return {"status": "done", "delay": delay}


@app.after_startup
async def run_demo() -> None:
    """Send commands with different timeout scenarios."""
    print("--- Fast command (should succeed) ---")
    try:
        response = await broker.request(
            {"action": "fast_task", "delay_seconds": 1},
            commands=CHANNEL,
            timeout=10,
        )
        print(f"Success: {response}")
    except Exception as exc:
        print(f"Error: {exc}")

    print("\n--- Slow command (should timeout) ---")
    try:
        response = await broker.request(
            {"action": "slow_task", "delay_seconds": 15},
            commands=CHANNEL,
            timeout=2,
        )
        print(f"Success: {response}")
    except Exception as exc:
        print(f"Timeout error (expected): {exc}")
```

Queries behave identically — wrap `broker.request(..., queries=...)` in a `try/except` and bound it with `timeout`:

```python title="query_timeout.py"
@broker.subscriber(queries="example.queries.timeout")
async def slow_query_handler(msg: dict) -> dict:
    """Simulate a query that may take too long."""
    delay = msg.get("delay_seconds", 0)
    print(f"Processing query (will take {delay}s)...")
    await asyncio.sleep(delay)
    return {"result": "computed", "delay": delay}


# Slow query (should timeout)
try:
    response = await broker.request(
        {"query": "slow_aggregation", "delay_seconds": 15},
        queries="example.queries.timeout",
        timeout=2,
    )
    print(f"Success: {response}")
except Exception as exc:
    print(f"Timeout error (expected): {exc}")
```

<Callout type="info">
  The broker's `default_cq_timeout` constructor option (default `30` seconds, or `KUBEMQ_DEFAULT_CQ_TIMEOUT`) sets the fallback timeout when you omit `timeout` on a request. See the [configuration reference](/integrations/faststream/reference/configuration) for all constructor options.
</Callout>

## Multiple Handlers with Group Load Balancing [#multiple-handlers-with-group-load-balancing]

When several subscribers listen on the **same** channel with the **same** `group`, the broker load-balances requests across them: each request is processed by exactly one handler in the group. This is how you scale RPC throughput horizontally — add more handler instances to the group.

```python title="commands/multiple_handlers.py"
CHANNEL = "example.commands.multi"
GROUP = "cmd-workers"


@broker.subscriber(commands=CHANNEL, group=GROUP)
async def handler_a(msg: dict) -> dict:
    """Command handler A."""
    print(f"[Handler-A] Executing: {msg}")
    return {"handler": "A", "status": "executed"}


@broker.subscriber(commands=CHANNEL, group=GROUP)
async def handler_b(msg: dict) -> dict:
    """Command handler B."""
    print(f"[Handler-B] Executing: {msg}")
    return {"handler": "B", "status": "executed"}


@app.after_startup
async def run_demo() -> None:
    """Send several commands that are load-balanced across handlers."""
    for i in range(1, 5):
        print(f"\nSending command {i}...")
        response = await broker.request(
            {"command_id": i, "action": f"task-{i}"},
            commands=CHANNEL,
        )
        print(f"Response from command {i}: {response}")
```

Queries use the identical pattern — two subscribers on the same `queries` channel sharing a `group`:

```python title="queries/multiple_handlers.py"
CHANNEL = "example.queries.multi"
GROUP = "query-workers"


@broker.subscriber(queries=CHANNEL, group=GROUP)
async def handler_a(msg: dict) -> dict:
    """Query handler A."""
    print(f"[Handler-A] Processing query: {msg}")
    return {"handler": "A", "result": msg.get("key", "none")}


@broker.subscriber(queries=CHANNEL, group=GROUP)
async def handler_b(msg: dict) -> dict:
    """Query handler B."""
    print(f"[Handler-B] Processing query: {msg}")
    return {"handler": "B", "result": msg.get("key", "none")}
```

Across the four requests, responses alternate between handler A and handler B, demonstrating that the broker distributes each request to a single group member rather than fanning out to all of them.

## Batch RPC [#batch-rpc]

To issue several RPC requests in a single call, use `broker.request_batch(...)`. All requests are dispatched together and their responses collected. Each request is processed independently by the subscriber. Optional `headers` and `metadata` are attached to the batch.

```python title="batch_operations/commands_batch.py"
CHANNEL = "example.batch.commands"


@broker.subscriber(commands=CHANNEL)
async def handle_command(msg: dict) -> dict:
    """Execute a deployment command and return the result."""
    action = msg.get("action", "unknown")
    target = msg.get("target", "unknown")
    print(f"Executing command: {action} {target}")
    return {"status": "executed", "action": action, "target": target, "success": True}


@app.after_startup
async def run_demo() -> None:
    """Send a batch of commands and inspect the responses."""
    commands_to_send = [
        {"action": "deploy", "target": "service-1"},
        {"action": "deploy", "target": "service-2"},
        {"action": "deploy", "target": "service-3"},
    ]

    responses = await broker.request_batch(
        *commands_to_send,
        commands=CHANNEL,
        timeout=10,
        headers={"batch-id": "deploy-batch-001"},
        metadata="deployment-batch",
    )
    print(f"Batch command responses: {responses}")
```

Batched queries work the same way — spread the request payloads with `*` and pass `queries=`:

```python title="batch_operations/queries_batch.py"
CHANNEL = "example.batch.queries"

USERS_DB = {
    101: {"name": "Alice", "email": "alice@example.com", "active": True},
    102: {"name": "Bob", "email": "bob@example.com", "active": True},
    103: {"name": "Carol", "email": "carol@example.com", "active": False},
}


@broker.subscriber(queries=CHANNEL)
async def handle_query(msg: dict) -> dict:
    """Look up a user by ID and return their profile."""
    user_id = msg.get("user_id")
    print(f"Looking up user {user_id}")
    user = USERS_DB.get(user_id, {"error": "not found"})
    return {"user_id": user_id, **user}


@app.after_startup
async def run_demo() -> None:
    """Send a batch of queries and inspect the responses."""
    queries_to_send = [
        {"user_id": 101},
        {"user_id": 102},
        {"user_id": 103},
    ]

    responses = await broker.request_batch(
        *queries_to_send,
        queries=CHANNEL,
        timeout=10,
        headers={"batch-id": "user-lookup-batch"},
        metadata="user-batch-query",
    )
    print(f"Batch query responses: {responses}")
```

## Distributed RPC Patterns [#distributed-rpc-patterns]

Commands and queries are the building blocks for distributed request-reply architectures.

### Request-Reply [#request-reply]

The request-reply pattern uses commands and queries side by side in one application: commands to *do* something and confirm it, queries to *ask* something and get data back.

```python title="patterns/request_reply.py"
COMMAND_CHANNEL = "example.patterns.reqreply.commands"
QUERY_CHANNEL = "example.patterns.reqreply.queries"


# --- Command handler: execute an action and confirm ---
@broker.subscriber(commands=COMMAND_CHANNEL)
async def handle_command(msg: dict) -> dict:
    """Execute the command and return confirmation."""
    action = msg.get("action", "unknown")
    print(f"[command handler] Executing: {action}")
    return {"status": "executed", "action": action, "success": True}


# --- Query handler: look up data and return it ---
@broker.subscriber(queries=QUERY_CHANNEL)
async def handle_query(msg: dict) -> dict:
    """Process the query and return data."""
    user_id = msg.get("user_id")
    print(f"[query handler] Looking up user_id={user_id}")
    return {"user_id": user_id, "name": "Alice", "email": "alice@example.com"}


@app.after_startup
async def run_demo() -> None:
    """Send a command and a query, printing the responses."""
    cmd_response = await broker.request(
        {"action": "create_user", "name": "Alice"},
        commands=COMMAND_CHANNEL,
    )
    print(f"[caller] Command response: {cmd_response}")

    query_response = await broker.request(
        {"user_id": 42},
        queries=QUERY_CHANNEL,
    )
    print(f"[caller] Query response: {query_response}")
```

### Scatter-Gather [#scatter-gather]

Scatter-gather sends out multiple queries and collects the responses. Combined with `group`-based load balancing, one of several handler instances answers each query — the foundation for aggregating data across a worker pool.

```python title="patterns/scatter_gather.py"
QUERY_CH = "example.patterns.scatter.pricing"


@broker.subscriber(queries=QUERY_CH, group="pricing-workers")
async def pricing_handler(msg: dict) -> dict:
    """Return a price quote for the requested item."""
    item = msg.get("item", "unknown")
    print(f"[Pricing] Computing price for: {item}")
    prices = {"widget": 9.99, "gadget": 24.99, "gizmo": 14.99}
    return {"item": item, "price": prices.get(item, 0.0), "currency": "USD"}


@app.after_startup
async def run_demo() -> None:
    """Send multiple queries and gather responses."""
    items = ["widget", "gadget", "gizmo"]
    responses = []

    for item in items:
        resp = await broker.request(
            {"item": item},
            queries=QUERY_CH,
            timeout=10,
        )
        responses.append(resp)
        print(f"Got price for {item}: {resp}")

    print(f"\nGathered {len(responses)} responses")
```

## Related [#related]

<Cards>
  <Card title="RPC (core concept)" href="/learn/rpc" description="The broker-level model behind commands and queries: request-reply, void vs data response, and caching." />

  <Card title="Queues" href="/integrations/faststream/how-to/queues" description="Point-to-point messaging with AckPolicy-controlled settlement and batch publishing — use when you need durable delivery rather than a synchronous reply." />

  <Card title="Getting Started" href="/integrations/faststream/tutorials/getting-started" description="Install kubemq-faststream, start a broker, and try all five patterns side by side." />

  <Card title="Reference" href="/integrations/faststream/reference/api" description="The broker.request() signature, default_cq_timeout, and the full API surface." />
</Cards>