Agent Development Cheat Sheet

This is a catalogue of features available in volttron that are frequently useful in agent development.


These functions can be found in the volttron.platform.agent.utils module. logging also needs to be imported to use the logger.


You’ll probably see the following lines near the top of agent files:

_log = logging.getLogger(__name__)

This code sets up the logger for this module so it can provide more useful output. In most cases it will be better to use the logger in lieu of simply printing messages with print.


load_config does just that. Give it the path to your config file and it will parse the json and return a dictionary.


This is the function that is called to start your agent. You’ll likely see it in the main methods at the bottom of agents’ files. Whatever is passed to it (a class name or a function that returns an instance of your agent) should accept a file path that can be parsed with load_config.

Core Agent Functionality

These tools module. Try importing

Agent Lifecycle Events

Each agent has four events that are triggered at different stages of its life. These are onsetup, onstart, onstop, and onfinish. Registering callbacks to these events are commonplace in agent development, with onstart being the most frequently used.

The easiest way to register a callback is with a function decorator:

def function(self, sender, **kwargs):

Periodic and Scheduled Function Calls

Functions and agent methods can be registered to be called periodically or scheduled to run at a particular time using the Core.schedule decorator or by calling an agent’s core.schedule() method. The latter is especially useful if, for example, a decision needs to be made in an agent’s onstart method as to whether a call should be scheduled.

from volttron.platform.scheduling import cron, periodic

def function(self):

def periodic_function(self):

@Core.schedule(cron('0 1 * * *'))
def cron_function(self):


# inside some agent method
self.core.schedule(t, function)
self.core.schedule(periodic(t), periodic_function)
self.core.schedule(cron('0 1 * * *'), cron_function)


These features are available to all Agent subclasses. No extra imports are required.

Remote Procedure Calls

Remote Procedure Calls, or RPCs are a powerful way to interact with other agents. To make a function available to call by a remote agent just add the export decorator:

def function(self, ...):

function can now be called by a remote agent agent with

# vip identity is the identity (a string) of the agent
# where function() is defined, 'function').get(timeout=t)


Agents can publish and subscribe to topics. Like RPC, pubsub functions can be invoked via decorators or inline through vip. The following function is called whenever the agent sees a message starting with topic_prefix.

@PubSub.subscribe('pubsub', topic_prefix)
def function(self, peer, sender, bus,  topic, headers, message):

An agent can publish to a topic topic with the method.

An agent can remove a subscriptions with Giving None as values for the prefix and callback argument will unsubscribe from everything on that bus. This is handy for subscriptions that must be updated base on a configuration setting.

Configuration Store

Support for the configuration store is done by subscribing to configuration changes with, actions=["NEW", "UPDATE"], pattern="config")

See Agent Configuration Store


The heartbeat subsystem provides access to a periodic publish so that others can observe the agent’s status. Other agents can subscibe to the heartbeat topic to see who is actively publishing to it.

It it turned off by default.


The health subsystem adds extra status information to the an agent’s heartbeat. Setting the status will start the heartbeat if it wasn’t already.

Agent Skeleton

import logging

from import Agent, Core, PubSub, RPC
from volttron.platform.agent import utils

_log = logging.getLogger(__name__)

class MyAgent(Agent):
    def __init__(self, config_path, **kwargs):
        self.config = utils.load_config(config_path)

    def onsetup(self, sender, **kwargs):

    def onstart(self, sender, **kwargs):

    def onstop(self, sender, **kwargs):

    def onfinish(self, sender, **kwargs):

    @PubSub.subscribe('pubsub', 'some/topic')
    def on_match(self, peer, sender, bus,  topic, headers, message):

    def my_method(self):

def main():

if __name__ == '__main__':
    except KeyboardInterrupt: