llmrix-python-sdk

Official Python SDK for llmrix — AI Agent Platform.

• Python 3.10+
• Built on httpx for async-ready HTTP and SSE streaming
• Fully typed — compatible with mypy --strict
• Context-manager support for automatic resource cleanup
• Sync streaming via a clean for event in ... interface

---

Installation

pip install llmrix-python-sdk

Or with uv:

uv add llmrix-python-sdk

---

Quick Start

from llmrix import LlmrixClient, ConversationCreateRequest, HitlDecision
from llmrix.streaming import MessageChunkEvent, RunEndEvent

with LlmrixClient(base_url="https://www.llmrix.com", api_key="sk-xxx") as client:
    # Create a conversation
    conv = client.conversations.create(ConversationCreateRequest(title="My Chat"))

    # Stream the response
    for event in client.chat(conv.id).stream("Hello, llmrix!"):
        if isinstance(event, MessageChunkEvent):
            print(event.content, end="", flush=True)
    print()

---

API Reference

LlmrixClient

LlmrixClient(
    base_url: str,
    api_key: str | None = None,
    timeout: float = 60.0,
)

Can be used as a context manager (with statement) or constructed directly. Call .close() when done if not using a context manager.

Parameter	Type	Description
base_url	str	Server base URL — required
api_key	str | None	Bearer token for Authorization header
timeout	float	Request timeout in seconds (default: 60.0)

client.conversations

Returns the ConversationsResource for CRUD and message history.

client.chat(conversation_id)

Returns a ChatResource scoped to the given conversation ID.

---

ConversationsResource

Method	Returns	Description
create(req)	Conversation	Create a new conversation
list(last_id?, size?)	PageResult[Conversation]	Cursor-paginated list
get(id)	Conversation	Fetch a single conversation
update(id, req)	Conversation	Update title / agent
delete(id)	None	Permanently delete
messages(id, last_id?, size?)	PageResult[Message]	Paginated message history

Cursor pagination:

page = client.conversations.list(size=20)
while page.has_more:
    page = client.conversations.list(last_id=page.items[-1].seq, size=20)

---

ChatResource

Obtained via client.chat(conversation_id).

Method	Returns	Description
stream(message, **opts)	Iterable[StreamEvent]	Stream a plain-text message
stream_request(req)	Iterable[StreamEvent]	Stream a full ChatRequest
stop()	None	Cancel the current run
decide(decisions)	None	Submit HITL decisions

Options for stream()

Parameter	Type	Description
model	str	Override the model for this turn
thinking	bool	Enable chain-of-thought reasoning
attachments	list[Attachment]	File/image attachments

---

Stream Event Types

from llmrix.streaming import (
    MessageChunkEvent,
    RunStartEvent,
    RunEndEvent,
    ToolStartEvent,
    ToolEndEvent,
    SubagentStartEvent,
    SubagentEndEvent,
    HitlInterruptEvent,
    ErrorEvent,
    CancelledEvent,
)

for event in client.chat(conv.id).stream("Search the web for cats"):
    match event:
        case MessageChunkEvent(content=c):
            print(c, end="", flush=True)
        case ToolStartEvent(name=n):
            print(f"\n[tool] {n}")
        case HitlInterruptEvent():
            client.chat(conv.id).decide([HitlDecision.approve()])
        case RunEndEvent():
            break
        case ErrorEvent(message=m):
            raise RuntimeError(m)

Class	Channel	Description
RunStartEvent	lifecycle	Agent run started
RunEndEvent	lifecycle	Agent run completed
MessageChunkEvent	messages	Incremental text chunk
ToolStartEvent	tools	Tool invocation started
ToolEndEvent	tools	Tool invocation finished
SubagentStartEvent	tools	Sub-agent spawned
SubagentEndEvent	tools	Sub-agent finished
HitlInterruptEvent	hitl	Human approval required
ErrorEvent	error	Run failed with error
CancelledEvent	error	Run was cancelled

---

Error Handling

from llmrix import LlmrixError, LlmrixApiError, LlmrixAuthError

try:
    conv = client.conversations.get("bad-id")
except LlmrixAuthError as e:
    print(f"Authentication failed: {e.status_code}")
except LlmrixApiError as e:
    print(f"API error {e.status_code}: {e}")
except LlmrixError as e:
    print(f"SDK error: {e}")

Exception	When raised
LlmrixError	Base class; network / serialization errors
LlmrixApiError	Non-2xx HTTP response; exposes .status_code and .response_body
LlmrixAuthError	HTTP 401 or 403; extends LlmrixApiError

---

Advanced Usage

HITL (Human-in-the-Loop) flow

chat = client.chat(conv.id)
for event in chat.stream("Delete all log files"):
    if isinstance(event, HitlInterruptEvent):
        answer = input(f"Approve '{event.action_name}'? (y/n): ")
        decision = HitlDecision.approve() if answer == "y" else HitlDecision.reject("user declined")
        chat.decide([decision])
    if isinstance(event, RunEndEvent):
        break

Collecting the full response

def run_to_completion(client: LlmrixClient, conv_id: str, message: str) -> str:
    chunks = []
    for event in client.chat(conv_id).stream(message):
        if isinstance(event, MessageChunkEvent):
            chunks.append(event.content)
        if isinstance(event, RunEndEvent):
            break
        if isinstance(event, ErrorEvent):
            raise RuntimeError(event.message)
    return "".join(chunks)

Registering custom event types

from llmrix import register_event
from dataclasses import dataclass

@dataclass
class MyCustomEvent:
    value: str

register_event("my_channel", "my_type", MyCustomEvent)

---

Development

git clone https://github.com/llmrix/llmrix-python-sdk.git
cd llmrix-python-sdk
pip install -e ".[dev]"
pytest

---

Contributing

Contributions are welcome! Please read CONTRIBUTING.md first.

---

License

Apache License 2.0 — see the LICENSE file.