Cocorum: Rumble Live Stream API Python Wrapper

A Python wrapper for the Rumble Live Stream API v1.0 (beta), with some quality of life additions, such as:

• Implementing a cache of the API data with an automatic time to live enforcer. If the stored API data is older than refresh_rate (and you are querying an endpoint that can actually change over time), the wrapper will automatically refresh in the background before handing you the fresh endpoint data.
• All timespamps are parsed to seconds since Epoch, UTC timezone.
• Additional new_followers and new_subscribers, plus chat.new_messages and chat.new_rants as endpoints that automatically clear themselves when read.

Installation:

You can find this project on PyPi.org, and install it using any PyPi-compatible method (including Pip). Alternatively, you can view and download the source code on GitHub.

Usage:

I recommend taking a gander at the full documentation, but here's a basic intro:

For the most part, attributes that are not added features have the same name as the direct JSON counterparts, with the exception of adding prefixes to some things that have the same name in the JSON as Python builtin functions. For example, thing/id in JSON is thing.thing_id in this Python wrapper.

from cocorum import RumbleAPI

## API_URL is Rumble Live Stream API URL with key
api = RumbleAPI(API_URL, refresh_rate = 10)

print(api.username)
## Should display your Rumble username

print("Latest follower:", api.latest_follower)
## Should display your latest follower, or None if you have none.

if api.latest_subscriber:
    print(api.latest_subscriber, f"subscribed for ${api.latest_subscriber.amount_dollars}")
## Should display your latest subscriber if you have one.

livestream = api.latest_livestream # None if there is no stream running

if livestream:
    print(livestream.title)
    print("Stream visibility is", livestream.visibility)

    # We will use this later
    STREAM_ID = livestream.stream_id

    print("Stream ID is", STREAM_ID)
    
    import time # We'll need this Python builtin for delays and knowing when to stop 

    # Get messages for one minute
    start_time = time.time()

    # Continue as long as we haven't been going for a whole minute, and the livestream is still live
    while time.time() - start_time < 60 and livestream.is_live:
        # For each new message...
        for message in livestream.chat.new_messages:
            # Display it
            print(message.username, "said", message)

        # Wait a bit, just to keep the loop from maxxing a CPU core
        time.sleep(0.1)

Rumble internal API submodules

This part of Cocorum's wrapping is not part of the official Rumble Live Stream API. It includes the following submodules:

• chatapi
• servicephp
• uploadphp
• scraping
• utils (very rare cases)

Example usage of cocorum.chatapi:

from cocorum import chatapi

# Stream ID can be base 10 int or base 36 str
with chatapi.ChatAPI(stream_id = STREAM_ID) as chat:
    chat.clear_mailbox() # Erase messages that were still visible before we connected

    # Get messages for one minute
    start_time = time.time()
    while time.time() - start_time < 60 and (msg := chat.get_message()):
        print(msg.user.username, "said", msg)

Conclusion

Hope this helps!

I, Wilbur Jaywright, and my brand, Marswide BGL, have no official association with Rumble Corp. beyond that of a normal user and/or channel on the Rumble Video platform. This wrapper is not officially endorsed by Rumble Corp. or its subsidiaries.

Copyright 2026 Wilbur Jaywright d.b.a. Marswide BGL.

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

S.D.G.