ChatKit Development Guide

Build production AI chat applications with OpenAI ChatKit.

Before You Start

Clarify these before proceeding:

Skip clarification if: User says “just get it working” – use defaults.

Version Compatibility

Breaking change risk: ChatKit is early-stage. Pin versions in production.

Next.js 16 verified patterns:

• Route Handlers use standard Web API Response with response.body for streaming
• Use request.text() to read request body
• Use Cache-Control: no-store for SSE responses

Architecture

Frontend (@openai/chatkit-react)  <--->  FastAPI + ChatKitServer  <--->  Agent (LLM)
              |                                    |
         domainKey                              Store (DB)

Fastest Path (60 seconds)

# Create project with script
python scripts/init_chatkit_project.py my-app --provider openai

# Add API key
echo "OPENAI_API_KEY=sk-xxx" > my-app/.env

# Run server
cd my-app && uv run uvicorn main:app --reload --port 8000

Verify it works:

curl -X POST http://localhost:8000/chatkit \
  -H "Content-Type: application/json" \
  -d '{"type":"new_thread"}'
# Should return: {"thread_id": "thread_xxx", ...}

Required Dependencies

# Backend (Python)
uv add openai-chatkit openai-agents fastapi "uvicorn[standard]"

# For non-OpenAI models (Gemini, Claude, etc.)
uv add "openai-agents[litellm]"

# For PostgreSQL persistence
uv add asyncpg

# Frontend (JavaScript/React)
npm install @openai/chatkit-react
# or vanilla JS
npm install @openai/chatkit

Minimal Working Server

from datetime import datetime
from typing import AsyncIterator
from collections import defaultdict

from fastapi import FastAPI, Request
from fastapi.responses import Response, StreamingResponse
from fastapi.middleware.cors import CORSMiddleware

from agents import Agent, Runner
from chatkit.server import ChatKitServer, StreamingResult
from chatkit.store import Store, NotFoundError
from chatkit.types import (
    ThreadMetadata, ThreadItem, Page, UserMessageItem,
    ThreadStreamEvent, ThreadItemDoneEvent, AssistantMessageItem, AssistantMessageContent
)
from chatkit.agents import AgentContext, stream_agent_response, simple_to_agent_input

app = FastAPI()

# CORS required for frontend
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # Restrict in production
    allow_methods=["*"],
    allow_headers=["*"],
)

class InMemoryStore(Store[dict]):
    """Development store - data lost on restart."""
    def __init__(self):
        self.threads, self.items = {}, defaultdict(list)

    async def load_thread(self, thread_id, context):
        if thread_id not in self.threads: raise NotFoundError(f"Thread {thread_id} not found")
        return self.threads[thread_id]
    async def save_thread(self, thread, context): self.threads[thread.id] = thread
    async def delete_thread(self, thread_id, context): self.threads.pop(thread_id, None); self.items.pop(thread_id, None)
    async def load_threads(self, limit, after, order, context):
        return Page(data=list(self.threads.values())[:limit], has_more=False, after=None)
    async def load_thread_items(self, thread_id, after, limit, order, context):
        return Page(data=self.items.get(thread_id, [])[:limit], has_more=False, after=None)
    async def add_thread_item(self, thread_id, item, context): self.items[thread_id].append(item)
    async def save_item(self, thread_id, item, context): pass
    async def load_item(self, thread_id, item_id, context): raise NotFoundError("")
    async def delete_thread_item(self, thread_id, item_id, context): pass
    async def save_attachment(self, attachment, context): raise NotImplementedError()
    async def load_attachment(self, attachment_id, context): raise NotImplementedError()
    async def delete_attachment(self, attachment_id, context): raise NotImplementedError()

class MyChatKitServer(ChatKitServer[dict]):
    agent = Agent(model="gpt-4.1", name="Assistant", instructions="You are helpful.")

    async def respond(self, thread: ThreadMetadata, input: UserMessageItem | None, context: dict) -> AsyncIterator[ThreadStreamEvent]:
        # Load recent conversation history (REQUIRED for multi-turn)
        items_page = await self.store.load_thread_items(thread.id, after=None, limit=20, order="asc", context=context)
        input_items = await simple_to_agent_input(items_page.data)

        agent_context = AgentContext(thread=thread, store=self.store, request_context=context)
        result = Runner.run_streamed(self.agent, input_items, context=agent_context)
        async for event in stream_agent_response(agent_context, result):
            yield event

store = InMemoryStore()
server = MyChatKitServer(store=store)

@app.post("/chatkit")
async def chatkit_endpoint(request: Request):
    result = await server.process(await request.body(), context={})
    if isinstance(result, StreamingResult):
        return StreamingResponse(result, media_type="text/event-stream")
    return Response(content=result.json, media_type="application/json")

Run: uv run uvicorn main:app --reload

Minimal Frontend (React)

import { ChatKit, useChatKit } from '@openai/chatkit-react';

export function Chat() {
  const { control } = useChatKit({
    api: {
      url: 'http://localhost:8000/chatkit',
      domainKey: 'local-dev',  // Required - use 'local-dev' for development
    },
  });

  return <ChatKit control={control} className="h-[600px] w-[400px]" />;
}

See frontend.md for complete frontend integration.

What Can Go Wrong

Full troubleshooting: troubleshooting.md

Choose Your Path

Store Selection

See store.md for implementation details.

LLM Provider

Adding Tools

from agents import function_tool

@function_tool
def search_products(query: str) -> str:
    """Search the product catalog."""
    return f"Found results for: {query}"

agent = Agent(tools=[search_products], ...)

See mcp.md for MCP servers and hosted tools.

Multi-User Sessions

from dataclasses import dataclass

@dataclass
class RequestContext:
    user_id: str
    thread_id: str | None = None

# Filter by user_id in all store queries
async def load_threads(self, limit, after, order, ctx: RequestContext):
    # WHERE user_id = ctx.user_id
    ...

See sessions.md for complete implementation.

Scripts

Providers: openai, gemini, openrouter, ollama

How Do I Know I’m Done?

Minimum Viable ChatKit

• Server starts without errors (uv run uvicorn main:app)
• curl -X POST .../chatkit -d '{"type":"new_thread"}' returns thread_id
• Frontend renders ChatKit component (check: element visible, no console errors)
• Sending a message returns a streaming response
• Multi-turn conversation works (agent remembers previous messages)

Production Ready

• PostgreSQL store connected and tables created
• CORS restricted to production domain
• API key in environment variable (not hardcoded)
• User authentication extracts user_id into RequestContext
• Store queries filter by user_id
• Error handling returns graceful messages
• domainKey registered with OpenAI (not ‘local-dev’)

References

External Documentation

• ChatKit Python: https://openai.github.io/chatkit-python/
• ChatKit JS: https://openai.github.io/chatkit-js/
• OpenAI Agents SDK: https://openai.github.io/openai-agents-python/