Implement RPC over RabbitMQ with AMQP (2026)

RabbitMQ can act as a high-performance RPC (Remote Procedure Call) broker, letting you treat message queues like function calls.

Let’s see it in action. Imagine a service that needs to find a user’s email address. Instead of directly calling another service, it sends a request message to a specific RabbitMQ queue. A worker service, listening to that queue, picks up the message, performs the lookup, and sends the result back to a reply-to queue specified in the original request. The original service then waits for a message on that reply queue and processes the returned email address.

Here’s a simplified Python example using pika:

Client (Requester):

import pika
import uuid

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# Declare a temporary queue for replies
reply_queue_name = channel.queue_declare(exclusive=True).method.queue

# Generate a unique correlation ID for this request
correlation_id = str(uuid.uuid4())

# Message payload
user_id_to_find = "user:12345"

# Publish the request
channel.basic_publish(
    exchange='',
    routing_key='rpc_queue',  # The queue where the worker listens
    properties=pika.BasicProperties(
        reply_to=reply_queue_name,
        correlation_id=correlation_id,
    ),
    body=user_id_to_find.encode()
)
print(f" [x] Sent '{user_id_to_find}'")

# Callback function to handle replies
def on_response(ch, method, properties, body):
    if properties.correlation_id == correlation_id:
        print(f" [.] Got '{body.decode()}'")
        # Stop consuming once the response is received
        channel.stop_consuming()
    else:
        print(f" [!] Received unexpected correlation_id: {properties.correlation_id} (expected {correlation_id})")

# Start consuming replies
channel.basic_consume(
    queue=reply_queue_name,
    on_message_callback=on_response,
    auto_ack=True
)

# Wait for the response
print(" [*] Waiting for reply...")
channel.start_consuming()

connection.close()

Worker (Responder):

import pika
import time

# Simulate a database or external service
user_db = {
    "user:12345": "alice@example.com",
    "user:67890": "bob@example.com"
}

def get_user_email(user_id):
    print(f" [.] Looking up: {user_id}")
    time.sleep(1) # Simulate work
    return user_db.get(user_id, "User not found")

connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
channel = connection.channel()

# Ensure the RPC queue exists
channel.queue_declare(queue='rpc_queue')

# Callback function to handle incoming requests
def on_request(ch, method, properties, body):
    user_id = body.decode()
    response = get_user_email(user_id)

    # Publish the response back to the reply_to queue
    ch.basic_publish(
        exchange='',
        routing_key=properties.reply_to,
        properties=pika.BasicProperties(correlation_id=properties.correlation_id),
        body=response.encode()
    )
    ch.basic_ack(delivery_tag=method.delivery_tag) # Acknowledge the request

# Set up the worker to process requests
channel.basic_qos(prefetch_count=1) # Process one message at a time
channel.basic_consume(
    queue='rpc_queue',
    on_message_callback=on_request
)

print(" [*] Awaiting RPC requests")
channel.start_consuming()

The fundamental problem this pattern solves is decoupling services. Instead of a direct, synchronous, and brittle point-to-point call, you have an asynchronous intermediary. The client doesn’t need to know where the worker is, only that it can send a message to rpc_queue. The worker doesn’t need to know who is asking, only that it should process messages from rpc_queue and send replies to the reply_to address. This isolation allows services to be updated, scaled, or even replaced independently.

Internally, the magic lies in AMQP’s routing and properties.

exchange='': This signifies the default exchange, which routes messages directly to queues whose names match the routing_key. So, routing_key='rpc_queue' means the message goes to the queue named rpc_queue.
routing_key: In the request, it’s the destination queue for the RPC worker. In the reply, it’s the reply_to queue specified by the client.
properties.reply_to: This is a crucial AMQP property. The client sets this to a unique, often temporary, queue name where it expects the response. The worker must use this property from the incoming request’s properties to send its reply.
properties.correlation_id: Without this, the client wouldn’t know which reply belongs to which request if it sent multiple requests concurrently. The client generates a unique ID for each request, includes it in the correlation_id property, and the worker echoes it back in the reply. The client then matches the incoming reply’s correlation_id to its outstanding requests.
Temporary Queues (exclusive=True): The client declares a temporary queue for replies. exclusive=True means this queue will be deleted automatically when the declaring client’s connection closes, preventing stale reply queues.

The basic_qos(prefetch_count=1) on the worker is vital for RPC. It ensures that a worker, once it has accepted a message, won’t be given another until it has finished processing the first one and sent its reply (or explicitly rejected it). This prevents a fast worker from hogging all the messages while a slow one gets nothing, and more importantly, ensures that the response is sent for the request that is currently being processed.

Many people overlook the importance of the correlation_id when building robust RPC systems. If a client sends multiple requests without unique correlation_ids, and the worker is slow or there are network delays, the client might receive replies out of order and incorrectly associate them with the wrong original requests, leading to subtle and hard-to-debug application errors. Always generate and match correlation_ids.

The next step is to handle worker failures and ensure message durability.