EventSub Webhook#

Note

EventSub Webhook is targeted at programs which have to subscribe to topics for multiple broadcasters.

Should you only need to target a single broadcaster or are building a client side projekt, look at EventSub Websocket

EventSub lets you listen for events that happen on Twitch.

The EventSub client runs in its own thread, calling the given callback function whenever an event happens.

Requirements#

Note

Please note that Your Endpoint URL has to be HTTPS, has to run on Port 443 and requires a valid, non self signed certificate This most likely means, that you need a reverse proxy like nginx. You can also hand in a valid ssl context to be used in the constructor.

In the case that you don’t hand in a valid ssl context to the constructor, you can specify any port you want in the constructor and handle the bridge between this program and your public URL on port 443 via reverse proxy.

You can check on whether or not your webhook is publicly reachable by navigating to the URL set in callback_url. You should get a 200 response with the text pyTwitchAPI eventsub.

Listening to topics#

After you started your EventSub client, you can use the listen_ prefixed functions to listen to the topics you are interested in.

Look at Available Topics and Callback Payloads to find the topics you are interested in.

The function you hand in as callback will be called whenever that event happens with the event data as a parameter, the type of that parameter is also listed in the link above.

Code Example#from twitchAPI.twitch import Twitch
from twitchAPI.helper import first
from twitchAPI.eventsub.webhook import EventSubWebhook
from twitchAPI.object.eventsub import ChannelFollowEvent
from twitchAPI.oauth import UserAuthenticator
from twitchAPI.type import AuthScope
import asyncio

TARGET_USERNAME = 'target_username_here'
EVENTSUB_URL = 'https://url.to.your.webhook.com'
APP_ID = 'your_app_id'
APP_SECRET = 'your_app_secret'
TARGET_SCOPES = [AuthScope.MODERATOR_READ_FOLLOWERS]


async def on_follow(data: ChannelFollowEvent):
    # our event happend, lets do things with the data we got!
    print(f'{data.event.user_name} now follows {data.event.broadcaster_user_name}!')


async def eventsub_webhook_example():
    # create the api instance and get the ID of the target user
    twitch = await Twitch(APP_ID, APP_SECRET)
    user = await first(twitch.get_users(logins=TARGET_USERNAME))

    # the user has to authenticate once using the bot with our intended scope.
    # since we do not need the resulting token after this authentication, we just discard the result we get from authenticate()
    # Please read up the UserAuthenticator documentation to get a full view of how this process works
    auth = UserAuthenticator(twitch, TARGET_SCOPES)
    await auth.authenticate()

    # basic setup, will run on port 8080 and a reverse proxy takes care of the https and certificate
    eventsub = EventSubWebhook(EVENTSUB_URL, 8080, twitch)

    # unsubscribe from all old events that might still be there
    # this will ensure we have a clean slate
    await eventsub.unsubscribe_all()
    # start the eventsub client
    eventsub.start()
    # subscribing to the desired eventsub hook for our user
    # the given function (in this example on_follow) will be called every time this event is triggered
    # the broadcaster is a moderator in their own channel by default so specifying both as the same works in this example
    await eventsub.listen_channel_follow_v2(user.id, user.id, on_follow)

    # eventsub will run in its own process
    # so lets just wait for user input before shutting it all down again
    try:
        input('press Enter to shut down...')
    finally:
        # stopping both eventsub as well as gracefully closing the connection to the API
        await eventsub.stop()
        await twitch.close()
    print('done')


# lets run our example
asyncio.run(eventsub_webhook_example())

class twitchAPI.eventsub.webhook.EventSubWebhook#

Bases: EventSubBase

__init__(callback_url, port, twitch, ssl_context=None, host_binding='0.0.0.0', subscription_url=None, callback_loop=None, revocation_handler=None, message_deduplication_history_length=50)#

Parameters:

callback_url¶ (str) – The full URL of the webhook.
port¶ (int) – the port on which this webhook should run
twitch¶ (Twitch) – a app authenticated instance of Twitch
ssl_context¶ (Optional[SSLContext]) – optional ssl context to be used
Default: None
host_binding¶ (str) – the host to bind the internal server to
Default: 0.0.0.0
subscription_url¶ (Optional[str]) – Alternative subscription URL, usefull for development with the twitch-cli
callback_loop¶ (Optional[AbstractEventLoop]) –
The asyncio eventloop to be used for callbacks.

Set this if you or a library you use cares about which asyncio event loop is running the callbacks. Defaults to the one used by EventSub Webhook.
revocation_handler¶ (Optional[Callable[[dict], Awaitable[None]]]) – Optional handler for when subscriptions get revoked.
Default: None
message_deduplication_history_length¶ (int) – The amount of messages being considered for the duplicate message deduplication.
Default: 50

callback_url: str#: The full URL of the webhook.

secret: str#: A random secret string. Set this for added security.
Default: A random 20 character long string

wait_for_subscription_confirm: bool#: Set this to false if you don’t want to wait for a subscription confirm.
Default: True

wait_for_subscription_confirm_timeout: int#: Max time in seconds to wait for a subscription confirmation. Only used if wait_for_subscription_confirm is set to True.
Default: 30

subscription_url: Optional[str]#: Alternative subscription URL, usefull for development with the twitch-cli

revokation_handler: Optional[Callable[[dict], Awaitable[None]]]#: Optional handler for when subscriptions get revoked.

unsubscribe_on_stop: bool#: Unsubscribe all currently active Webhooks on calling stop()
Default: True

start()#

Starts the EventSub client

Return type:: None
Raises:: RuntimeError – if EventSub is already running

async stop()#

Stops the EventSub client

This also unsubscribes from all known subscriptions if unsubscribe_on_stop is True

Return type:: None
Raises:: RuntimeError – if EventSub is not running

async listen_channel_ad_break_begin(broadcaster_user_id, callback)#

A midroll commercial break has started running.

Requires the CHANNEL_READ_ADS scope.

Parameters:

broadcaster_user_id¶ (str) – The ID of the broadcaster that you want to get Channel Ad Break begin notifications for.
callback¶ (Callable[[ChannelAdBreakBeginEvent], Awaitable[None]]) – function for callback

Return type: