Skip to main content

Python SDK

pip install dialogbrain

For LangChain/LangGraph support:

pip install "dialogbrain[langchain]"

Requirements

  • Python 3.10+
  • httpx (installed automatically)

Basic Usage

import asyncio
from dialogbrain import DialogBrainClient

async def main():
async with DialogBrainClient(api_key="db_live_YOUR_KEY") as client:
# List conversations
conversations = await client.conversations.list(limit=10)

# Get messages from a conversation
messages = await client.conversations.get_messages("telegram:123456789")

# Send a message
await client.messages.send(
thread_id="telegram:123456789",
message_text="Hello from the Python SDK!",
)

# Search contacts
results = await client.contacts.search(query="Alice", channel="telegram")

# Request handoff
await client.handoff(thread_id="telegram:123456789", reason="Customer escalation")

asyncio.run(main())

API Reference

DialogBrainClient

client = DialogBrainClient(
api_key="db_live_...",
base_url="https://api.dialogbrain.com", # optional
)

Supports async with context manager or manual await client.close().

client.conversations

MethodDescription
list(limit=50, offset=0)List conversations
get(conversation_id)Get one conversation
get_messages(conversation_id, limit=50)Get messages

client.messages

MethodDescription
send(thread_id, message_text)Send a message

client.contacts

MethodDescription
list()List contacts
get(contact_id)Get one contact
update(contact_id, **fields)Update contact
search(query, channel=None)Semantic search

client.search

MethodDescription
rag(query, limit=5)Search knowledge base

client.handoff(thread_id, reason=None)

Request human takeover for a thread.

Webhook Verification

from dialogbrain import verify_webhook

@app.post("/webhook")
async def handle(request: Request):
body = await request.body()
sig = request.headers["x-dialogbrain-signature"]
secret = "whsec_YOUR_SECRET"

if not verify_webhook(body, sig, secret):
raise HTTPException(401)

event = json.loads(body)
...

LangChain Tools

See LangChain Integration.