chatkit-botbuilder

What this skill does

# 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)**
   - ChatKit UI component with conversation history
   - JWT token management in localStorage
   - Custom fetch wrapper with Bearer token authentication
   - Real-time auto-refresh to sync with backend changes

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

3. **Agent (OpenAI Agents SDK)**
   - Task management agent with instructions
   - Tool registration and execution
   - Session management

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

5. **Database**
   - SQLModel ORM models
   - Per-user task filtering
   - Conversation persistence

---

## Quick Start Workflow

### Phase 1: Backend Setup (FastAPI)

**1. Create ChatKit Server Class**

```python
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**

```python
# 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**

```python
@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**

```python
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**

```typescript
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**

```typescript
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**

```typescript
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**

```typescript
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**

```python
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**

```python
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

```python
# 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 ✓"