Python SDK
rumoro on PyPI. Every Rumoro API operation from Python, sync and async.
The rumoro package has one method per endpoint, grouped by resource, and a typed model for every response. It is built from the OpenAPI document behind the API reference. It needs Python 3.11 or newer and works sync and async.
pip install rumoroFirst call
from rumoro import Rumoro
rumoro = Rumoro(api_key="ref_...")
page = rumoro.mentions.search(platform="reddit", relevant=True, limit=20)
for mention in page.data:
score = mention.classification.relevance if mention.classification else None
print(mention.post.platform.value, score, mention.post.url)A call returns the endpoint's model, a str for CSV exports, or None for a 204. An error raises RumoroError with status, code and message.
from rumoro import RumoroError
try:
rumoro.keywords.create(term="a", kind="brand")
except RumoroError as err:
print(err.status, err.code, err.message) # 400 validation_error term: ...If the body is not an API error, such as a proxy's HTML page, code is http_<status>.
Calls
- Ids are positional.
- Query parameters are keyword arguments in snake_case, such as
keyword_ids.rangeandfrombecomerange_andfrom_. - A body can be keyword arguments, a dict or a model, with the API's field names such as
channelIds. - Enum values can be plain strings.
- Time parameters (
since,until,snoozed_until) work withdatetimeanddateobjects, ISO 8601 strings and epoch milliseconds.
rumoro.keywords.create(term="driftwood", kind="brand", platforms=["reddit", "github"])
rumoro.keywords.update("kw_...", muted=True)
rumoro.mentions.update("mm_...", status="done", note="Answered in the thread")
rumoro.people.merge("aut_...", into="aut_...")
rumoro.alerts.create(name="Pricing questions", mode="instant", filter={"intents": ["pricing"]}, channelIds=["dest_..."])
rumoro.analytics.summary(range_="7d", compare=True, timezone="America/New_York")For the next page, send next_cursor as cursor, and stop when it is None.
cursor = None
while True:
page = rumoro.mentions.search(sentiment="negative", limit=100, cursor=cursor)
for mention in page.data:
print(mention.id, mention.post.url)
cursor = page.next_cursor
if not cursor:
breakA CSV export returns a string.
csv_text = rumoro.mentions.export(since="2026-10-01T00:00:00Z")
with open("mentions.csv", "w", encoding="utf-8") as file:
file.write(csv_text)Async
AsyncRumoro has the same methods, as coroutines.
import asyncio
from rumoro import AsyncRumoro
async def main() -> None:
async with AsyncRumoro(api_key="ref_...") as rumoro:
summary = await rumoro.analytics.summary(range_="30d")
print(summary.matched, summary.relevant)
asyncio.run(main())Resources
| Attribute | Methods |
|---|---|
keywords | create, list, get, update, delete, health |
groups | create, list, get, update, delete |
mentions | search, get, update, export, export_json |
attention | list, dismiss |
views | create, list, get, update, delete |
filters | get, update |
people | list, get, update, merge, split, export, activities, log_activity, delete_activity |
segments | create, list, get, update, delete |
alerts | create, list, get, update, delete, test, run, mute, unmute |
channels | create, list, get, update, delete, test, rotate_secret, deliveries |
analytics | summary, series, breakdown, share_of_voice, reviews |
company | get, update |
members | list, remove, invitations, invite, revoke_invitation |
usage | get, breakdown |
billing | wallet, ledger, top_up, invoices, invoice_url |
api_keys | create, list, revoke |
auth | whoami |
system | health |
Lower level
rumoro.client is the generated httpx-based client with your key. Each rumoro.api module is one operation, and its sync_detailed returns the full response.
from rumoro.api.mentions import search_mentions
rumoro = Rumoro("ref_...", base_url="https://api.rumoro.dev", timeout=10.0, headers={"x-request-source": "crm-sync"})
response = search_mentions.sync_detailed(client=rumoro.client, limit=10)