Commands & Queries (RPC)
Request-reply RPC with KubeMQ FastStream — commands return execution status, queries return data with optional caching.
Overview
Commands and Queries are KubeMQ's two request-reply patterns → see 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.
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.
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:
@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
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:
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
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.
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()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.
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.
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:
@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
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.
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
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:
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.
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.
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:
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:
@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}")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 for all constructor options.
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.
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:
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
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.
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=:
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
Commands and queries are the building blocks for distributed request-reply architectures.
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.
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 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.
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
RPC (core concept)
The broker-level model behind commands and queries: request-reply, void vs data response, and caching.
Queues
Point-to-point messaging with AckPolicy-controlled settlement and batch publishing — use when you need durable delivery rather than a synchronous reply.
Getting Started
Install kubemq-faststream, start a broker, and try all five patterns side by side.
Reference
The broker.request() signature, default_cq_timeout, and the full API surface.
Was this page helpful?