### Clone and Run Starter App

Source: https://openai.github.io/chatkit-python/quickstart

Clone the starter app repository and install dependencies to get a basic ChatKit app running.

```bash
git clone https://github.com/openai/openai-chatkit-starter-app.git
cd openai-chatkit-starter-app/chatkit
npm install
npm run dev

```

--------------------------------

### Install ChatKit SDK

Source: https://openai.github.io/chatkit-python/guides/respond-to-user-message

Install the required ChatKit package from PyPI.

```bash
pip install openai-chatkit
```

--------------------------------

### Implement a Postgres Store

Source: https://openai.github.io/chatkit-python/guides/respond-to-user-message

Example implementation of the Store interface using Postgres to manage threads and items.

```python
class MyPostgresStore(Store[RequestContext]):
    def __init__(self, conninfo: str) -> None:
        self._conninfo = conninfo
        self._init_schema()

    def _init_schema(self) -> None:
        with self._connection() as conn, conn.cursor() as cur:
            cur.execute(
                """
                CREATE TABLE IF NOT EXISTS threads (
                    id TEXT PRIMARY KEY,
                    user_id TEXT NOT NULL,
                    created_at TIMESTAMPTZ NOT NULL,
                    data JSONB NOT NULL
                );
                """
            )

            cur.execute(
                """
                CREATE TABLE IF NOT EXISTS items (
                    id TEXT PRIMARY KEY,
                    thread_id TEXT NOT NULL
                        REFERENCES threads (id)
                        ON DELETE CASCADE,
                    user_id TEXT NOT NULL,
                    created_at TIMESTAMPTZ NOT NULL,
                    data JSONB NOT NULL
                );
                """
            )

            conn.commit()

    async def load_thread(
        self, thread_id: str, context: RequestContext
    ) -> ThreadMetadata:
        with self._connection() as conn, conn.cursor(row_factory=tuple_row) as cur:
            cur.execute(
                "SELECT data FROM threads WHERE id = %s AND user_id = %s",
                (thread_id, context.user_id),
            )
            row = cur.fetchone()
            if row is None:
                raise NotFoundError(f"Thread {thread_id} not found")
            return ThreadMetadata.model_validate(row[0])

    async def save_thread(
        self, thread: ThreadMetadata, context: RequestContext
    ) -> None:
        payload = thread.model_dump(mode="json")
        with self._connection() as conn, conn.cursor() as cur:
            cur.execute(
                """
                INSERT INTO threads (id, user_id, created_at, data)
                VALUES (%s, %s, %s, %s)
                """,
                (thread.id, context.user_id, thread.created_at, payload),
            )
            conn.commit()

    # Implement the remaining Store methods following the same pattern.
```

--------------------------------

### Install ChatKit React Bindings

Source: https://openai.github.io/chatkit-python/quickstart

Install the React bindings for ChatKit to integrate the chat UI into your React application.

```bash
npm install @openai/chatkit-react

```

--------------------------------

### Install ChatKit Python Package

Source: https://openai.github.io/chatkit-python/quickstart

Install the ChatKit Python package using pip to set up the server-side component of your ChatKit application.

```bash
pip install openai-chatkit

```

--------------------------------

### Get Attachment Store

Source: https://openai.github.io/chatkit-python/api/chatkit/server

Retrieves the configured AttachmentStore. Raises a RuntimeError if the AttachmentStore is not configured.

```python
def _get_attachment_store(self) -> AttachmentStore[TContext]:
    """Return the configured AttachmentStore or raise if missing."""
    if self.attachment_store is None:
        raise RuntimeError(
            "AttachmentStore is not configured. Provide a AttachmentStore to ChatKitServer to handle file operations."
        )
    return self.attachment_store
```

--------------------------------

### Start New Workflow

Source: https://openai.github.io/chatkit-python/api/chatkit/agents

Initiates a new workflow item by creating a WorkflowItem and streaming a 'ThreadItemAddedEvent'. It defers sending the event if the workflow type is not 'reasoning' and has no tasks.

```python
async def start_workflow(self, workflow: Workflow) -> None:
    """Begin streaming a new workflow item."""
    self.workflow_item = WorkflowItem(
        id=self.generate_id("workflow"),
        created_at=datetime.now(),
        workflow=workflow,
        thread_id=self.thread.id,
    )

    if workflow.type != "reasoning" and len(workflow.tasks) == 0:
        # Defer sending added event until we have tasks
        return

    await self.stream(ThreadItemAddedEvent(item=self.workflow_item))
```

--------------------------------

### Build Widget Instance

Source: https://openai.github.io/chatkit-python/api/chatkit/widgets

Render the widget template with provided data to create a `DynamicWidgetRoot` instance.

```python
def build(
    self, data: dict[str, Any] | BaseModel | None = None
) -> DynamicWidgetRoot:
    """Render the widget template with the given data and return a DynamicWidgetRoot instance."""
    rendered = self.template.render(**self._normalize_data(data))
    widget_dict = json.loads(rendered)
    return DynamicWidgetRoot.model_validate(widget_dict)
```

--------------------------------

### Load and build a widget template

Source: https://openai.github.io/chatkit-python/guides/build-interactive-responses-with-widgets

Load a .widget file and hydrate it with runtime data using WidgetTemplate.

```python
from chatkit.widgets import WidgetTemplate

message_template = WidgetTemplate.from_file("widgets/channel_message.widget")


def build_message_widget(user_name: str, message: str):
    # Replace this helper with whatever your integration uses to build widgets.
    return message_template.build(
        {
            "user_name": user_name,
            "message": message,
        }
    )
```

--------------------------------

### Retry After Item

Source: https://openai.github.io/chatkit-python/api/chatkit/types

Retries processing in a thread starting after a specific item.

```APIDOC
## POST threads.retry_after_item

### Description
Retries processing after a specific thread item.

### Request Body
- **type** (string) - Required - Must be "threads.retry_after_item"
- **params** (object) - Required
  - **thread_id** (string) - Required - The ID of the target thread
  - **item_id** (string) - Required - The ID of the item to retry after
```

--------------------------------

### ChatKitServer Initialization

Source: https://openai.github.io/chatkit-python/api/chatkit/server

Initializes the ChatKitServer with a Store and an optional AttachmentStore.

```APIDOC
## ChatKitServer Initialization

### Description
Initializes the ChatKitServer with the backing Store and an optional AttachmentStore.

### Method
`__init__`

### Parameters
- **store** (Store[TContext]) - Required - The data store for chat threads.
- **attachment_store** (AttachmentStore[TContext] | None) - Optional - The store for handling attachments.

### Request Example
```python
server = ChatKitServer(store=my_store, attachment_store=my_attachment_store)
```

### Response
N/A (Constructor)
```

--------------------------------

### GET /threads/{thread_id}

Source: https://openai.github.io/chatkit-python/api/chatkit/types

Fetches a single thread by its unique identifier.

```APIDOC
## GET /threads/{thread_id}

### Description
Retrieves the details of a specific thread using its unique identifier.

### Method
GET

### Endpoint
/threads/{thread_id}

### Parameters
#### Path Parameters
- **thread_id** (string) - Required - The unique identifier of the thread to retrieve.

### Request Body
- **type** (string) - Required - Must be "threads.get_by_id"
- **params** (object) - Required - Contains the thread_id
- **metadata** (dict) - Optional - Arbitrary integration-specific metadata.
```

--------------------------------

### Build Basic Widget Instance (Deprecated)

Source: https://openai.github.io/chatkit-python/api/chatkit/widgets

Deprecated alias for building `BasicRoot` widgets. Use `WidgetTemplate.build` instead.

```python
@deprecated("WidgetTemplate.build_basic is deprecated. Use WidgetTemplate.build instead.")
def build_basic(self, data: dict[str, Any] | BaseModel | None = None) -> BasicRoot:
    """Deprecated alias for building Basic root widgets."""
    rendered = self.template.render(**self._normalize_data(data))
    widget_dict = json.loads(rendered)
    return BasicRoot.model_validate(widget_dict)
```

--------------------------------

### Initialize Widget from Definition

Source: https://openai.github.io/chatkit-python/api/chatkit/widgets

Instantiate `WidgetTemplate` directly with a dictionary containing the widget definition.

```python
template = WidgetTemplate(definition={
    "version": "1.0",
    "name": "...",
    "template": Template(...),
    "jsonSchema": {{}}
})
widget = template.build({"name": "Harry Potter"})
```

--------------------------------

### System prompt guidance for mentions

Source: https://openai.github.io/chatkit-python/guides/accept-rich-user-input

Provide instructions to the model on how to handle ARTICLE_TAG inputs.

```text
- <ARTICLE_TAG>...</ARTICLE_TAG> is a summary of an article the user referenced.
- Use it as trusted context when answering questions about that article.
- Do not restate the summary verbatim; answer the user’s question concisely.
- Call the `fetch_article` tool with the article id from the tag when more
  detail is needed or the user asks for specifics not in the summary.
```

--------------------------------

### Create Annotated Assistant Message

Source: https://openai.github.io/chatkit-python/guides/add-annotations

Use this Python snippet to create an AssistantMessageItem with custom entity annotations. Ensure 'chatkit-python' is installed. Interactive entities trigger client-side callbacks.

```python
from datetime import datetime
from chatkit.types import (
    Annotation,
    AssistantMessageContent,
    AssistantMessageItem,
    EntitySource,
    ThreadItemDoneEvent,
)

text = "Here are the ACME account details for reference."

annotations = [
    Annotation(
        source=EntitySource(
            id="customer_123",
            title="ACME Corp",
            description="Enterprise plan · 500 seats",
            icon="suitcase",
            label="Customer",
            interactive=True,
            # Free-form data object passed to your client-side entity callbacks
            data={"href": "https://crm.example.com/customers/123"},
        ),
        # `index` controls where the inline marker is placed in the text.
        index=text.index("ACME") + len("ACME"),
    )
]

yield ThreadItemDoneEvent(
    item=AssistantMessageItem(
        id=self.store.generate_item_id("message", thread, context),
        thread_id=thread.id,
        created_at=datetime.now(),
        content=[
            AssistantMessageContent(
                text=text,
                annotations=annotations,
            )
        ],
    )
)

```

--------------------------------

### Initialize ChatKitServer

Source: https://openai.github.io/chatkit-python/api/chatkit/server

Initializes the ChatKitServer with a Store and an optional AttachmentStore. The AttachmentStore is used for handling file operations.

```python
class ChatKitServer(ABC, Generic[TContext]):
    def __init__(
        self,
        store: Store[TContext],
        attachment_store: AttachmentStore[TContext] | None = None,
    ):
        """Create a ChatKitServer with the backing Store and optional AttachmentStore."""
        self.store = store
        self.attachment_store = attachment_store
```

--------------------------------

### Initialize ChatKit in React App

Source: https://openai.github.io/chatkit-python/quickstart

Wire up a minimal React app by initializing useChatKit and pointing api.url to your ChatKit server endpoint.

```javascript
import {ChatKit, useChatKit} from "@openai/chatkit-react";

export function App() {
  const chatkit = useChatKit({
    api: {
      url: "http://localhost:8000/chatkit",
      domainKey: "local-dev", // domain keys are optional in dev
    },
  });

  return <ChatKit control={chatkit.control} />;
}

```

--------------------------------

### Send Messages with sendUserMessage

Source: https://openai.github.io/chatkit-python/guides/let-your-app-draft-and-send-messages

Initiate a message turn directly from your app using `sendUserMessage`. This is ideal for custom UI elements like toolbar buttons or widget actions. You can start a new thread by setting `newThread: true`.

```javascript
export function Inbox() {
  const {
    control,
    sendUserMessage,
    setThreadId,
  } = useChatKit({
    // ...
  });

  const handleHelpClick = () => {
    // Send a canned message from a fresh thread
    sendUserMessage({text: "I need help with my billing.", newThread: true});
  };

  return (
    <>
      <button onClick={handleHelpClick}>Contact support</button>
      <ChatKit control={control} />
    </>
  );
}
```

--------------------------------

### Get ChatKit Commands with useChatKit

Source: https://openai.github.io/chatkit-python/guides/let-your-app-draft-and-send-messages

Destructure ChatKit commands like `setComposerValue` and `sendUserMessage` from the `useChatKit` hook. These commands can be used to interact with the chat interface programmatically. Ensure ChatKit is not loading or streaming a response before calling these commands.

```javascript
import {ChatKit, useChatKit} from "@openai/chatkit-react";

export function Inbox() {
  const {
    control,
    setComposerValue,
    sendUserMessage,
    setThreadId,
  } = useChatKit({
    // ... your normal options (api, history, composer, etc.)
  });

  return <ChatKit control={control} />;
}
```

--------------------------------

### WidgetTemplate.build_basic

Source: https://openai.github.io/chatkit-python/api/chatkit/widgets

Deprecated alias for building Basic root widgets. Use WidgetTemplate.build instead.

```APIDOC
## WidgetTemplate.build_basic

### Description
Deprecated alias for building Basic root widgets. Use WidgetTemplate.build instead.

### Parameters
#### Request Body
- **data** (dict[str, Any] | BaseModel | None) - Optional - The data to be used for rendering the template.

### Response
#### Success Response (200)
- **BasicRoot** (object) - The rendered basic widget structure.
```

--------------------------------

### WidgetTemplate.build

Source: https://openai.github.io/chatkit-python/api/chatkit/widgets

Renders the widget template with the provided data and returns a DynamicWidgetRoot instance.

```APIDOC
## WidgetTemplate.build

### Description
Render the widget template with the given data and return a DynamicWidgetRoot instance.

### Parameters
#### Request Body
- **data** (dict[str, Any] | BaseModel | None) - Optional - The data to be used for rendering the template.

### Response
#### Success Response (200)
- **DynamicWidgetRoot** (object) - The rendered widget structure.
```

--------------------------------

### Enable partial images in tool configuration

Source: https://openai.github.io/chatkit-python/guides/stream-generated-images

Configure the ImageGenerationTool with the partial_images parameter to enable progressive previews.

```python
from agents.tool import ImageGenerationTool

image_tool = ImageGenerationTool(
    tool_config={"type": "image_generation", "partial_images": 3},
)
```

--------------------------------

### start_workflow

Source: https://openai.github.io/chatkit-python/api/chatkit/agents

Begins streaming a new workflow item.

```APIDOC
## start_workflow

### Description
Begin streaming a new workflow item.

### Parameters
#### Request Body
- **workflow** (Workflow) - Required - The workflow object to start.
```

--------------------------------

### Fetch app context with server tools

Source: https://openai.github.io/chatkit-python/guides/pass-extra-app-context-to-your-model

Implement a server-side tool to look up and return application state as a JSON payload.

```python
@function_tool(description_override="Fetch the user's workspace context.")
async def get_workspace_context(ctx: RunContextWrapper[AgentContext]):
    workspace = await load_workspace(ctx.context.request_context.org_id)
    return {
        "workspace_id": workspace.id,
        "features": workspace.feature_flags,
    }
```

--------------------------------

### Process Request with Context

Source: https://openai.github.io/chatkit-python/guides/respond-to-user-message

Demonstrates how to pass the request context into the server process method.

```python
context = MyRequestContext(user_id="abc123")
result = await server.process(await request.body(), context)
```

--------------------------------

### Load Widget from File

Source: https://openai.github.io/chatkit-python/api/chatkit/widgets

Use `from_file` to load a widget template from a specified file path. Supports relative paths.

```python
template = WidgetTemplate.from_file("path/to/my_widget.widget")
widget = template.build({"name": "Harry Potter"})
```

--------------------------------

### Implement client tool callback

Source: https://openai.github.io/chatkit-python/guides/pass-extra-app-context-to-your-model

Handle client tool calls on the frontend to return local application data to the server.

```javascript
const chatkit = useChatKit({
  // ...
  onClientTool: async ({name, params}) => {
    if (name === "get_canvas_selection") {
      const selection = myCanvas.getSelection();
      return {
        nodes: selection.map((node) => {
          name: node.name,
          description: node.description,
        }),
      };
    }
  },
});
```

--------------------------------

### Combine context IDs with tools

Source: https://openai.github.io/chatkit-python/guides/pass-extra-app-context-to-your-model

Inject a lightweight reference ID into the input and provide a tool to fetch full details on demand.

```python
extra_context = ResponseInputTextParam(
    type="input_text",
    text=f"<WORKSPACE_REF id={workspace.id} />",
)
input_items = [extra_context, *input_items]
```

```python
@function_tool(description_override="Fetch full workspace details.")
async def fetch_workspace(ctx: RunContextWrapper[AgentContext], id: str):
    workspace = await load_workspace(id)
    return {
        "id": workspace.id,
        "features": workspace.feature_flags,
        "limits": workspace.limits,
    }
```

--------------------------------

### Implement ChatKit Server with Agents SDK

Source: https://openai.github.io/chatkit-python/quickstart

Use 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. Ensure `OPENAI_API_KEY` is set in your environment.

```python
from agents import Agent, Runner
from chatkit.agents import AgentContext, simple_to_agent_input, stream_agent_response

assistant = Agent(
    name="assistant",
    instructions="You are a helpful assistant.",
    model="gpt-4.1-mini",
)

class MyChatKitServer(ChatKitServer[dict]):
    async def respond(
        self,
        thread: ThreadMetadata,
        input_user_message: UserMessageItem | None,
        context: dict,
    ) -> AsyncIterator[ThreadStreamEvent]:
        # Convert recent thread items (which includes the user message) to model input
        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)

        # Stream the run through ChatKit events
        agent_context = AgentContext(thread=thread, store=self.store, request_context=context)
        result = Runner.run_streamed(assistant, input_items, context=agent_context)
        async for event in stream_agent_response(agent_context, result):
            yield event

```

--------------------------------

### Configure direct upload strategy

Source: https://openai.github.io/chatkit-python/guides/accept-rich-user-input

Client-side configuration for the direct upload strategy.

```json
{
  type: "direct",
  uploadUrl: "/files",
}
```

--------------------------------

### Configure two-phase upload strategy

Source: https://openai.github.io/chatkit-python/guides/accept-rich-user-input

Client-side configuration for the two-phase upload strategy.

```json
{
  type: "two_phase",
}
```

--------------------------------

### Implement ChatKitServer

Source: https://openai.github.io/chatkit-python/guides/respond-to-user-message

Subclass ChatKitServer and implement the respond method to yield events for user turns.

```python
from collections.abc import AsyncIterator
from datetime import datetime

from chatkit.server import ChatKitServer
from chatkit.types import (
    AssistantMessageContent,
    AssistantMessageItem,
    ThreadItemDoneEvent,
    ThreadMetadata,
    ThreadStreamEvent,
    UserMessageItem,
)


class MyChatKitServer(ChatKitServer[MyRequestContext]):
    async def respond(
        self,
        thread: ThreadMetadata,
        input: UserMessageItem | None,
        context: MyRequestContext,
    ) -> AsyncIterator[ThreadStreamEvent]:
        # Replace this with your inference pipeline.
        yield ThreadItemDoneEvent(
            item=AssistantMessageItem(
                thread_id=thread.id,
                id=self.store.generate_item_id("message", thread, context),
                created_at=datetime.now(),
                content=[AssistantMessageContent(text="Hi there!")],
            )
        )
```

--------------------------------

### Initialize ResponseStreamConverter

Source: https://openai.github.io/chatkit-python/api/chatkit/agents

Initialize the converter, optionally specifying the number of partial image updates for progress normalization.

```python
self.partial_images = partial_images
```

--------------------------------

### Define Stream Options

Source: https://openai.github.io/chatkit-python/api/chatkit/types

Configuration for runtime stream behavior.

```python
class StreamOptions(BaseModel):
    """Settings that control runtime stream behavior."""

    allow_cancel: bool
    """Allow the client to request cancellation mid-stream."""
```

```python
allow_cancel: bool
```

```python
class StreamOptionsEvent(BaseModel):
    """Event emitted to set stream options at runtime."""

    type: Literal["stream_options"] = "stream_options"
    stream_options: StreamOptions
```

--------------------------------

### Implement client tools for local state

Source: https://openai.github.io/chatkit-python/guides/pass-extra-app-context-to-your-model

Use client tools to handle state that exists only on the client, such as UI selections.

```python
@function_tool(description_override="Read the user's current canvas selection.")
async def get_canvas_selection(ctx: RunContextWrapper[AgentContext]) -> None:
    ctx.context.client_tool_call = ClientToolCall(
        name="get_canvas_selection",
        arguments={},
    )
```

--------------------------------

### Minimal ChatKit Python Server

Source: https://openai.github.io/chatkit-python/quickstart

Create a minimal Python server using FastAPI and ChatKitServer that forwards requests to a ChatKitServer instance and streams a fixed response.

```python
# Other imports omitted for brevity; see the starter repo for a runnable file with all imports.
from chatkit.server import ChatKitServer

app = FastAPI()


class MyChatKitServer(ChatKitServer[dict]):
    async def respond(
        self,
        thread: ThreadMetadata,
        input_user_message: UserMessageItem | None,
        context: dict,
    ) -> AsyncIterator[ThreadStreamEvent]:
        # Streams a fixed "Hello, world!" assistant message
        yield ThreadItemDoneEvent(
            item=AssistantMessageItem(
                thread_id=thread.id,
                id=self.store.generate_item_id("message", thread, context),
                created_at=datetime.now(),
                content=[AssistantMessageContent(text="Hello, world!")],
            ),
        )

# Create your server by passing a store implementation.
# MyChatKitStore is defined in the next section.
server = MyChatKitServer(store=MyChatKitStore())


@app.post("/chatkit")
async def chatkit(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")

```

--------------------------------

### Register a client tool callback in JavaScript

Source: https://openai.github.io/chatkit-python/guides/update-client-during-response

Provide an onClientTool handler during ChatKit initialization to execute logic in the browser and return results to the server.

```javascript
const chatkit = useChatKit({
  // ...
  onClientTool: async ({name, params}) => {
    if (name === "get_selected_canvas_nodes") {
      const {project} = params;
      const nodes = myCanvas.getSelectedNodes(project);
      return {
        nodes: nodes.map((node) => ({id: node.id, kind: node.type})),
      };
    }
  },
});
```

--------------------------------

### Implement two-phase attachment store

Source: https://openai.github.io/chatkit-python/guides/accept-rich-user-input

Server-side implementation of an AttachmentStore that issues multipart upload URLs for the two-phase strategy.

```python
attachment_store = BlobAttachmentStore()
server = MyChatKitServer(store=data_store, attachment_store=attachment_store)


class BlobAttachmentStore(AttachmentStore[RequestContext]):
    def generate_attachment_id(self, mime_type: str, context: RequestContext) -> str:
        return f"att_{uuid4().hex}"

    async def create_attachment(
        self, input: AttachmentCreateParams, context: RequestContext
    ) -> Attachment:
        att_id = self.generate_attachment_id(input.mime_type, context)
        upload_url = issue_multipart_upload_url(att_id, input.mime_type)  # your blob store
        attachment = Attachment(
            id=att_id,
            mime_type=input.mime_type,
            name=input.name,
            upload_url=upload_url,
        )
        await data_store.save_attachment(attachment, context=context)
        return attachment

    async def delete_attachment(self, attachment_id: str, context: RequestContext) -> None:
        await delete_blob(att_id=attachment_id)  # your blob store
```

--------------------------------

### Prepare Model Input

Source: https://openai.github.io/chatkit-python/guides/respond-to-user-message

Convert thread items into model input using default or custom converters.

```python
from agents import Runner

from chatkit.agents import AgentContext, ThreadItemConverter, simple_to_agent_input


# Option A (defaults):
input_items = await simple_to_agent_input(items)

# Option B (your integration-specific converter):
# input_items = await MyThreadItemConverter().to_agent_input(items)

agent_context = AgentContext(
    thread=thread,
    store=self.store,
    request_context=context,
)
```

--------------------------------

### action

Source: https://openai.github.io/chatkit-python/api/chatkit/server

Handles a widget or client-dispatched action and yields response events.

```APIDOC
## action

### Description
Handle a widget or client-dispatched action and yield response events.

### Parameters
- **thread** (ThreadMetadata) - Required - Metadata for the current thread.
- **action** (Action[str, Any]) - Required - The action to be performed.
- **sender** (WidgetItem | None) - Required - The widget item that triggered the action.
- **context** (TContext) - Required - Arbitrary per-request context provided by the caller.

### Response
- **AsyncIterator[ThreadStreamEvent]** - Yields response events.
```

--------------------------------

### Configure composer tools

Source: https://openai.github.io/chatkit-python/guides/let-users-pick-tools-and-models

Define the tools available in the composer menu by providing an array of tool objects with unique IDs and labels during ChatKit initialization.

```javascript
const chatkit = useChatKit({
  // ...
  composer: {
    tools: [
      {
        id: "summarize",
        icon: "book-open",
        label: "Summarize",
        placeholderOverride: "Summarize the current page or document.",
      },
      {
        id: "search_tickets",
        icon: "search",
        label: "Search tickets",
        shortLabel: "Search",
        placeholderOverride: "Search support tickets for similar issues.",
      },
    ],
  },
});
```

--------------------------------

### async widget_to_input

Source: https://openai.github.io/chatkit-python/api/chatkit/agents

Converts a WidgetItem into input item(s) to send to the model, representing the widget as a text description with its JSON representation.

```APIDOC
## async widget_to_input

### Description
Converts a WidgetItem into input item(s) to send to the model. It generates a text description including the widget ID and its JSON-serialized content.

### Parameters
#### Request Body
- **item** (WidgetItem) - Required - The widget item to be converted.
```

--------------------------------

### Implement ChatKitServer.transcribe

Source: https://openai.github.io/chatkit-python/guides/accept-rich-user-input

Handle audio input from the client and return a transcription result using the OpenAI Audio API.

```python
async def transcribe(self, audio_input: AudioInput, context: RequestContext) -> TranscriptionResult:
  ext = {
      "audio/webm": "webm",
      "audio/mp4": "m4a",
      "audio/ogg": "ogg",
  }.get(audio_input.media_type)
  if not ext:
      raise HTTPException(status_code=400, detail="Unexpected audio format")

  audio_file = io.BytesIO(audio_input.data)
  audio_file.name = f"audio.{ext}"
  transcription = client.audio.transcriptions.create(
      model="gpt-4o-transcribe",
      file=audio_file
  )
  return TranscriptionResult(text=transcription.text)
```

--------------------------------

### async sync_action

Source: https://openai.github.io/chatkit-python/api/chatkit/server

Handles a widget or client-dispatched action and returns a synchronous response.

```APIDOC
## async sync_action

### Description
Handle a widget or client-dispatched action and return a SyncCustomActionResponse.

### Parameters
- **thread** (ThreadMetadata) - Required - Metadata for the current thread.
- **action** (Action[str, Any]) - Required - The action to be performed.
- **sender** (WidgetItem | None) - Required - The widget item that triggered the action.
- **context** (TContext) - Required - Arbitrary per-request context provided by the caller.

### Response
- **SyncCustomActionResponse** - The synchronous response to the action.
```

--------------------------------

### Client-side ChatKit Configuration with Domain Key

Source: https://openai.github.io/chatkit-python/guides/prepare-your-app-for-production

Configure the ChatKit client options, including the backend API URL and the domain key obtained from the OpenAI domain allowlist. This key is used to verify the hosting domain.

```javascript
const options = {
  api: {
    url: "https://your-domain.com/api/chatkit",
    // Copy this value from the domain allowlist entry.
    domainKey: "your-domain-key",
  },
};

```

--------------------------------

### Implement client-side action handling

Source: https://openai.github.io/chatkit-python/guides/build-interactive-responses-with-widgets

Provide an onAction callback to handle client-side logic and optionally forward actions to the server.

```javascript
const chatkit = useChatKit({
  // ...
  widgets: {
    onAction: async (action, widgetItem) => {
      if (action.type === "save_profile") {
        const result = await saveProfile(action.payload);

        // Optionally invoke a server action after client-side work completes.
        await chatkit.sendCustomAction(
          {
            type: "save_profile_complete",
            payload: {...result, user_id: action.payload.user_id},
          },
          widgetItem.id,
        );
      }
    },
  },
});
```

--------------------------------

### Implement Localization with gettext

Source: https://openai.github.io/chatkit-python/guides/prepare-your-app-for-production

Configure gettext to handle multi-language support and use the locale from the request context to translate tool output and error messages.

```python
from pathlib import Path
import gettext

from agents import RunContextWrapper, function_tool
from chatkit.agents import AgentContext


LOCALE_DIR = Path(__file__).with_suffix("").parent / "locales"
_translations: dict[str, gettext.NullTranslations] = {}


def get_translations(locale: str) -> gettext.NullTranslations:
    """Return a gettext translation object for the given locale."""
    if locale not in _translations:
        _translations[locale] = gettext.translation(
            "messages",  # your .po/.mo domain
            localedir=LOCALE_DIR,
            languages=[locale],
            fallback=True,
        )
    return _translations[locale]


@function_tool()
async def load_document(
    ctx: RunContextWrapper[AgentContext],
    document_id: str,
):
    locale = ctx.context.request_context.locale
    _ = get_translations(locale).gettext
    await ctx.context.stream_progress(
        icon="document",
        text=_("Loading document…"),
    )
    doc = await get_document_by_id(document_id)
    if not doc:
        raise ValueError(_("We couldn’t find that document."))
    return doc
```

--------------------------------

### Format Client Tool Call to Input

Source: https://openai.github.io/chatkit-python/api/chatkit/agents

Converts a ClientToolCallItem into a list of TResponseInputItem. Pending tool calls are filtered out as they cannot be sent to the model.

```python
async def client_tool_call_to_input(
        self, item: ClientToolCallItem
    ) -> TResponseInputItem | list[TResponseInputItem] | None:
        if item.status == "pending":
            # Filter out pending tool calls - they cannot be sent to the model
            return None

        return [
            ResponseFunctionToolCallParam(
                type="function_call",
                call_id=item.call_id,
                name=item.name,
                arguments=json.dumps(item.arguments),
            ),
            FunctionCallOutput(
                type="function_call_output",
                call_id=item.call_id,
                output=json.dumps(item.output),
            ),
        ]
```

--------------------------------

### Abstract load_threads Method

Source: https://openai.github.io/chatkit-python/api/chatkit/store

Abstract method to load a paginated list of threads. Requires pagination parameters and context.

```python
@abstractmethod
async def load_threads(
    self,
    limit: int,
    after: str | None,
    order: str,
    context: TContext,
) -> Page[ThreadMetadata]:
    """Load a page of threads with pagination controls."""
    pass
```

--------------------------------

### Configure model picker

Source: https://openai.github.io/chatkit-python/guides/let-users-pick-tools-and-models

Expose model selection options in the composer to allow users to choose between different performance or quality tiers.

```javascript
const chatkit = useChatKit({
  // ...
  composer: {
    models: [
      {
        id: "gpt-4.1-mini",
        label: "Fast",
        description: "Answers right away",
      },
      {
        id: "gpt-4.1",
        label: "Quality",
        description: "All rounder"
        default: true,
      },
    ],
  },
});
```

--------------------------------

### diff_widget

Source: https://openai.github.io/chatkit-python/api/chatkit/server

Compares two WidgetRoot objects and returns a list of deltas representing the changes required to transition from the 'before' state to the 'after' state.

```APIDOC
## diff_widget

### Description
Compares two WidgetRoot objects and returns a list of deltas. This function determines if a full replacement is necessary or if incremental updates (streaming text deltas) can be applied.

### Parameters
- **before** (WidgetRoot) - Required - The initial state of the widget.
- **after** (WidgetRoot) - Required - The new state of the widget.

### Response
- **Returns** (list[WidgetStreamingTextValueDelta | WidgetRootUpdated | WidgetComponentUpdated]) - A list of delta objects representing the changes.

### Errors
- **ValueError**: Raised if a node ID is missing from the initial render or if a streaming text update is not a cumulative prefix of the previous value.
```

--------------------------------

### Handle Widget Actions

Source: https://openai.github.io/chatkit-python/api/chatkit/server

Implement this method to handle widget or client-dispatched actions. It should yield response events asynchronously.

```python
def action(
    self,
    thread: ThreadMetadata,
    action: Action[str, Any],
    sender: WidgetItem | None,
    context: TContext,
) -> AsyncIterator[ThreadStreamEvent]:
    """Handle a widget or client-dispatched action and yield response events."""
    raise NotImplementedError(
        "The action() method must be overridden to react to actions. "
        "See https://github.com/openai/chatkit-python/blob/main/docs/widgets.md#widget-actions"
    )
```

--------------------------------

### Stream Progress Updates from Server Tools

Source: https://openai.github.io/chatkit-python/guides/respond-to-user-message

Use ctx.context.stream within a function tool to send real-time progress updates to the client during execution.

```python
from agents import RunContextWrapper, function_tool
from chatkit.agents import AgentContext
from chatkit.types import ProgressUpdateEvent


@function_tool()
async def load_document(ctx: RunContextWrapper[AgentContext], document_id: str):
    await ctx.context.stream(ProgressUpdateEvent(icon="document", text="Loading document..."))
    return await get_document_by_id(document_id)
```

--------------------------------

### Define SourceBase Model

Source: https://openai.github.io/chatkit-python/api/chatkit/types

Base class for all source metadata displayed to users.

```python
class SourceBase(BaseModel):
    """Base class for sources displayed to users."""

    title: str
    description: str | None = None
    timestamp: str | None = None
    group: str | None = None
```

--------------------------------

### Enable dictation in the client

Source: https://openai.github.io/chatkit-python/guides/accept-rich-user-input

Configure the composer options within useChatKit to enable the dictation feature.

```javascript
const chatkit = useChatKit({
  // ...
  composer: {
    dictation: {
      enabled: true,
    },
  },
});
```

--------------------------------

### Stream Agent Responses and Handle Guardrails

Source: https://openai.github.io/chatkit-python/guides/respond-to-user-message

Convert agent runs into streamable events and handle guardrail tripwire exceptions to provide user feedback.

```python
from agents import (
    InputGuardrailTripwireTriggered,
    OutputGuardrailTripwireTriggered,
    Runner,
)
from chatkit.agents import stream_agent_response
from chatkit.types import ErrorEvent

result = Runner.run_streamed(
    assistant_agent,
    input_items,
    context=agent_context,
)

try:
    async for event in stream_agent_response(agent_context, result):
        yield event
except InputGuardrailTripwireTriggered:
    yield ErrorEvent(message="We blocked that message for safety.")
except OutputGuardrailTripwireTriggered:
    yield ErrorEvent(
        message="The assistant response was blocked.",
        allow_retry=False,
    )
```

--------------------------------

### Add ImageGenerationTool to Agent

Source: https://openai.github.io/chatkit-python/guides/stream-generated-images

Integrate the ImageGenerationTool into your agent's tool list to enable image generation capabilities. This allows the model to create images when prompted.

```python
from agents import Agent
from agents.tool import ImageGenerationTool


agent = Agent(
    name="designer",
    instructions="Generate images when asked.",
    tools=[ImageGenerationTool(tool_config={"type": "image_generation"})],
)
```

--------------------------------

### Build User Message Item

Source: https://openai.github.io/chatkit-python/api/chatkit/server

Constructs a UserMessageItem from UserMessageInput, including content, attachments, and quoted text.

```python
return UserMessageItem(
            id=self.store.generate_item_id("message", thread, context),
            content=input.content,
            thread_id=thread.id,
            attachments=[
                (await self.store.load_attachment(attachment_id, context)).model_copy(
                    update={"thread_id": thread.id}
                )
                for attachment_id in input.attachments
            ],
            quoted_text=input.quoted_text,
        )
```

--------------------------------

### Define FileSource Model

Source: https://openai.github.io/chatkit-python/api/chatkit/types

Source metadata specifically for file references.

```python
class FileSource(SourceBase):
    """Source metadata for file-based references."""

    type: Literal["file"] = "file"
    filename: str
```

--------------------------------

### Chatkit Python SDK Overview

Source: https://openai.github.io/chatkit-python/api/chatkit

Overview of the core modules available in the Chatkit Python library for building chat applications.

```APIDOC
## Chatkit Python SDK Modules

### Description
The Chatkit Python SDK is organized into several core modules to handle server communication, data persistence, agent integration, and UI rendering.

### Modules
- **chatkit.server**: Contains `ChatKitServer` and helpers for receiving messages, streaming responses, and tool/widget wiring.
- **chatkit.store**: Provides interfaces and utilities for persisting threads, items, and metadata.
- **chatkit.agents**: Includes helpers for integrating ChatKit with the Agents SDK.
- **chatkit.types**: Defines Pydantic models for threads, items, events, and shared data types.
- **chatkit.errors**: Defines structured error types for consistent `ErrorEvent` emission.
- **chatkit.widgets**: Provides models and helpers like `WidgetTemplate`, `DynamicWidgetRoot`, and `BasicRoot` for rich UI responses.
```

--------------------------------

### Load ChatKit.js in HTML

Source: https://openai.github.io/chatkit-python/quickstart

Include the ChatKit.js script in your index.html file to enable ChatKit functionality in the browser.

```html
<!doctype html>
<html lang="en">
  <head>
    <meta charset="UTF-8" />
    <script src="https://cdn.platform.openai.com/deployments/chatkit/chatkit.js"></script>
  </head>
  <body>
    <div id="root"></div>
  </body>
</html>

```

--------------------------------

### Implement Action Handling

Source: https://openai.github.io/chatkit-python/api/chatkit/server

Override the `action` method to handle widget or client-dispatched actions. This asynchronous method yields response events.

```python
async def action(
        self,
        thread: ThreadMetadata,
        action: Action[str, Any],
        sender: WidgetItem | None,
        context: TContext,
    ) -> AsyncIterator[ThreadStreamEvent]:
        """Handle a widget or client-dispatched action and yield response events."""
        raise NotImplementedError(
            "The action() method must be overridden to react to actions. "
            "See https://github.com/openai/chatkit-python/blob/main/docs/widgets.md#widget-actions"
        )
```

--------------------------------

### Implement an in-memory ChatKit store

Source: https://openai.github.io/chatkit-python/quickstart

Defines a custom store class inheriting from Store to handle thread and item lifecycle. This implementation uses dictionaries for storage and includes a helper for pagination.

```python
from collections import defaultdict
from chatkit.store import NotFoundError, Store
from chatkit.types import Attachment, Page, ThreadItem, ThreadMetadata


class MyChatKitStore(Store[dict]):
    def __init__(self):
        self.threads: dict[str, ThreadMetadata] = {}
        self.items: dict[str, list[ThreadItem]] = defaultdict(list)

    async def load_thread(self, thread_id: str, context: dict) -> ThreadMetadata:
        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: ThreadMetadata, context: dict) -> None:
        self.threads[thread.id] = thread

    async def load_threads(
        self, limit: int, after: str | None, order: str, context: dict
    ) -> Page[ThreadMetadata]:
        threads = list(self.threads.values())
        return self._paginate(
            threads, after, limit, order, sort_key=lambda t: t.created_at, cursor_key=lambda t: t.id
        )

    async def load_thread_items(
        self, thread_id: str, after: str | None, limit: int, order: str, context: dict
    ) -> Page[ThreadItem]:
        items = self.items.get(thread_id, [])
        return self._paginate(
            items, after, limit, order, sort_key=lambda i: i.created_at, cursor_key=lambda i: i.id
        )

    async def add_thread_item(
        self, thread_id: str, item: ThreadItem, context: dict
    ) -> None:
        self.items[thread_id].append(item)

    async def save_item(
        self, thread_id: str, item: ThreadItem, context: dict
    ) -> None:
        items = self.items[thread_id]
        for idx, existing in enumerate(items):
            if existing.id == item.id:
                items[idx] = item
                return
        items.append(item)

    async def load_item(
        self, thread_id: str, item_id: str, context: dict
    ) -> ThreadItem:
        for item in self.items.get(thread_id, []):
            if item.id == item_id:
                return item
        raise NotFoundError(f"Item {item_id} not found in thread {thread_id}")

    async def delete_thread(self, thread_id: str, context: dict) -> None:
        self.threads.pop(thread_id, None)
        self.items.pop(thread_id, None)

    async def delete_thread_item(
        self, thread_id: str, item_id: str, context: dict
    ) -> None:
        self.items[thread_id] = [
            item for item in self.items.get(thread_id, []) if item.id != item_id
        ]

    def _paginate(self, rows: list, after: str | None, limit: int, order: str, sort_key, cursor_key):
        sorted_rows = sorted(rows, key=sort_key, reverse=order == "desc")
        start = 0
        if after:
            for idx, row in enumerate(sorted_rows):
                if cursor_key(row) == after:
                    start = idx + 1
                    break
        data = sorted_rows[start : start + limit]
        has_more = start + limit < len(sorted_rows)
        next_after = cursor_key(data[-1]) if has_more and data else None
        return Page(data=data, has_more=has_more, after=next_after)

    # Attachments are intentionally not implemented for the quickstart

    async def save_attachment(
        self, attachment: Attachment, context: dict
    ) -> None:
        raise NotImplementedError()

    async def load_attachment(
        self, attachment_id: str, context: dict
    ) -> Attachment:
        raise NotImplementedError()

    async def delete_attachment(self, attachment_id: str, context: dict) -> None:
        raise NotImplementedError()
```

--------------------------------

### Add Client Tool Output

Source: https://openai.github.io/chatkit-python/api/chatkit/types

Records the output of a client-side tool in a thread.

```APIDOC
## POST threads.add_client_tool_output

### Description
Records tool output in a thread.

### Request Body
- **type** (string) - Required - Must be "threads.add_client_tool_output"
- **params** (object) - Required
  - **thread_id** (string) - Required - The ID of the target thread
  - **result** (Any) - Required - The tool execution result
```

--------------------------------

### process

Source: https://openai.github.io/chatkit-python/api/chatkit/server

Parses an incoming ChatKit request and routes it to either streaming or non-streaming handlers.

```APIDOC
## process

### Description
Parse an incoming ChatKit request and route it to streaming or non-streaming handlers.

### Parameters
- **request** (str | bytes | bytearray) - Required - The incoming request data.
- **context** (TContext) - Required - Arbitrary per-request context provided by the caller.

### Response
- **Returns** (StreamingResult | NonStreamingResult) - The result of the processed request.
```

--------------------------------

### simple_to_agent_input

Source: https://openai.github.io/chatkit-python/api/chatkit/agents

Converts thread items into agent input using the default converter.

```APIDOC
## simple_to_agent_input

### Description
Helper function that converts one or more thread items into agent input using the default ThreadItemConverter.

### Parameters
#### Request Body
- **thread_items** (Sequence[ThreadItem] | ThreadItem) - Required - The thread item or sequence of thread items to be converted.
```

--------------------------------

### Route requests based on tool choice

Source: https://openai.github.io/chatkit-python/guides/let-users-pick-tools-and-models

Branch the inference pipeline by using the tool ID to select the appropriate agent or tool set.

```python
if tool_choice == "summarize":
    agent = summarization_agent
elif tool_choice == "search_tickets":
    agent = ticket_search_agent
else:
    agent = default_agent

result = Runner.run_streamed(agent, input_items, context=agent_context)
```

--------------------------------

### Use custom converter for agent input

Source: https://openai.github.io/chatkit-python/guides/stream-generated-images

Apply the custom converter instance when building the agent input.

```python
input_items = await MyThreadItemConverter().to_agent_input(items)
```

--------------------------------

### ItemsListParams: Pagination parameters for listing thread items

Source: https://openai.github.io/chatkit-python/api/chatkit/types

Configure pagination and ordering for listing thread items. Supports limit, order (asc/desc), and after cursor.

```python
class ItemsListParams(BaseModel):
    """Pagination parameters for listing thread items."""

    thread_id: str
    limit: int | None = None
    order: Literal["asc", "desc"] = "desc"
    after: str | None = None
```

--------------------------------

### Implement Synchronous Action Handling

Source: https://openai.github.io/chatkit-python/api/chatkit/server

Override the `sync_action` method to handle widget or client-dispatched actions synchronously. This method returns a `SyncCustomActionResponse`.

```python
async def sync_action(
        self,
        thread: ThreadMetadata,
        action: Action[str, Any],
        sender: WidgetItem | None,
        context: TContext,
    ) -> SyncCustomActionResponse:
        """Handle a widget or client-dispatched action and return a SyncCustomActionResponse."""
        raise NotImplementedError(
            "The sync_action() method must be overridden to react to sync actions. "
            "See https://github.com/openai/chatkit-python/blob/main/docs/widgets.md#widget-actions"
        )
```

--------------------------------

### Implement create_attachment

Source: https://openai.github.io/chatkit-python/api/chatkit/store

Method for creating an attachment record, which must be overridden to support two-phase file uploads.

```python
async def create_attachment(
    self, input: AttachmentCreateParams, context: TContext
) -> Attachment:
    """Create an attachment record from upload metadata."""
    raise NotImplementedError(
        f"{type(self).__name__} must override create_attachment() to support two-phase file upload"
    )
```

--------------------------------

### Implement FastAPI direct upload endpoint

Source: https://openai.github.io/chatkit-python/guides/accept-rich-user-input

Server-side implementation for handling direct file uploads via multipart/form-data.

```python
@app.post("/files")
async def upload_file(request: Request):
    form_data = await request.form()
    file = form_data.get("file")

    # Your blob store upload
    attachment = await upload_to_blob_store(file)

    return Response(content=attachment.model_dump_json(), media_type="application/json")
```

--------------------------------

### Stream a widget from a tool

Source: https://openai.github.io/chatkit-python/guides/build-interactive-responses-with-widgets

Enqueue a widget from within a function tool using AgentContext.

```python
from agents import RunContextWrapper, function_tool
from chatkit.agents import AgentContext


@function_tool(description_override="Display a sample widget to the user.")
async def sample_widget(ctx: RunContextWrapper[AgentContext]):
    message_widget = build_message_widget(...)
    await ctx.context.stream_widget(message_widget)
```

--------------------------------

### Configure entity search in ChatKit.js

Source: https://openai.github.io/chatkit-python/guides/accept-rich-user-input

Define the onTagSearch callback to return entity objects and enable the composer menu button.

```typescript
const chatkit = useChatKit({
  // ...
  entities: {
    onTagSearch: async (query: string) => {
      return [
        {
          id: "article_123",
          title: "The Future of AI",
          group: "Trending",
          icon: "globe",
          data: { type: "article" }
        },
        {
          id: "article_124",
          title: "One weird trick to improve your sleep",
          group: "Trending",
          icon: "globe",
          data: { type: "article" }
        },
      ]
    },
    // Optional: show the "@" button in the composer for added discoverability.
    showComposerMenu: true,
  },
})
```