chatkit-botbuilder

ChatKit Botbuilder

Overview

Create production-grade chatbots using the OpenAI ChatKit framework. This skill enables building chatbots that:

• Integrate AI Agents: Use OpenAI Agents SDK for intelligent conversation handling
• Execute Tools: Connect MCP (Model Context Protocol) tools for real-world task execution
• Support Custom Backends: Build FastAPI backends with full protocol support
• Ensure User Isolation: Implement multi-user systems with JWT authentication
• Real-Time Synchronization: Enable live UI updates when chatbot performs actions
• Flexible Deployment: Deploy to web, mobile, or desktop applications

This skill provides the complete architecture pattern for ChatKit integration, from frontend configuration to backend server implementation.

When to Use This Skill

Use this skill when you need to:

1. Build a task management chatbot - Create conversational interfaces for task creation, updates, completion
2. Integrate AI into existing apps - Add ChatKit to dashboards, web apps, or platforms
3. Create specialized AI assistants - Build domain-specific chatbots with custom tool integrations
4. Implement multi-user chatbots - Create systems where each user has isolated conversations and data
5. Add real-time capabilities - Build chatbots that trigger actual application changes
6. Deploy AI conversations - Create chatbots that interact with your database and APIs

Architecture Overview

High-Level Flow

User Message
    ↓
ChatKit Frontend (React/Next.js)
    ↓ [JWT Token in Authorization Header]
    ↓
FastAPI Backend (ChatKit Server)
    ↓ [Extract user_id from JWT]
    ↓
OpenAI Agent (Agents SDK)
    ↓ [Needs tool execution]
    ↓
MCP Tools (Custom Tool Functions)
    ↓ [Creates/Updates/Lists data]
    ↓
Database (User-Isolated Data)
    ↓
Response → ChatKit → Frontend → User

Key Components

1. Frontend (Next.js + ChatKit SDK)
2. ChatKit UI component with conversation history
3. JWT token management in localStorage
4. Custom fetch wrapper with Bearer token authentication
5. Real-time auto-refresh to sync with backend changes

1. Backend (FastAPI + ChatKit Server)
2. ChatKit protocol endpoint handling requests
3. MyChatKitServer class extending ChatKitServer
4. User isolation through JWT middleware
5. Tool wrapper functions for automatic user_id injection

1. Agent (OpenAI Agents SDK)
2. Task management agent with instructions
3. Tool registration and execution
4. Session management

1. Tools (MCP + Custom Functions)
2. Wrapped functions injecting user_id automatically
3. Database operations with user isolation
4. Consistent error handling

1. Database
2. SQLModel ORM models
3. Per-user task filtering
4. Conversation persistence

Quick Start Workflow

Phase 1: Backend Setup (FastAPI)

1. Create ChatKit Server Class

from chatkit.server import ChatKitServer
from chatkit.store import Store

class MyChatKitServer(ChatKitServer):
    def __init__(self):
        store = CustomChatKitStore()
        super().__init__(store=store)

    async def respond(self, thread, input, context):
        """Process user message and stream AI response"""
        user_id = getattr(context, 'user_id', None)
        # Create agent with wrapped tools
        # Stream response using official pattern

2. Create MCP Tool Wrappers

# Extract user_id from context and inject into tool calls
def add_task_wrapper(title: str, description: str = None):
    return mcp_add_task(user_id=user_id, title=title, description=description)

def list_tasks_wrapper(status: str = "all"):
    return mcp_list_tasks(user_id=user_id, status=status)

3. Create FastAPI Endpoint

@router.post("/api/v1/chatkit")
async def chatkit_protocol_endpoint(request: Request):
    user_id = request.state.user_id  # From JWT middleware
    context = create_context_object(user_id=user_id)
    result = await chatkit_server.process(body, context)
    return StreamingResponse(result, media_type="text/event-stream")

4. Configure JWT Middleware

class JWTAuthMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        # Extract JWT token from Authorization header
        # Decode and set request.state.user_id
        # All endpoints have access to authenticated user_id

Phase 2: Frontend Setup (Next.js + React)

1. Configure ChatKit SDK

const chatKitConfig: UseChatKitOptions = {
  api: {
    url: `${API_BASE_URL}/api/v1/chatkit`,
    domainKey: 'your-domain-key',
    fetch: authenticatedFetch, // Custom fetch with JWT
  },
  theme: 'light',
  header: { enabled: true, title: { text: 'AI Chat' } },
  history: { enabled: true },
}

2. Create Authenticated Fetch Wrapper

async function authenticatedFetch(input, options) {
  const token = localStorage.getItem('access_token')
  const headers = {
    ...options?.headers,
    'Authorization': `Bearer ${token}`,
  }
  return fetch(input, { ...options, headers })
}

3. Integrate ChatKit Widget

import { ChatKitWidget } from '@openai/chatkit-react'

export default function Dashboard() {
  return (
    <div className="flex gap-4">
      {/* Your app content */}
      {showChat && (
        <ChatKitWidget {...chatKitConfig} />
      )}
    </div>
  )
}

4. Add Auto-Refresh for Real-Time Sync

useEffect(() => {
  if (!showChatKit) return

  // Refresh immediately when chat opens
  fetchTasks()

  // Refresh every 1 second for real-time updates
  const interval = setInterval(() => {
    fetchTasks()
  }, 1000)

  return () => clearInterval(interval)
}, [showChatKit])

Phase 3: Tool Implementation (MCP)

1. Create MCP Tools with User Isolation

def add_task(user_id: str, title: str, description: Optional[str] = None):
    """Create task - receives user_id from wrapper"""
    task = Task(
        id=str(uuid.uuid4()),
        user_id=user_id,  # Critical: ensure user isolation
        title=title,
        description=description,
        completed=False,
        created_at=datetime.utcnow(),
    )
    with Session(engine) as session:
        session.add(task)
        session.commit()

2. Register MCP Tools

mcp_server = MCPServer()
mcp_server.register_tool("add_task", add_task)
mcp_server.register_tool("list_tasks", list_tasks)
mcp_server.register_tool("delete_task", delete_task)
# ... more tools

Core Patterns & Best Practices

1. User Isolation Strategy

Three-Level Guarantee:

1. Middleware Level - JWT validation ensures only authenticated users
2. Tool Level - Wrapper functions automatically inject user_id
3. Database Level - All queries filtered by user_id

# Middleware extracts user_id from token
request.state.user_id = payload.get("user_id")

# Tool wrapper captures and injects it
def add_task_wrapper(title):
    return mcp_add_task(user_id=user_id, ...)

# Database enforces it
WHERE user_id = ? AND task_id = ?

2. Message Flow with User Context

User sends: "Create a task called 'Buy milk'"
    ↓
ChatKit Protocol: POST /api/v1/chatkit
    Header: Authorization: Bearer <JWT>
    Body: { "type": "message", "text": "Create..." }
    ↓
JWT Middleware:
    Extracts user_id from token → request.state.user_id
    ↓
ChatKit Server (MyChatKitServer.respond):
    Gets user_id from context
    Creates wrapper functions capturing user_id
    Passes wrappers to Agent
    ↓
OpenAI Agent:
    Receives message: "Create a task..."
    Selects tool: add_task_wrapper
    Calls: add_task_wrapper(title="Buy milk")
    ↓
Wrapper Function:
    Calls: mcp_add_task(user_id="user-123", title="Buy milk")
    ↓
MCP Tool:
    Creates task with correct user_id
    Returns: {"task_id": "...", "title": "Buy milk"}
    ↓
Agent Response:
    "I've created 'Buy milk' task ✓"