Python SDK

pip install rsflowhub
# poetry add rsflowhub · uv add rsflowhub

export RSFLOWHUB_API_KEY=rfh_your_api_key

from rsflowhub import RSFlowHub

client = RSFlowHub()  # reads RSFLOWHUB_API_KEY, or RSFlowHub(api_key="rfh_...")

res = client.ai.sentiment_analysis(
    text="Setup took two minutes and support answered right away.",
)

print(res.result["sentiment"])  # "positive"
print(res.result)               # the whole AI output
print(res.request_id)           # "req_..." quote it when you contact support

import asyncio
from rsflowhub import AsyncRSFlowHub


async def main() -> None:
    async with AsyncRSFlowHub() as client:
        # run several calls at the same time
        sentiment, summary = await asyncio.gather(
            client.ai.sentiment_analysis(text=review),
            client.ai.text_summarize(text=review, length="short"),
        )
        print(sentiment.result["sentiment"], summary.result)


asyncio.run(main())

# The method name is the endpoint slug in snake_case (camelCase also works)
client.ai.translation(text="Hello", target_language="French")
client.ai.pii_redaction(text="Call me on 555-0100")
client.ai.ticket_analyze(ticket="My order never arrived")

# The body can also be one dict
client.ai.chat({"message": "Hi!"})

# An AI API released after you installed the SDK works the same way
client.ai.brand_new_endpoint(text="...")
client.call("ai", "brand_new_endpoint", text="...")

# A resource the SDK does not know yet
client.resource("webhooks").list()

# Any route: method, path, body, query
usage = client.request("GET", "/account/usage", query={"page": 2})

from pathlib import Path

# Ids in the URL come first, then the body
client.knowledge_bases.ask(12, question="Refund policy?")
client.chatbots.get_conversation(3, "conv_abc123")

# ...or by name
client.knowledge_bases.ask(id=12, question="Refund policy?")

# A pathlib.Path, bytes or an open binary file makes it a file upload
client.knowledge_bases.create_document(
    12,
    file=Path("refund-policy.pdf"),
    metadata={"source": "help-center"},  # sent as metadata[source]
)

from pathlib import Path

# 1. A knowledge base for your help center
kb = client.knowledge_bases.create(name="Help center")
kb_id = kb.get_path("knowledge_base.id")

# 2. Add content: a document, or a single FAQ item
client.knowledge_bases.create_document(kb_id, file=Path("refund-policy.pdf"))
client.knowledge_bases.create_item(
    kb_id,
    content="How long do refunds take?",
    answer="Refunds reach your card in 5 to 7 business days.",
)

# 3. Ask questions grounded in that content
answer = client.knowledge_bases.ask(kb_id, question="When will I get my money back?")
print(answer.result)

# A chatbot (created in the dashboard or with client.chatbots.create(...))
reply = client.chatbots.send_message(chatbot_id, message="Where is my order?")
conversation_id = reply["conversation_id"]

import rsflowhub

try:
    client.ai.text_summarize(text=article)
except rsflowhub.ValidationError as e:
    print(e.errors)                       # {"text": ["The text field is required."]}
except rsflowhub.InsufficientCreditsError as e:
    print(e.required_credits, e.available_credits)
except rsflowhub.RateLimitError as e:
    print(e.retry_after)                  # already retried max_retries times
except rsflowhub.APIError as e:
    print(e.status_code, e.code, e.message, e.request_id)
except rsflowhub.APIConnectionError:
    print("network problem or timeout")   # APITimeoutError is a subclass

client = RSFlowHub(
    api_key="rfh_...",                       # default: RSFLOWHUB_API_KEY
    base_url="https://rsflowhub.com/api/v1", # default: RSFLOWHUB_BASE_URL
    timeout=60.0,                            # seconds per attempt
    max_retries=2,                           # for 429, 502, 503, 504 and network errors
    default_headers={},                      # extra headers on every request
    http_client=None,                        # your own httpx.Client (proxies, event hooks, tests)
    live_spec=True,                          # look up unknown non-AI methods in the OpenAPI document
)

# Per call
client.ai.chat(message="Hi", timeout=10, max_retries=0, extra_headers={"X-Trace": "abc"})

res.rate_limit   # {"limit": 60, "remaining": 59, "reset": None}

from contextlib import asynccontextmanager

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

import rsflowhub
from rsflowhub import AsyncRSFlowHub

client = AsyncRSFlowHub()


@asynccontextmanager
async def lifespan(app: FastAPI):
    yield
    await client.aclose()


app = FastAPI(lifespan=lifespan)


class Comment(BaseModel):
    text: str


@app.post("/moderate")
async def moderate(comment: Comment):
    try:
        res = await client.ai.comment_moderate(comment=comment.text)
    except rsflowhub.ValidationError as e:
        raise HTTPException(422, detail=e.errors)
    except rsflowhub.APIError as e:
        raise HTTPException(e.status_code, detail=e.message)
    return res.result