Python SDK

Building conversational AI applications in Python should be simple. ChatBotKit's Python SDK provides a lightweight, fully-async interface for integrating intelligent bots, agents, and conversational workflows into Python applications. Instead of running heavy AI logic locally, the SDK offloads orchestration, model management, and execution to ChatBotKit's servers - so your Python code stays focused on business logic while benefiting from enterprise-grade AI infrastructure.

Unlike local AI frameworks that burden your application with model loading, token management, and execution complexity, the Python SDK keeps your application lean. As new AI models emerge and ChatBotKit adds capabilities, your applications automatically benefit without code changes. This server-first architecture means you build faster, scale easier, and stay compatible with tomorrow's AI innovations.

What You Can Build

Autonomous Agents and Workflows

Create intelligent agents that operate autonomously or in response to user requests:

• Autonomous Agents: Build agents that make decisions, take actions through skillsets, and accomplish goals with minimal human intervention
• Workflow Automation: Orchestrate multi-step processes where agents create and manage todo lists, process files, and coordinate with other agents
• Background Job Processing: Run long-running operations asynchronously using the SDK's fully async architecture
• Scheduled Triggers: Execute agents on timelines or in response to events

Conversational Interfaces

Build chat experiences and conversational AI into your Python applications:

• Chat Completions: Stream responses token-by-token for responsive, real-time chat interfaces
• Multi-turn Conversations: Maintain full conversation history and context across exchanges
• Session Management: Automatically handle user sessions and conversation persistence
• Streaming Responses: Real-time token streaming that enables progressive message rendering

Data Access and Manipulation

Access and manage all ChatBotKit resources programmatically:

• Bots and Conversations: Create, retrieve, and manage bots and ongoing conversations
• Datasets: Build and search knowledge bases that your bots use for context and retrieval
• Skillsets and Abilities: Configure agent capabilities and custom actions
• Spaces and Files: Access collaborative workspaces and managed files
• Memories: Create, search, and leverage persistent memory systems
• Contacts and Teams: Manage users, contacts, and team collaboration

Key Features

Fully Async Architecture

Built from the ground up with async/await, the SDK scales efficiently and enables concurrent operations:

import asyncio
from chatbotkit import ChatBotKit

async def main():
    async with ChatBotKit(secret="your-api-key") as cbk:
        # Create multiple conversations concurrently
        conversations = await asyncio.gather(
            cbk.conversation.create(botId="bot-1"),
            cbk.conversation.create(botId="bot-2"),
            cbk.conversation.create(botId="bot-3"),
        )
        
        # Process them in parallel
        tasks = [
            cbk.conversation.send(c["id"], {"text": "Hello"})[0]
            for c in conversations
        ]
        results = await asyncio.gather(*tasks)

asyncio.run(main())

Server-Side AI Orchestration

Complex AI operations run on ChatBotKit's infrastructure, not your application. This means:

• Lightweight Runtime: Your application stays lean - no model loading, no token management overhead
• Automatic Model Selection: The platform chooses optimal models based on task requirements
• Future-Proof: New models, capabilities, and optimizations automatically benefit your applications
• Enterprise Reliability: Benefit from platform-level error handling, retries, and failover

Type-Safe Development

Comprehensive type hints throughout the SDK enable IDE autocomplete and catch errors before runtime:

from chatbotkit import ChatBotKit
from chatbotkit.types import (
    ConversationCompleteRequest,
    ConversationCompleteStreamItemType
)

async def chat_with_ai():
    async with ChatBotKit(secret="your-api-key") as cbk:
        # Full type hints for requests and responses
        request: ConversationCompleteRequest = {
            "messages": [
                {"type": "user", "text": "What's the weather?"}
            ]
        }
        
        completion = cbk.conversation.complete(None, request)
        
        # Type-safe streaming
        async for event in completion.stream():
            if event.type == ConversationCompleteStreamItemType.TOKEN:
                print(event.data.token, end="", flush=True)

Unified Resource Access

Access all ChatBotKit resources through a single, intuitive client interface:

async with ChatBotKit(secret="api-key") as cbk:
    # Bot management
    await cbk.bot.create({"name": "Support Bot"})
    
    # Dataset operations
    await cbk.dataset.record.create(datasetId="ds-1", {"content": "FAQ"})
    
    # Conversation handling
    await cbk.conversation.send(conversationId, {"text": "message"})
    
    # File management
    await cbk.file.upload(fileId, fileContent)
    
    # Memory operations
    await cbk.memory.create({"content": "important info"})

Common Use Cases

Python Web Applications

Integrate conversational AI into FastAPI, Django, or other Python web frameworks. The SDK's async nature fits perfectly with modern async web applications:

from fastapi import FastAPI, WebSocket
from chatbotkit import ChatBotKit

app = FastAPI()

@app.websocket("/ws/chat")
async def websocket_chat(websocket: WebSocket):
    await websocket.accept()
    async with ChatBotKit(secret="api-key") as cbk:
        while True:
            user_message = await websocket.receive_text()
            completion = cbk.conversation.complete(None, {
                "messages": [{"type": "user", "text": user_message}]
            })
            
            async for event in completion.stream():
                if event.type == "token":
                    await websocket.send_text(event.data.token)

Data Processing Pipelines

Build AI-powered data workflows that leverage ChatBotKit agents for analysis, extraction, and transformation:

• Process documents by asking agents to extract key information
• Generate insights by having agents analyze data and produce summaries
• Classify and categorize data using conversational AI
• Create intelligent automation workflows that adapt based on data

Scheduled AI Workflows

Create background jobs that run agents on a schedule or in response to events. Agents can read from datasets, process files from Spaces, create records, and send results to integrations - all without user interaction.

Multi-Agent Orchestration

Coordinate multiple specialized agents working toward a common goal:

async def orchestrate_research():
    async with ChatBotKit(secret="api-key") as cbk:
        # Send research request to researcher agent
        researcher_result = await cbk.bot.conversation(
            botId="researcher-bot",
            {"text": "Research machine learning trends"}
        )
        
        # Pass results to analyst agent
        analyst_result = await cbk.bot.conversation(
            botId="analyst-bot",
            {"text": f"Analyze these findings: {researcher_result}"}
        )
        
        # Get executive summary from summarizer agent
        summary = await cbk.bot.conversation(
            botId="summary-bot",
            {"text": f"Summarize this analysis: {analyst_result}"}
        )

Getting Started

Installation

Install the SDK from PyPI with pip:

pip install chatbotkit

Or directly from GitHub:

pip install "chatbotkit @ git+https://github.com/chatbotkit/python-sdk.git"

For agent functionality, install with the agent extra:

pip install "chatbotkit[agent]"

Requires Python 3.10 or later. The SDK uses httpx for HTTP operations and is fully async.

Basic Setup

Create a client and start chatting:

import asyncio
from chatbotkit import ChatBotKit

async def main():
    async with ChatBotKit(secret="your-api-key") as cbk:
        # Create a conversation
        conversation = await cbk.conversation.create(
            botId="your-bot-id"
        )
        
        # Send a message and stream the response
        completion = cbk.conversation.complete(
            conversation["id"],
            {"messages": [{"type": "user", "text": "Hello!"}]}
        )
        
        async for event in completion.stream():
            if event.type == "token":
                print(event.data.token, end="", flush=True)

asyncio.run(main())

Configuration

Customize the client with environment-specific settings:

cbk = ChatBotKit(
    secret="your-api-key",
    base_url="https://api.chatbotkit.com",  # API endpoint
    run_as_user_id="user-123",               # Execute as specific user
    timezone="America/New_York",             # Timezone for operations
)

Agent Framework

For advanced use cases, build local agent logic with the agent framework:

from chatbotkit.agent import Tool, execute

# Define custom tools
search_tool = Tool(
    name="search",
    description="Search for information",
    execute=lambda query: search_database(query)
)

# Execute agent with tools
result = await execute(
    instruction="Find information about...",
    tools=[search_tool],
    model="gpt-4"
)

Advanced Features

Streaming and Real-Time Updates

The SDK fully supports streaming for real-time token-by-token responses:

completion = cbk.conversation.complete(
    conversationId,
    {"messages": messages}
)

async for event in completion.stream():
    match event.type:
        case ConversationCompleteStreamItemType.TOKEN:
            # Handle token
            print(event.data.token, end="", flush=True)
        case ConversationCompleteStreamItemType.DONE:
            # Completion finished
            print("\nDone!")

Context Management

Use Python's async context managers for clean resource handling:

async with ChatBotKit(secret="key") as cbk:
    # Client automatically closes connections
    result = await cbk.conversation.send(...)
    
# Cleanup happens automatically

Error Handling

The SDK provides detailed error information for robust applications:

try:
    await cbk.conversation.create(botId="nonexistent")
except Exception as e:
    print(f"Error: {e}")
    # Handle error appropriately

Concurrent Operations

Leverage Python's async nature to handle multiple operations concurrently:

# Process 100 items in parallel with controlled concurrency
semaphore = asyncio.Semaphore(10)

async def process_item(cbk, item):
    async with semaphore:
        return await cbk.conversation.complete(None, {
            "messages": [{"type": "user", "text": f"Process: {item}"}]
        })

tasks = [process_item(cbk, item) for item in items]
results = await asyncio.gather(*tasks)

Integration with Other Features

The Python SDK works seamlessly with all ChatBotKit capabilities:

• Spaces and Files: Upload files, manage shared workspaces, and access file contents programmatically
• Datasets: Create and search datasets that power your bots' knowledge bases
• Memory System: Create, search, and leverage persistent memories within conversations
• Skillsets: Configure custom abilities and actions that agents can use
• Tasks and Triggers: Schedule agent execution or trigger workflows based on events
• Multi-Agent Orchestration: Coordinate multiple specialized agents toward complex goals

Developer Resources

• GitHub Repository: chatbotkit/python-sdk
• PyPI Package: chatbotkit on PyPI
• Python 3.10+: Full async/await support for modern Python applications
• Example Projects: Sample implementations and patterns in the repository
• Community Support: Reach out on Discord or GitHub for questions

The Python SDK makes it straightforward to build intelligent, conversational applications in Python while maintaining a lightweight, efficient architecture. Install the package and start building in minutes!