Azure Notification Hub + FCM v1 returns 201 Created, but notifications not received in Flutter Web app

Hi @Rondla Harish Reddy
Azure Notification Hubs (ANH) with FCM v1 requires specific configuration adjustments for successful notification delivery. you need to ensure that your payload is formatted correctly. The payload you provided seems to be aligned with FCM v1 requirements, but you should ensure that the ServiceBusNotification-Format header is set to "fcm" when sending notifications with FCM v1.

payload = {
  "token": fcm_token,
  "notification": {
    "title": "Test Notification",
    "body": "This is a test message"
  }
} 

1.Enable diagnostic logs in Azure Portal → Notification Hub → Diagnostic settings

Stream to Azure Monitor/Event Hubs for analysis

Key metrics: outgoing.allpns.success and outgoing.allpns.invalidpayload

2.Use Azure Portal → Notification Hub → Registrations blade

Filter by FCM v1 platform type
references:
https://learn.microsoft.com/en-us/azure/notification-hubs/firebase-migration-rest#rest-api
https://learn.microsoft.com/en-us/azure/notification-hubs/notification-hubs-gcm-to-fcm#how-do-i-create-fcm-v1-template-registrations-with-sdks-or-rest-apis

Hi @Rondla Harish Reddy
We haven’t heard from you on the last response and was just checking back to see if you have a resolution yet. In case if you have any resolution please do share that same with the community as it can be helpful to others. Otherwise, will respond with more details and we will try to help.

Hi @Rondla Harish Reddy
We haven’t heard from you on the last response and was just checking back to see if you have a resolution yet. In case if you have any resolution please do share that same with the community as it can be helpful to others. Otherwise, will respond with more details and we will try to help.

Hello @Laxman Reddy Revuri ,Thanks for the code response.
I am using below code based on your reply.

import time
import base64
import hmac
import hashlib
from urllib.parse import quote_plus
import requests

AZURE_NOTIFICATION_HUB_NAME = "hub name"
AZURE_NAMESPACE = "azure hub namespace"
AZURE_SAS_KEY_NAME = "Key name"
AZURE_SAS_KEY_VALUE = "key here" 

def generate_sas_token(resource_uri, key_name, key_value, expiry_in_minutes=60):
    encoded_uri = quote_plus(resource_uri.lower()).lower()
    expiry = int(time.time()) + int(expiry_in_minutes) * 60
    string_to_sign = f"{encoded_uri}\n{expiry}"
    decoded_key = base64.b64decode(key_value)
    signature = hmac.new(decoded_key, string_to_sign.encode('utf-8'), hashlib.sha256).digest()
    encoded_signature = quote_plus(base64.b64encode(signature))
    token = f"SharedAccessSignature sr={encoded_uri}&sig={encoded_signature}&se={expiry}&skn={key_name}"
    return token

def register_fcm_v1_device(fcm_token, tags=["myTag"]):
    resource_uri = f"https://{AZURE_NAMESPACE}.servicebus.windows.net/{AZURE_NOTIFICATION_HUB_NAME}/registrations"
    sas_token = generate_sas_token(resource_uri, AZURE_SAS_KEY_NAME, AZURE_SAS_KEY_VALUE)

    headers = {
        'Content-Type': 'application/atom+xml;type=entry;charset=utf-8',
        'Authorization': sas_token,
        'x-ms-version': '2015-01'
    }

    body_template = """{
        "notification": {
            "title": "$(title)",
            "body": "$(message)"
        }
    }"""

    xml_body = f"""<?xml version="1.0" encoding="utf-8"?>
    <entry xmlns="http://www.w3.org/2005/Atom">
        <content type="application/xml">
            <FcmV1TemplateRegistrationDescription xmlns:i="http://www.w3.org/2001/XMLSchema-instance"
                xmlns="http://schemas.microsoft.com/netservices/2010/10/servicebus/connect">
                <Tags>{','.join(tags)}</Tags>
                <FcmV1RegistrationId>{fcm_token}</FcmV1RegistrationId>
                <BodyTemplate><![CDATA[{body_template}]]></BodyTemplate>
            </FcmV1TemplateRegistrationDescription>
        </content>
    </entry>"""

    uri = f"https://{AZURE_NAMESPACE}.servicebus.windows.net/{AZURE_NOTIFICATION_HUB_NAME}/registrations/?api-version=2015-01"
    response = requests.post(uri, data=xml_body, headers=headers)
    return response.status_code, response.text

def send_push_notification(tag, title, message):
    resource_uri = f"https://{AZURE_NAMESPACE}.servicebus.windows.net/{AZURE_NOTIFICATION_HUB_NAME}/messages"
    sas_token = generate_sas_token(resource_uri, AZURE_SAS_KEY_NAME, AZURE_SAS_KEY_VALUE)

    payload = {
        "title": title,
        "message": message
    }

    uri = f"{resource_uri}/?api-version=2020-06"
    headers = {
        'Content-Type': 'application/json',
        'ServiceBusNotification-Format': 'fcm',
        'ServiceBusNotification-Tags': tag,
        'Authorization': sas_token
    }

    response = requests.post(uri, json=payload, headers=headers)
    return response.status_code, response.text

# Sample test run
fcm_token = "fcm token"
status, response = register_fcm_v1_device(fcm_token, tags=["315"])
print("Registration:", status, response)

status, response = send_push_notification("315", "Hello FCMv1", "This is a test push via Azure")
print("Notification Sent:", status, response)

But I am still getting the error: Notification Sent: 401 Access Denied. Invalid SAS token. Could you please suggest what I might be doing wrong? I have followed the provided links and used the correct credentials. Can you please confirm if the code is correct?

Hello @ Rondla Harish Reddy , can you try below code and let me know

import time
import base64
import hmac
import hashlib
from urllib.parse import quote_plus
import requests
# === CONFIGURATION ===
AZURE_NOTIFICATION_HUB_NAME = "your-hub-name"
AZURE_NAMESPACE = "your-namespace"  # without ".servicebus.windows.net"
AZURE_SAS_KEY_NAME = "your-SAS-Key-Name"
AZURE_SAS_KEY_VALUE = "your-SAS-Key-Value"
FCM_TOKEN = "your-client-fcm-token"
# === SAS TOKEN GENERATOR ===
def generate_sas_token(resource_uri, key_name, key_value, expiry_in_minutes=60):
    expiry = int(time.time()) + expiry_in_minutes * 60
    string_to_sign = f"{quote_plus(resource_uri)}\n{expiry}"
    decoded_key = base64.b64decode(key_value)
    signature = hmac.new(decoded_key, string_to_sign.encode('utf-8'), hashlib.sha256).digest()
    encoded_signature = quote_plus(base64.b64encode(signature).decode())
    return f"SharedAccessSignature sr={quote_plus(resource_uri)}&sig={encoded_signature}&se={expiry}&skn={key_name}"
# === REGISTER FCM V1 DEVICE ===
def register_fcm_v1_device(fcm_token, tags=["defaultTag"]):
    resource_uri = f"https://{AZURE_NAMESPACE}.servicebus.windows.net/{AZURE_NOTIFICATION_HUB_NAME}"
    sas_token = generate_sas_token(resource_uri, AZURE_SAS_KEY_NAME, AZURE_SAS_KEY_VALUE)
    headers = {
        'Content-Type': 'application/atom+xml;type=entry;charset=utf-8',
        'Authorization': sas_token,
        'x-ms-version': '2015-01'
    }
    body_template = """{
        "message": {
            "notification": {
                "title": "$(title)",
                "body": "$(message)"
            }
        }
    }"""
    xml_body = f"""<?xml version="1.0" encoding="utf-8"?>
    <entry xmlns="http://www.w3.org/2005/Atom">
      <content type="application/xml">
        <FcmV1RegistrationDescription xmlns:i="http://www.w3.org/2001/XMLSchema-instance" 
                                     xmlns="http://schemas.microsoft.com/netservices/2010/10/servicebus/connect">
          <Tags>{','.join(tags)}</Tags>
          <FcmV1RegistrationId>{fcm_token}</FcmV1RegistrationId>
          <BodyTemplate><![CDATA[{body_template}]]></BodyTemplate>
        </FcmV1RegistrationDescription>
      </content>
    </entry>"""
    uri = f"{resource_uri}/registrations/?api-version=2015-01"
    response = requests.post(uri, headers=headers, data=xml_body)
    return response.status_code, response.text
# === SEND NOTIFICATION ===
def send_push_notification(tag, title, message):
    resource_uri = f"https://{AZURE_NAMESPACE}.servicebus.windows.net/{AZURE_NOTIFICATION_HUB_NAME}"
    sas_token = generate_sas_token(resource_uri, AZURE_SAS_KEY_NAME, AZURE_SAS_KEY_VALUE)
    headers = {
        'Content-Type': 'application/json;charset=utf-8',
        'Authorization': sas_token,
        'ServiceBusNotification-Format': 'template',
        'ServiceBusNotification-Tags': tag
    }
    payload = {
        "title": title,
        "message": message
    }
    uri = f"{resource_uri}/messages/?api-version=2020-06"
    response = requests.post(uri, headers=headers, json=payload)
    return response.status_code, response.text
# === TEST ===
if __name__ == "__main__":
    tag = "flutterUser"
    print("Registering device...")
    status, response = register_fcm_v1_device(FCM_TOKEN, tags=[tag])
    print("Register status:", status)
    print(response)
    print("\nSending notification...")
    status, response = send_push_notification(tag, "Hello!", "This is a test FCM v1 push from Azure.")
    print("Notification status:", status)
    print(response)