Scheduling tasks

Run delayed and repeating tasks with the scheduler.

The scheduler runs work later, or on a repeating interval. Time is measured in ticks - 20 ticks is roughly one second. Reach it through self.server.scheduler.

Run something later

Pass a delay and leave period at its default to run a task once, after the wait:

src/endstone_my_plugin/my_plugin.py
from endstone.plugin import Plugin

class MyPlugin(Plugin):
    api_version = "0.11"

    def on_enable(self) -> None:
        self.server.scheduler.run_task(self, self.announce, delay=100)

    def announce(self) -> None:
        self.server.broadcast_message("The event starts now!")

delay=100 waits about five seconds before the single run.

Repeat on an interval

Add a period to run the task again and again with that gap between runs:

    def on_enable(self) -> None:
        self.server.scheduler.run_task(self, self.say_hi, delay=0, period=20)

    def say_hi(self) -> None:
        for player in self.server.online_players:
            player.send_popup("Hi!")

With delay=0, period=20, say_hi runs immediately and then once a second.

Stop a task

run_task returns a Task. Keep it, and cancel it when you're done:

    def on_enable(self) -> None:
        self.task = self.server.scheduler.run_task(self, self.say_hi, delay=0, period=20)

    def on_disable(self) -> None:
        self.task.cancel()

You can also cancel by id with self.server.scheduler.cancel_task(self.task.task_id), or drop everything your plugin scheduled at once with cancel_tasks(self).

A self-cancelling countdown

For a task that should stop on its own, hold onto its Task and cancel it from inside the callback once the work is finished:

    def on_enable(self) -> None:
        self.countdown = 5
        self.task = self.server.scheduler.run_task(self, self.tick, delay=0, period=20)

    def tick(self) -> None:
        if self.countdown > 0:
            self.server.broadcast_message(f"Starting in {self.countdown}...")
            self.countdown -= 1
        else:
            self.server.broadcast_message("Go!")
            self.task.cancel()

Asynchronous tasks

The scheduler only runs tasks synchronously, on the server's main thread - there is no run_task_async, and that's deliberate. Work that shouldn't block the server (network requests, database queries) belongs on a background thread instead, which is what endstone.asyncio is for. Run the work there, then use run_task to bring the result back to the main thread before touching the API.

On this page