Creating webhooks

Copy for LLM

Webhooks allow you to configure the Apify platform to perform an action when a certain event occurs. For example, you can use them to start another Actor when the current run finishes or fails.

Creating an ad-hoc webhook dynamically

Besides creating webhooks manually in Apify Console, or through the Apify API, you can also create ad-hoc webhooks dynamically from the code of your Actor using the Actor.add_webhook method:

Run on

import asyncio

from apify import Actor, Webhook


async def main() -> None:
    async with Actor:
        # Create a webhook that will be triggered when the Actor run fails.
        webhook = Webhook(
            event_types=['ACTOR.RUN.FAILED'],
            request_url='https://example.com/run-failed',
        )

        # Add the webhook to the Actor.
        await Actor.add_webhook(webhook)

        # Raise an error to simulate a failed run.
        raise RuntimeError('I am an error and I know it!')


if __name__ == '__main__':
    asyncio.run(main())

Note that webhooks are only supported when running on the Apify platform. When running the Actor locally, the method will print a warning and have no effect.

Preventing duplicate webhooks

To ensure that duplicate ad-hoc webhooks won't get created in a case of Actor restart, you can use the idempotency_key parameter. The idempotency key must be unique across all the webhooks of a user so that only one webhook gets created for a given value. You can use, for example, the Actor run ID as the idempotency key:

Run on

import asyncio

from apify import Actor, Webhook


async def main() -> None:
    async with Actor:
        # Create a webhook with an idempotency key to prevent duplicates on retries.
        webhook = Webhook(
            event_types=['ACTOR.RUN.FAILED'],
            request_url='https://example.com/run-failed',
            idempotency_key=Actor.configuration.actor_run_id,
        )

        # Add the webhook to the Actor.
        await Actor.add_webhook(webhook)

        # Raise an error to simulate a failed run.
        raise RuntimeError('I am an error and I know it!')


if __name__ == '__main__':
    asyncio.run(main())