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_followersandnew_subscribers, pluschat.new_messagesandchat.new_rantsas 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.
1 maintainer
Metadata
1 maintainer
Release files for cocorum 2.18.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| cocorum-2.18.0.tar.gz | 404.1 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| cocorum-2.18.0-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 460.9 kB
Release files / cocorum-2.18.0.tar.gz
| Download URL | cocorum-2.18.0.tar.gz |
|---|---|
| Size | 404.1 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
734c7dd40c854a1ba056c5a86ff544c914aa8a72d315d1d1102731c4c9378740
|
|
BLAKE2b-256 checksum How to use checksums |
bd8cefd47e42470e214f3120cee4ab11e3097417768519b86b59101cb95c9ab5
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Release files / cocorum-2.18.0-py3-none-any.whl
| Download URL | cocorum-2.18.0-py3-none-any.whl |
|---|---|
| Size | 56.7 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
760e99125ac32d4ee5307f836ff8291bfb29bf5903612da63ed20c3243bf4384
|
|
BLAKE2b-256 checksum How to use checksums |
cabfaca46be0e36110af6568b55d33bdfa7dc287af741bfa1f6f97f90fa068d9
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
No |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
