Python: Azure Service Bus Without SDK (REST API Guide)

Want to send and receive notifications on Azure Service Bus using Python, but don’t want to rely on the official SDK? This guide shows you how to authenticate and interact with Azure Service Bus queues directly using HTTP requests and SAS tokens. Let’s dive in!

Azure Service Bus (ASB) uses Azure Active Directory (AAD) or Shared Access Signature (SAS) tokens for authentication. In this example, we assume you have owner access and can generate a Send/Listen SAS key from the Azure Portal. Here’s how to create a valid SAS token:

def get_auth_token(sb_name, eh_name, sas_name, sas_value):
    # generate SAS token
    uri = "https://{}.servicebus.windows.net/{}".format(sb_name, eh_name)
    sas = sas_value.encode('utf-8')
    expiry = str(int(time.time() + 10000))
    string_to_sign = (urllib.parse.quote_plus(uri) + 'n' + expiry).encode('utf-8')
    signed_hmac_sha256 = hmac.HMAC(sas, string_to_sign, hashlib.sha256)
    signature = urllib.parse.quote(base64.b64encode(signed_hmac_sha256.digest()))
    return  {"uri": uri,
             "token":'SharedAccessSignature sr={}&sig={}&se={}&skn={}'
                     .format(uri, signature, expiry, sas_name)
            }

Once you have generated the token, sending and receiving messages is straightforward. Below is a complete code snippet that generates a SAS token and sends your machine’s IP address via Azure Service Bus.

import time
import urllib
import hmac
import hashlib
import base64
import requests
import socket

h_name = socket.gethostname()
IP_address = socket.gethostbyname(h_name)

def get_auth_token(sb_name, eh_name, sas_name, sas_value):
    # generate SAS token
    uri = "https://{}.servicebus.windows.net/{}".format(sb_name, eh_name)
    sas = sas_value.encode('utf-8')
    expiry = str(int(time.time() + 10000))
    string_to_sign = (urllib.parse.quote_plus(uri) + 'n' + expiry).encode('utf-8')
    signed_hmac_sha256 = hmac.HMAC(sas, string_to_sign, hashlib.sha256)
    signature = urllib.parse.quote(base64.b64encode(signed_hmac_sha256.digest()))
    return  {"uri": uri,
             "token":'SharedAccessSignature sr={}&sig={}&se={}&skn={}'
                     .format(uri, signature, expiry, sas_name)
            }

def send_message(token, message):
    # POST http{s}://{serviceNamespace}.servicebus.windows.net/{queuePath}/messages
    r = requests.post(token['uri'] + "/messages",
        headers={
            "Authorization": token['token'],
            "Content-Type": "application/json"
        },
        json=message)

def recieve_message(token): 
    # DELETE http{s}://{serviceNamespace}.servicebus.windows.net/{queuePath}/messages/head
    # 204 if no message
    r = requests.delete(token['uri'] + "/messages/head",
        headers={
            "Authorization": token['token'],
        })
    return r.text

sb_name = "<service bus name>"
q_name = "<service bus queue name>"

skn = "<key name for that access key>"
key = "<access key created in portal>"

token = get_auth_token(sb_name, q_name, skn, key)
print(token['token'])

uri = "https://" + sb_name + ".servicebus.windows.net/" + q_name + "/messages"

send_message(token, {'ip': IP_address})
recieve_message(token)