Python SDK

rydhm is the Python client for the Rydhm API. It is generated from the API’s OpenAPI document, so its models match what the API accepts and returns.

Install

pip install rydhm

The SDK is built on httpx and pydantic.

Initialize

import os
from rydhm import Rydhm
client = Rydhm(api_key=os.environ["RYDHM_API_KEY"])

If you pass no api_key, the client reads RYDHM_API_KEY from the environment.

For asyncio, use AsyncRydhm the same way and await each call.

How methods are named

Methods are grouped by resource: client.experts, client.knowledge, client.conversations, client.chat, client.voice, client.meetings, client.tools, client.analytics. Path parameters come first, as positional arguments; everything else is a keyword argument in snake_case:

client.experts.get(expert_id)                       # GET /v1/agents/{id}
client.knowledge.search(expert_id, q="refunds")     # GET /v1/agents/{id}/kb/search?q=refunds
client.chat.send(conversation_id, content="Hi")     # POST /v1/chat/conversations/{id}/messages

Responses are pydantic models: read fields as attributes, such as expert.name.

Every method needs a scope on your key. See Scopes.

Common tasks

List experts

experts = client.experts.list()
for expert in experts:
    print(expert.id, expert.name, expert.visibility)
expert = client.experts.get(experts[0].id)

Scope: experts:read.

Start a chat and send a message

conversation = client.chat.start(agent_id=expert_id)
message = client.chat.send(conversation.id, content="What can you help with?")

chat.send stores your message. Read the expert’s reply from the stream. Scope: chat:write.

Stream the reply

reply = ""
for frame in client.chat.stream(conversation.id):
    if frame.type == "sources":
        print(f"Grounded in {frame.count} passages")
    elif frame.type == "delta":
        reply += frame.content
        print(frame.content, end="", flush=True)
    elif frame.type == "error":
        raise RuntimeError(frame.message)

Each frame is a model with a type: sources, delta, tool_call, tool_result, message, done or error. The loop ends after done or error. See Stream the reply.

To continue the conversation, call chat.send and chat.stream again with the same id.

Read a conversation

messages = client.conversations.get(conversation.id)
for m in messages:
    print(f"{m.role}: {m.content}")
recent = client.conversations.list(agent_id=expert_id)  # the 50 most recent

Scope: conversations:read.

Search knowledge

hits = client.knowledge.search(expert_id, q="refund policy")
for hit in hits:
    print(round(hit.score, 2), hit.text[:80])

Scope: knowledge:read. mode ("hybrid" or "vector") is optional.

Add a document

doc = client.knowledge.add(
    expert_id,
    filename="returns.md",
    mime_type="text/markdown",
    content="# Returns\n\nItems can be returned within 30 days.",
)
print(doc.status)  # "ready": indexed before the call returns

Scope: knowledge:write. For a PDF, Word, Markdown, text or CSV file, use client.knowledge.add_file(expert_id, filename=..., data=...) with the file as base64.

Send an expert to a meeting

meeting = client.meetings.join(
    idempotency_key="crm-event-4821",
    agent_id=expert_id,
    join_url="https://meet.google.com/abc-defg-hij",
    mode="speaker",
)

Scope: meetings:join. Always pass an idempotency key: a retry with the same key returns the first result instead of sending a second bot.

Errors

A non-2xx response raises. Each common status has its own exception, and all of them extend ApiError:

from rydhm import ForbiddenError, Rydhm
from rydhm.core.api_error import ApiError
try:
    client.experts.create(name="Support")
except ForbiddenError as e:
    print(e.body.error)           # "insufficient_scope"
    print(e.body.missing_scopes)  # ["experts:write"]
except ApiError as e:
    print(e.status_code, e.body)

e.body is the API’s error body, with error (or code), message and the request id.

Retries

By default the SDK retries a request up to 2 times, with exponential backoff, when the response is 408, 409, 429 or any 5xx. Change it for the whole client with max_retries, or for one call:

client.chat.start(agent_id=expert_id, request_options={"max_retries": 0})

Headers and request ids

with_raw_response gives you the response headers too, such as the request id and rate-limit state:

response = client.chat.with_raw_response.start(agent_id=expert_id)
print(response.headers.get("x-request-id"))
print(response.headers.get("x-ratelimit-remaining"))
conversation = response.data

Every method also takes request_options, a dictionary with timeout (seconds), max_retries and additional_headers.