Webhooks API Quickstart

The Webhooks API manages transaction notifications using webhooks.

Alchemy Webhooks are used to subscribe to event notifications that occur on your application. You create a webhook to receive notifications on different types of on-chain activity. Here are a few quick-links and an introduction video:

Don’t have an API key?

Sign up or upgrade your plan for access.

Webhook types

Below are the Alchemy webhook types and their supported networks.

Webhook TypeDescriptionNetwork
Custom WebhooksCustom Webhooks allows you to track any smart contract or marketplace activity, monitor any contract creation, or any other on-chain interaction. This gives you infinite data access with precise filter controls to get the blockchain data you need.Ethereum, zkSync, Optimism, Polygon PoS, Shape, Arbitrum, Arbitrum Nova, ZetaChain, Fantom Opera, Blast, Linea, Base, Gnosis, Metis
Mined TransactionThe Mined Transaction webhook notifies your app when a transaction sent through your app (using your API key) gets mined. This is useful for you to further notify the users of your app about the status of the transaction.Ethereum, zkSync, Optimism, Polygon PoS, Shape, Arbitrum, Arbitrum Nova, ZetaChain, Fantom Opera, Blast, Linea, Base, Gnosis, Metis
Dropped TransactionsThe Dropped Transaction webhook notifies your app when a transaction sent through your app (using your API key) gets dropped. This is useful for you to further notify the users of your app about the status of the transaction.Ethereum, zkSync, Optimism, Polygon PoS, Shape, Arbitrum, Arbitrum Nova, ZetaChain, Fantom Opera, Blast, Linea, Base, Gnosis, Metis
Address ActivityAlchemy's Address Activity webhook tracks all ETH, ERC20, ERC721 and ERC1155 transfers. This provides your app with real-time state changes when an address sends/receives tokens or ETH. You can specify the addresses for which you want to track this activity. A maximum of 50,000 addresses can be added to a single webhook.Ethereum, zkSync, Optimism, Polygon PoS, Shape, Arbitrum, Arbitrum Nova, ZetaChain, Fantom Opera, Blast, Linea, Base, Gnosis, Metis
NFT ActivityThe NFT Activity webhook allows you to track ERC721 and ERC1155 token contracts. This provides your app with real-time state changes when an NFT is transferred between addresses.Ethereum, zkSync, Optimism, Polygon PoS, Shape, Arbitrum, Arbitrum Nova, ZetaChain, Fantom Opera, Blast, Linea, Base, Gnosis, Metis
NFT Meta UpdatesThe NFT Metadata Updates webhook allows you to track metadata updates for ERC721 and ERC1155 token contracts for Ethereum and Polygon NFTs. This notifies your app when the metadata for an NFT is updated.Ethereum Mainnet & Goerli; Polygon Mainnet & Mumbai

V2 Webhook

Field definitions

FieldDescriptionValue
webhookIdUnique ID of the webhook destination.wh_octjglnywaupz6th
idID of the event.whevt_ogrc5v64myey69ux
createdAtThe timestamp when webhook was created.2021-12-07T03:52:45.899Z
typeWebhook event type.TYPE_STRING
eventObject-mined transaction.OBJECT

Example

{
 "webhookId": "wh_octjglnywaupz6th",
 "id": "whevt_ogrc5v64myey69ux",
 "createdAt": "2021-12-07T03:52:45.899Z",
 "type": TYPE_STRING,
 "event": OBJECT
}

V1 Webhook Event Object

Field definitions

FieldDescriptionValue
appAlchemy app name sending the transaction webhook.Demo
networkNetwork for the webhook event.MAINNET
webhookTypeThe type of webhook.MINED_TRANSACTION
timestampTimestamp when the webhook was created.2020-07-29T00:29:18.414Z
event nameWebhook event type.OBJECT

For Webhooks full dependencies and more code examples, [go to the GitHub repo] (https://github.com/alchemyplatform/webhook-examples).

Example Response

{
  "app": "Demo", 
  "network": "MAINNET",
  "webhookType": "MINED_TRANSACTION",
  "timestamp": "2020-07-29T00:29:18.414Z",
  "event name": OBJECT
}

How to set-up webhooks

Setting up a webhook is as simple as adding a new URL to your application.

❗️

NOTE:

If you need to add over 10 addresses to the address activity webhook, we recommend adding them through an API call.

Set-up webhooks in your dashboard

  1. Navigate to the Webhooks tab in your Alchemy Dashboard.
  2. Determine which type of webhook you want to activate.
  3. Click the Create Webhook button.
  4. Select the app to add the webhook notifications.
  5. Add in your unique webhook URL. This is the link to receive requests (Please note that localhost domains are not allowed as webhook URLs). The webhook payload might not always be compatible for 3rd party integrations.
  6. Test your webhook by clicking the Test Webhook button.
  7. Click Create Webhook and your webhook appears in the list.
  8. Check your endpoint to see responses.

Set-up webhooks programmatically

Use the create-webhook endpoint: https://docs.alchemy.com/reference/create-webhook.

Webhook IP addresses

As an added security measure, you can ensure your webhook notification originates from Alchemy by using one of the following IP addresses:

  • 54.236.136.17
  • 34.237.24.169

Create webhook listeners

Webhook listeners receive requests and process event data.

The listener responds to the Alchemy server with a 200 status code once you've successfully received the webhook event. Your webhook listener can be a simple server or Slack integration to receive the webhook listener data.

After setting up the webhooks in your Alchemy dashboard (or programmatically) use the starter code in JavaScript, Python, Go, and Rust below. Here's the GitHub repository for the entire code.

import express from "express";
import { getRequiredEnvVar, setDefaultEnvVar } from "./envHelpers";
import {
  addAlchemyContextToRequest,
  validateAlchemySignature,
  AlchemyWebhookEvent,
} from "./webhooksUtil";

async function main(): Promise<void> {
  const app = express();

  setDefaultEnvVar("PORT", "8080");
  setDefaultEnvVar("HOST", "127.0.0.1");
  setDefaultEnvVar("SIGNING_KEY", "whsec_test");

  const port = +getRequiredEnvVar("PORT");
  const host = getRequiredEnvVar("HOST");
  const signingKey = getRequiredEnvVar("SIGNING_KEY");

  // Middleware needed to validate the alchemy signature
  app.use(
    express.json({
      verify: addAlchemyContextToRequest,
    })
  );
  app.use(validateAlchemySignature(signingKey));

  // Register handler for Alchemy Notify webhook events
  // TODO: update to your own webhook path
  app.post("/webhook-path", (req, res) => {
    const webhookEvent = req.body as AlchemyWebhookEvent;
    // Do stuff with with webhook event here!
    console.log(`Processing webhook event id: ${webhookEvent.id}`);
    // Be sure to respond with 200 when you successfully process the event
    res.send("Alchemy Webhooks are the best!");
  });

  // Listen to Alchemy Notify webhook events
  app.listen(port, host, () => {
    console.log(`Example Alchemy Webhooks app listening at ${host}:${port}`);
  });
}

main();
import hmac
import hashlib
from django.core.exceptions import PermissionDenied
from webhook_server.settings import SIGNING_KEY
import json
from types import SimpleNamespace


def is_valid_signature_for_string_body(
    body: str, signature: str, signing_key: str
) -> bool:
    digest = hmac.new(
        bytes(signing_key, "utf-8"),
        msg=bytes(body, "utf-8"),
        digestmod=hashlib.sha256,
    ).hexdigest()

    return signature == digest


class AlchemyWebhookEvent:
    def __init__(self, webhookId, id, createdAt, type, event):
        self.webhook_id = webhookId
        self.id = id
        self.created_at = createdAt
        self.type = type
        self.event = event


class AlchemyRequestHandlerMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response

    def __call__(self, request):
        str_body = str(request.body, request.encoding or "utf-8")
        signature = request.headers["x-alchemy-signature"]
        if not is_valid_signature_for_string_body(str_body, signature, SIGNING_KEY):
            raise PermissionDenied("Signature validation failed, unauthorized!")

        webhook_event = json.loads(str_body)
        request.alchemy_webhook_event = AlchemyWebhookEvent(**webhook_event)
        response = self.get_response(request)
        return response
package notify

import (
	"crypto/hmac"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"io/ioutil"
	"log"
	"net/http"
	"time"
)

type AlchemyWebhookEvent struct {
	WebhookId string
	Id        string
	CreatedAt time.Time
	Type      string
	Event     map[string]interface{}
}

func jsonToAlchemyWebhookEvent(body []byte) (*AlchemyWebhookEvent, error) {
	event := new(AlchemyWebhookEvent)
	if err := json.Unmarshal(body, &event); err != nil {
		return nil, err
	}
	return event, nil
}

// Middleware helpers for handling an Alchemy Notify webhook request
type AlchemyRequestHandler func(http.ResponseWriter, *http.Request, *AlchemyWebhookEvent)

type AlchemyRequestHandlerMiddleware struct {
	handler    AlchemyRequestHandler
	signingKey string
}

func NewAlchemyRequestHandlerMiddleware(handler AlchemyRequestHandler, signingKey string) *AlchemyRequestHandlerMiddleware {
	return &AlchemyRequestHandlerMiddleware{handler, signingKey}
}

func isValidSignatureForStringBody(
	body []byte,
	signature string,
	signingKey []byte,
) bool {
	h := hmac.New(sha256.New, signingKey)
	h.Write([]byte(body))
	digest := hex.EncodeToString(h.Sum(nil))
	return digest == signature
}

func (arh *AlchemyRequestHandlerMiddleware) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	signature := r.Header.Get("x-alchemy-signature")
	body, err := ioutil.ReadAll(r.Body)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		log.Panic(err)
		return
	}
	r.Body.Close()

	isValidSignature := isValidSignatureForStringBody(body, signature, []byte(arh.signingKey))
	if !isValidSignature {
		errMsg := "Signature validation failed, unauthorized!"
		http.Error(w, errMsg, http.StatusForbidden)
		log.Panic(errMsg)
		return
	}

	event, err := jsonToAlchemyWebhookEvent(body)
	if err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		log.Panic(err)
		return
	}
	arh.handler(w, r, event)
}
use chrono::{DateTime, FixedOffset};
use hex;
use hmac::{Hmac, Mac};
use sha2::Sha256;
use std::{
    future::{ready, Ready},
    rc::Rc,
};

use serde::{de, Deserialize, Deserializer};

use actix_web::{
    dev::{self, Service, ServiceRequest, ServiceResponse, Transform},
    error::ErrorBadRequest,
    error::ErrorUnauthorized,
    web::BytesMut,
    Error, HttpMessage,
};
use futures_util::{future::LocalBoxFuture, stream::StreamExt};

#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]

pub struct AlchemyWebhookEvent {
    pub webhook_id: String,
    pub id: String,
    #[serde(deserialize_with = "deserialize_date_time")]
    pub created_at: DateTime<FixedOffset>,
    #[serde(rename = "type")]
    pub webhook_type: String,
    pub event: serde_json::Value,
}

fn deserialize_date_time<'a, D>(deserializer: D) -> Result<DateTime<FixedOffset>, D::Error>
where
    D: Deserializer<'a>,
{
    let date_time_string: String = Deserialize::deserialize(deserializer)?;
    let date_time = DateTime::<FixedOffset>::parse_from_rfc3339(&date_time_string)
        .map_err(|e| de::Error::custom(e.to_string()))?;
    Ok(date_time)
}

fn is_valid_signature_for_string_body(
    body: &[u8],
    signature: &str,
    signing_key: &str,
) -> Result<bool, Box<dyn std::error::Error>> {
    let signing_key_bytes: Vec<u8> = signing_key.bytes().collect();
    let mut mac = Hmac::<Sha256>::new_from_slice(&signing_key_bytes)?;
    mac.update(&body);
    let hex_decode_signature = hex::decode(signature)?;
    let verification = mac.verify_slice(&hex_decode_signature).is_ok();
    Ok(verification)
}

pub struct AlchemyRequestHandlerMiddleware<S> {
    signing_key: String,
    service: Rc<S>,
}

impl<S, B> Service<ServiceRequest> for AlchemyRequestHandlerMiddleware<S>
where
    S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error> + 'static,
    S::Future: 'static,
    B: 'static,
{
    type Response = ServiceResponse<B>;
    type Error = Error;
    type Future = LocalBoxFuture<'static, Result<Self::Response, Self::Error>>;

    dev::forward_ready!(service);

    fn call(&self, mut req: ServiceRequest) -> Self::Future {
        let svc = self.service.clone();
        let signing_key = self.signing_key.clone();

        Box::pin(async move {
            let mut body = BytesMut::new();
            let mut stream = req.take_payload();
            while let Some(chunk) = stream.next().await {
                body.extend_from_slice(&chunk?);
            }

            let signature = req
                .headers()
                .get("x-alchemy-signature")
                .ok_or(ErrorBadRequest(
                    "Signature validation failed, missing x-alchemy-signature header!",
                ))?
                .to_str()
                .map_err(|_| {
                    ErrorBadRequest(
                        "Signature validation failed, x-alchemy-signature header is not a string!",
                    )
                })?;

            let is_valid_signature =
                is_valid_signature_for_string_body(&body, signature, &signing_key)?;

            if !is_valid_signature {
                return Err(ErrorUnauthorized(
                    "Signature validation failed, signature and body do not match!",
                ));
            }

            let webhook: AlchemyWebhookEvent = serde_json::from_slice(&body).map_err(|_| {
                ErrorBadRequest("Bad format, body could not be mapped to AlchemyWebhookEvent")
            })?;

            req.extensions_mut()
                .insert::<Rc<AlchemyWebhookEvent>>(Rc::new(webhook));

            let res = svc.call(req).await?;

            Ok(res)
        })
    }
}

pub struct AlchemyRequestHandlerMiddlewareFactory {
    signing_key: String,
}

impl AlchemyRequestHandlerMiddlewareFactory {
    pub fn new(signing_key: String) -> Self {
        AlchemyRequestHandlerMiddlewareFactory { signing_key }
    }
}

impl<S, B> Transform<S, ServiceRequest> for AlchemyRequestHandlerMiddlewareFactory
where
    B: 'static,
    S: Service<ServiceRequest, Response = ServiceResponse<B>, Error = Error> + 'static,
{
    type Response = ServiceResponse<B>;
    type Error = Error;
    type Transform = AlchemyRequestHandlerMiddleware<S>;
    type InitError = ();
    type Future = Ready<Result<Self::Transform, Self::InitError>>;

    fn new_transform(&self, service: S) -> Self::Future {
        ready(Ok(AlchemyRequestHandlerMiddleware {
            signing_key: self.signing_key.clone(),
            service: Rc::new(service),
        }))
    }
}

Test webhooks with Ngrok

  1. Sign-up for a free Ngrok account.
  2. Install Ngrok using the Ngrok guide. On macOS run brew install ngrok.
  3. Connect your Ngrok account by running ngrok authtoken YOUR_AUTH_TOKEN.
  4. Start your local forwarding tunnel: ngrok http 80.

Once you have a URL to test your webhook (in this case https://461a-199-116-73-171.ngrok.io pictured above), follow the steps below:

  1. Navigate to your Webhooks dashboard.
  2. Click Create Webhook on the webhook you want to test.
  3. Paste your unique URL and hit the Test Webhook button.
  4. You'll see the webhooks here: http://localhost:4040/inspect/http.

Webhook signature & security

To make your webhooks secure, you can verify that they originated from Alchemy by generating a HMAC SHA-256 hash code using your unique webhook signing key.

Find your signing key

Navigate to the Webhooks page in your dashboard. Next, click on the three dots of the webhook you want to get the signature for and copy the Signing Key.

Validate the signature received

Every outbound request contains a hashed authentication signature in the header. It's computed by concatenating your signing key and request body. Then generates a hash using the HMAC SHA256 hash algorithm.

To verify the signature came from Alchemy, you generate the HMAC SHA256 hash and compare it with the signature received.

Example request header

POST /yourWebhookServer/push HTTP/1.1
Content-Type: application/json;
X-Alchemy-Signature: your-hashed-signature

Example signature validation

import * as crypto from "crypto";

function isValidSignatureForStringBody(
    body: string, // must be raw string body, not json transformed version of the body
    signature: string, // your "X-Alchemy-Signature" from header
    signingKey: string, // taken from dashboard for specific webhook
  ): boolean {
    const hmac = crypto.createHmac("sha256", signingKey); // Create a HMAC SHA256 hash using the signing key
    hmac.update(body, "utf8"); // Update the token hash with the request body using utf8
    const digest = hmac.digest("hex");
    return signature === digest;
}
import hmac
import hashlib

def isValidSignatureForStringBody(body: str, signature: str, signing_key: str)->bool:
    digest = hmac.new(
        bytes(signing_key, "utf-8"),
        msg=bytes(body, "utf-8"),
        digestmod=hashlib.sha256,
    ).hexdigest()
    
    return signature == digest
func isValidSignatureForStringBody(
  body []byte, // must be raw string body, not json transformed version of the body
  signature string, // your "X-Alchemy-Signature" from header
  signingKey []byte, // taken from dashboard for specific webhook
) bool {
  h := hmac.New(sha256.New, signingKey)
  h.Write([]byte(body))
  digest := hex.EncodeToString(h.Sum(nil))
  return digest == signature
}
fn is_valid_signature_for_string_body(
    body: &[u8], // must be raw string body, not json transformed version of the body
    signature: &str, // your "X-Alchemy-Signature" from header
    signing_key: &str, // taken from dashboard for specific webhook
) -> Result<bool, Box<dyn std::error::Error>> {
    let signing_key_bytes: Vec<u8> = signing_key.bytes().collect();
    let mut mac = Hmac::<Sha256>::new_from_slice(&signing_key_bytes)?;
    mac.update(&body);
    let hex_decode_signature = hex::decode(signature)?;
    let verification = mac.verify_slice(&hex_decode_signature).is_ok();
    Ok(verification)
}

Webhook retry logic

Alchemy Webhooks have built-in retry-logic for webhooks. Requests are retried for non-200 response codes in failures to reach your server.

Requests are retried up to 6 times before failing. Below are the request retry intervals.

  • 15 seconds
  • 1 minute
  • 10 minutes
  • 1 hour
  • 1 day