Using Locust as a library

It is possible to start a load test from your own Python code, instead of running Locust using the locust command.

Start by creating an Environment instance:

from locust.env import Environment

env = Environment(user_classes=[MyTestUser])

The Environment instance’s create_local_runner, create_master_runner can then be used to start a Runner instance, which can be used to start a load test:

env.runner.start(5000, spawn_rate=20)

It is also possible to bypass the dispatch and distribution logic, and manually control the spawned users:

new_users = env.runner.spawn_users({MyUserClass.__name__: 2})
new_users[1].my_custom_token = "custom-token-2"
new_users[0].my_custom_token = "custom-token-1"

The above example only works on standalone/local runner mode and is an experimental feature. A more common/better approach would be to use init or test_start Event hooks to get/create a list of tokens and use on_start and on_stop methods to read from that list and set them on your individual User instances.


While it is possible to create locust workers this way (using create_worker_runner), that almost never makes sense. Every worker needs to be in a separate Python process and interacting directly with the worker runner might break things. Just launch workers using the regular locust --worker ... command.

We could also use the Environment instance’s create_web_ui method to start a Web UI that can be used to view the stats, and to control the runner (e.g. start and stop load tests):


Full example

import gevent
from locust import HttpUser, task
from locust.env import Environment
from locust.stats import stats_printer, stats_history
from locust.log import setup_logging

setup_logging("INFO", None)

class MyUser(HttpUser):
    host = ""

    def t(self):

# setup Environment and Runner
env = Environment(user_classes=[MyUser])
runner = env.create_local_runner()

# start a WebUI instance
web_ui = env.create_web_ui("", 8089)

# execute init event handlers (only really needed if you have registered any), runner=runner, web_ui=web_ui)

# start a greenlet that periodically outputs the current stats

# start a greenlet that save current stats to history
gevent.spawn(stats_history, env.runner)

# start the test
runner.start(1, spawn_rate=10)

# in 60 seconds stop the runner
gevent.spawn_later(60, lambda: runner.quit())

# wait for the greenlets

# stop the web server for good measures