dnotifier 1.0.0

Real-time notification and messaging SDK for Python. WebSocket & HTTP, auth, binary frames, AI, RAG, and workflows.

DNotifier Python SDK

DNotifier is a real-time notification and messaging SDK for Python 3.9+. It supports WebSocket and HTTP transports, binary streaming, framed packets, an AI pipeline, knowledge-base (RAG) APIs, optional session logging, and multi-step workflows with agents and observability.

---

Features

• WebSocket and HTTP transport
• Secure authentication handshake
• Real-time messaging (text and binary)
• Large payloads with automatic chunking and reassembly
• AI: send_ai, fetch_ai_history, semantic search
• Knowledge base: add, update, get, list, delete, and search documents
• Chat history fetch and delete
• Optional SDK session logging (logs=True)
• Workflows: named agents, sequential pipelines, dashboard observability
• Plan limits from auth (get_plan_limits)

---

Installation

pip install dnotifier

---

Quick start

import asyncio
from dnotifier import DNotifier

async def main():
    notifier = DNotifier(
        app_id="your-app-id",
        secret="your-app-secret",
        transport="ws",  # or "http"
        user_id="user-123",
        on_connected=lambda: print("Connected"),
        on_message=lambda msg: print(msg.payload.to_json()),
        on_disconnected=lambda **kw: print("Disconnected", kw),
    )

    await notifier.connect()

    await notifier.send(
        sender_id="user-123",
        receiver_id="user-456",
        data={"type": "text", "text": "Hello from DNotifier"},
    )

    await notifier.disconnect()

asyncio.run(main())

---

Connection options

Parameter	Type	Description
app_id	str	Your DNotifier application id
secret	str	Application secret
transport	"ws" | "http"	WebSocket for realtime; HTTP for RPC-style AI/RAG
user_id	str	End-user or agent id for auth and routing
logs	bool	Enable SDK session logging (default False)
url	str	WebSocket URL override (default production wss://api.dnotifier.com/ecs/ws)
on_connected	callable	Called after successful connect
on_message	callable	Called for each incoming message
on_disconnected	callable	Called on disconnect (code, reason)

---

API overview

Method	Description
connect()	Authenticate and open transport
disconnect()	Close WebSocket connection
send(...)	Send realtime message
send_binary(...)	Send binary payload (WebSocket)
send_ai(...)	Send to AI pipeline
fetch_ai_history(...)	Fetch AI conversation history
delete_ai_history_message(...)	Delete one AI history entry
fetch_chat_history(...)	Request chat history
delete_chat_history_message(...)	Delete one chat message
search(...)	Semantic / vector search
add_document(...)	Add RAG document
update_document(...)	Update RAG document
delete_document(...)	Delete RAG document
list_documents(...)	List RAG documents
get_document(...)	Get RAG document + chunks
get_plan_limits()	Plan limits from last auth
run_workflow(...)	Run a registered workflow

---

Workflows

from dnotifier import DNotifier, Workflow, define_agent

researcher = define_agent(
    name="researcher",
    run=lambda ctx: ctx.send_ai(message=ctx.input, label="Research"),
)

workflow = Workflow(
    name="qa-pipeline",
    observability=True,
    entry=lambda ctx: ctx.run_agent("researcher"),
)
workflow.register_agent("researcher", researcher)

async def main():
    notifier = DNotifier(...)
    await notifier.connect()
    result = await notifier.run_workflow(workflow=workflow, input_data="What is RAG?")
    print(result)

---

Development

From dnotifier-python-sdk/:

pip install -e ".[dev]"
python -c "from dnotifier import DNotifier; print(DNotifier)"

---

Publishing (maintainers)

The published PyPI wheel is minified and encrypted so source logic is not shipped in readable form (same approach as the JS and Dart SDKs).

# Prepare minified + PyArmor-encrypted package in publish_dist/
python scripts/minify_for_publish.py

# Bump version, build wheel, upload to PyPI
python scripts/publish_prod.py

---

License

Proprietary — see LICENSE.