GitVita
New repositoryImport repositoryNew issue
Your repositories Your dashboard Sign in

openai/chatkit-python

Public

mirrored fromhttps://github.com/openai/chatkit-pythonAvailable

Watch0Fork0Star0
CodeCommitsIssuesPull requestsActionsInsightsSecurity
49b49f9bcc3b92f40a314377de2530d99784a48e

Branches

Tags

  • No tags available.
0Branches0Tags
Go to file
Add file
Code

Clone

HTTPS

Download ZIP

chatkit-python/docs

docs/quickstart.md

261lines · modecode

RawDownload
Latest commit unavailable.
1# Quick start
2
3To get a basic ChatKit app running—a React chat UI talking to a Python server—clone and run the starter app:
4
5
6```sh
7git clone https://github.com/openai/openai-chatkit-starter-app.git
8cd openai-chatkit-starter-app/chatkit
9npm install
10npm run dev
11```
12
13The sections below explain the core components and steps behind the starter app.
14
15## Render chat UI
16
17!!! note "Quick start uses React"
18 This section shows the React integration using `@openai/chatkit-react`.
19 If you’re not using React, you can render ChatKit directly with vanilla JavaScript using `@openai/chatkit`. See the [ChatKit.js quick start](https://openai.github.io/chatkit-js/quickstart/) for details.
20
21Install the React bindings:
22
23```sh
24npm install @openai/chatkit-react
25```
26
27In your index.html, load ChatKit.js:
28
29```html
30<!doctype html>
31<html lang="en">
32 <head>
33 <meta charset="UTF-8" />
34 <script src="https://cdn.platform.openai.com/deployments/chatkit/chatkit.js"></script>
35 </head>
36 <body>
37 <div id="root"></div>
38 </body>
39</html>
40```
41
42Wire up a minimal React app. Point `api.url` at your ChatKit server endpoint and pass the domain key you configured there.
43
44```tsx
45import {ChatKit, useChatKit} from "@openai/chatkit-react";
46
47export function App() {
48 const chatkit = useChatKit({
49 api: {
50 url: "http://localhost:8000/chatkit",
51 domainKey: "local-dev", // domain keys are optional in dev
52 },
53 });
54
55 return <ChatKit control={chatkit.control} />;
56}
57```
58
59The chat UI will render, but sending messages will fail until you start the server below and provide a store for threads and messages.
60
61## Run your ChatKit server
62
63Install the ChatKit Python package and expose a single `/chatkit` endpoint that forwards requests to a `ChatKitServer` instance.
64
65```sh
66pip install openai-chatkit
67```
68
69Create `main.py` with a minimal server that is hard-coded to always reply “Hello, world!”—you'll replace this with an actual call to a model in [Respond to a user message](guides/respond-to-user-message.md).
70
71```python
72# Other imports omitted for brevity; see the starter repo for a runnable file with all imports.
73from chatkit.server import ChatKitServer
74
75app = FastAPI()
76
77
78class MyChatKitServer(ChatKitServer[dict]):
79 async def respond(
80 self,
81 thread: ThreadMetadata,
82 input_user_message: UserMessageItem | None,
83 context: dict,
84 ) -> AsyncIterator[ThreadStreamEvent]:
85 # Streams a fixed "Hello, world!" assistant message
86 yield ThreadItemDoneEvent(
87 item=AssistantMessageItem(
88 thread_id=thread.id,
89 id=self.store.generate_item_id("message", thread, context),
90 created_at=datetime.now(),
91 content=[AssistantMessageContent(text="Hello, world!")],
92 ),
93 )
94
95# Create your server by passing a store implementation.
96# MyChatKitStore is defined in the next section.
97server = MyChatKitServer(store=MyChatKitStore())
98
99
100@app.post("/chatkit")
101async def chatkit(request: Request):
102 result = await server.process(await request.body(), context={})
103 if isinstance(result, StreamingResult):
104 return StreamingResponse(result, media_type="text/event-stream")
105 return Response(content=result.json, media_type="application/json")
106```
107
108All ChatKit requests go to this single endpoint. Set `api.url` on the React side to match (`/chatkit` here), and `ChatKitServer` routes each request internally.
109
110
111## Store chat data
112
113ChatKit servers require a store to load and save threads, messages, and other items.
114
115For this quickstart, use a small in-memory store so conversations persist while the process is running, without introducing a database. This keeps the example minimal while still matching real ChatKit behavior.
116
117
118```python
119from collections import defaultdict
120from chatkit.store import NotFoundError, Store
121from chatkit.types import Attachment, Page, ThreadItem, ThreadMetadata
122
123
124class MyChatKitStore(Store[dict]):
125 def __init__(self):
126 self.threads: dict[str, ThreadMetadata] = {}
127 self.items: dict[str, list[ThreadItem]] = defaultdict(list)
128
129 async def load_thread(self, thread_id: str, context: dict) -> ThreadMetadata:
130 if thread_id not in self.threads:
131 raise NotFoundError(f"Thread {thread_id} not found")
132 return self.threads[thread_id]
133
134 async def save_thread(self, thread: ThreadMetadata, context: dict) -> None:
135 self.threads[thread.id] = thread
136
137 async def load_threads(
138 self, limit: int, after: str | None, order: str, context: dict
139 ) -> Page[ThreadMetadata]:
140 threads = list(self.threads.values())
141 return self._paginate(
142 threads, after, limit, order, sort_key=lambda t: t.created_at, cursor_key=lambda t: t.id
143 )
144
145 async def load_thread_items(
146 self, thread_id: str, after: str | None, limit: int, order: str, context: dict
147 ) -> Page[ThreadItem]:
148 items = self.items.get(thread_id, [])
149 return self._paginate(
150 items, after, limit, order, sort_key=lambda i: i.created_at, cursor_key=lambda i: i.id
151 )
152
153 async def add_thread_item(
154 self, thread_id: str, item: ThreadItem, context: dict
155 ) -> None:
156 self.items[thread_id].append(item)
157
158 async def save_item(
159 self, thread_id: str, item: ThreadItem, context: dict
160 ) -> None:
161 items = self.items[thread_id]
162 for idx, existing in enumerate(items):
163 if existing.id == item.id:
164 items[idx] = item
165 return
166 items.append(item)
167
168 async def load_item(
169 self, thread_id: str, item_id: str, context: dict
170 ) -> ThreadItem:
171 for item in self.items.get(thread_id, []):
172 if item.id == item_id:
173 return item
174 raise NotFoundError(f"Item {item_id} not found in thread {thread_id}")
175
176 async def delete_thread(self, thread_id: str, context: dict) -> None:
177 self.threads.pop(thread_id, None)
178 self.items.pop(thread_id, None)
179
180 async def delete_thread_item(
181 self, thread_id: str, item_id: str, context: dict
182 ) -> None:
183 self.items[thread_id] = [
184 item for item in self.items.get(thread_id, []) if item.id != item_id
185 ]
186
187 def _paginate(self, rows: list, after: str | None, limit: int, order: str, sort_key, cursor_key):
188 sorted_rows = sorted(rows, key=sort_key, reverse=order == "desc")
189 start = 0
190 if after:
191 for idx, row in enumerate(sorted_rows):
192 if cursor_key(row) == after:
193 start = idx + 1
194 break
195 data = sorted_rows[start : start + limit]
196 has_more = start + limit < len(sorted_rows)
197 next_after = cursor_key(data[-1]) if has_more and data else None
198 return Page(data=data, has_more=has_more, after=next_after)
199
200 # Attachments are intentionally not implemented for the quickstart
201
202 async def save_attachment(
203 self, attachment: Attachment, context: dict
204 ) -> None:
205 raise NotImplementedError()
206
207 async def load_attachment(
208 self, attachment_id: str, context: dict
209 ) -> Attachment:
210 raise NotImplementedError()
211
212 async def delete_attachment(self, attachment_id: str, context: dict) -> None:
213 raise NotImplementedError()
214
215```
216
217This store implements only the methods required for basic chat while the server is running; persistence across restarts and attachments are intentionally omitted.
218
219For production, replace this with a database-backed store (for example, Postgres or MySQL) so threads and items persist across restarts.
220
221
222## Generate model responses
223
224Replace the hardcoded "Hello, World!" reply from [Run your ChatKit server](#run-your-chatkit-server) with an Agents SDK call to generate real responses. Set `OPENAI_API_KEY` in your environment before running.
225
226Use ChatKit's Agents SDK helpers to simplify request conversion and streaming. The `simple_to_agent_input` helper translates ChatKit thread items to agent input items, and `stream_agent_response` turns the streamed run into ChatKit events:
227
228
229```python
230from agents import Agent, Runner
231from chatkit.agents import AgentContext, simple_to_agent_input, stream_agent_response
232
233assistant = Agent(
234 name="assistant",
235 instructions="You are a helpful assistant.",
236 model="gpt-4.1-mini",
237)
238
239class MyChatKitServer(ChatKitServer[dict]):
240 async def respond(
241 self,
242 thread: ThreadMetadata,
243 input_user_message: UserMessageItem | None,
244 context: dict,
245 ) -> AsyncIterator[ThreadStreamEvent]:
246 # Convert recent thread items (which includes the user message) to model input
247 items_page = await self.store.load_thread_items(
248 thread.id,
249 after=None,
250 limit=20,
251 order="asc",
252 context=context,
253 )
254 input_items = await simple_to_agent_input(items_page.data)
255
256 # Stream the run through ChatKit events
257 agent_context = AgentContext(thread=thread, store=self.store, request_context=context)
258 result = Runner.run_streamed(assistant, input_items, context=agent_context)
259 async for event in stream_agent_response(agent_context, result):
260 yield event
261```
262