ROS 2 Lifecycle Nodes Guide 2026State Machine · Callbacks · Manager · CLI

#!/usr/bin/env python3
import rclpy
from rclpy.lifecycle import LifecycleNode, TransitionCallbackReturn
from rclpy.lifecycle import State
from lifecycle_msgs.msg import Transition


class MyLifecycleNode(LifecycleNode):
    """A managed lifecycle node with 4 state-transition callbacks."""

    def __init__(self):
        super().__init__("my_lifecycle_node")

    # Called when: Unconfigured → Inactive
    def on_configure(self, state: State) -> TransitionCallbackReturn:
        self.get_logger().info("Configuring... loading params, creating pub/sub")
        self.declare_parameter("sensor_topic", "/scan")

        # Create publisher (inactive by default — doesn't publish yet)
        self._pub = self.create_lifecycle_publisher(
            from_type=None,  # replaced with real type below
        )
        return TransitionCallbackReturn.SUCCESS

    # Called when: Inactive → Active
    def on_activate(self, state: State) -> TransitionCallbackReturn:
        self.get_logger().info("Activating... starting timers")
        self._timer = self.create_timer(0.1, self._timer_cb)
        return TransitionCallbackReturn.SUCCESS

    # Called when: Active → Inactive
    def on_deactivate(self, state: State) -> TransitionCallbackReturn:
        self.get_logger().info("Deactivating... stopping timers")
        self.destroy_timer(self._timer)
        return TransitionCallbackReturn.SUCCESS

    # Called when: Inactive → Unconfigured
    def on_cleanup(self, state: State) -> TransitionCallbackReturn:
        self.get_logger().info("Cleaning up... releasing all resources")
        self.destroy_publisher(self._pub)
        return TransitionCallbackReturn.SUCCESS

    # Called when: any state → Finalized (error or shutdown)
    def on_shutdown(self, state: State) -> TransitionCallbackReturn:
        self.get_logger().info(f"Shutting down from state: {state.label}")
        return TransitionCallbackReturn.SUCCESS

    def _timer_cb(self):
        self.get_logger().info("Running — node is ACTIVE")


def main():
    rclpy.init()
    node = MyLifecycleNode()
    rclpy.spin(node)
    node.destroy_node()
    rclpy.shutdown()


if __name__ == "__main__":
    main()

#!/usr/bin/env python3
import rclpy
from rclpy.lifecycle import LifecycleNode, TransitionCallbackReturn
from rclpy.lifecycle import State
from std_msgs.msg import String


class TalkerLifecycle(LifecycleNode):
    def __init__(self):
        super().__init__("lifecycle_talker")
        self._pub = None
        self._timer = None

    def on_configure(self, state: State) -> TransitionCallbackReturn:
        # LifecyclePublisher — inactive until node activates
        self._pub = self.create_lifecycle_publisher(String, "chatter", 10)
        self.get_logger().info("Publisher created (inactive)")
        return TransitionCallbackReturn.SUCCESS

    def on_activate(self, state: State) -> TransitionCallbackReturn:
        super().on_activate(state)  # activates the lifecycle publisher
        self._timer = self.create_timer(1.0, self._publish)
        self.get_logger().info("Activated — publishing to /chatter")
        return TransitionCallbackReturn.SUCCESS

    def on_deactivate(self, state: State) -> TransitionCallbackReturn:
        super().on_deactivate(state)  # deactivates the lifecycle publisher
        self.destroy_timer(self._timer)
        return TransitionCallbackReturn.SUCCESS

    def on_cleanup(self, state: State) -> TransitionCallbackReturn:
        self.destroy_publisher(self._pub)
        return TransitionCallbackReturn.SUCCESS

    def on_shutdown(self, state: State) -> TransitionCallbackReturn:
        return TransitionCallbackReturn.SUCCESS

    def _publish(self):
        if self._pub.is_activated:
            msg = String()
            msg.data = "Hello from lifecycle node!"
            self._pub.publish(msg)


def main():
    rclpy.init()
    node = TalkerLifecycle()
    rclpy.spin(node)
    rclpy.shutdown()


if __name__ == "__main__":
    main()

from launch import LaunchDescription
from launch.actions import DeclareLaunchArgument
from launch.substitutions import LaunchConfiguration
from launch_ros.actions import Node


def generate_launch_description():
    use_sim_time = LaunchConfiguration("use_sim_time", default="false")

    # The nodes to manage (launched separately, just referenced here by name)
    lifecycle_nodes = [
        "robot_state_publisher",
        "map_server",
        "amcl",
        "controller_server",
        "bt_navigator",
    ]

    lifecycle_manager = Node(
        package="nav2_lifecycle_manager",
        executable="lifecycle_manager",
        name="lifecycle_manager_navigation",
        output="screen",
        parameters=[
            {"use_sim_time": use_sim_time},
            # The node names to manage (activate in order, deactivate in reverse)
            {"node_names": lifecycle_nodes},
            # Wait up to 20 seconds for each node to configure/activate
            {"autostart": True},
            {"bond_timeout": 4.0},
        ],
    )

    return LaunchDescription([
        DeclareLaunchArgument("use_sim_time", default_value="false"),
        lifecycle_manager,
    ])

# List all lifecycle nodes
ros2 lifecycle nodes

# Get the current state of a node
ros2 lifecycle get /my_lifecycle_node

# Trigger a state transition
ros2 lifecycle set /my_lifecycle_node configure
ros2 lifecycle set /my_lifecycle_node activate
ros2 lifecycle set /my_lifecycle_node deactivate
ros2 lifecycle set /my_lifecycle_node cleanup
ros2 lifecycle set /my_lifecycle_node shutdown

# List all available transitions from the current state
ros2 lifecycle list /my_lifecycle_node

# View the lifecycle state topic (publishes on /my_lifecycle_node/transition_event)
ros2 topic echo /my_lifecycle_node/transition_event

# Service interface (used internally by lifecycle_manager)
ros2 service call /my_lifecycle_node/change_state   lifecycle_msgs/srv/ChangeState   "{transition: {id: 1}}"   # id=1=configure, 3=activate, 4=deactivate, 7=cleanup