ChatKit Python SDK

https://openai.github.io/chatkit-python
Admin
ChatKit Python SDK provides tools and functionalities for integrating ChatKit services and features...
Tokens:72,360
Snippets:576
Trust Score:8.3
Update:1 month ago
Show doc for...
Tokens:
Raw
### 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.
```

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

### All code examples

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

All code examples from the source, each representing a distinct class definition.

```python
class ThreadsAddStructuredInputReq(BaseReq):
    """Request to submit answers for a pending structured input item."""

    type: Literal["threads.add_structured_input"] = "threads.add_structured_input"
    params: ThreadAddStructuredInputParams
```

```python
class ThreadAddStructuredInputParams(BaseModel):
    """Parameters for adding structured input to a thread."""

    thread_id: str
    item_id: str
    input: StructuredInputSubmission
```

```python
class ThreadsCustomActionReq(BaseReq):
    """Request to execute a custom action within a thread."""

    type: Literal["threads.custom_action"] = "threads.custom_action"
    params: ThreadCustomActionParams
```

```python
class ThreadsSyncCustomActionReq(BaseReq):
    """Request to execute a custom action and return a single item update."""

    type: Literal["threads.sync_custom_action"] = "threads.sync_custom_action"
    params: ThreadSyncCustomActionParams
```

```python
class ThreadSyncCustomActionParams(BaseModel):
    """Parameters describing the custom action to execute."""

    thread_id: str
    item_id: str | None = None
    action: Action[str, Any]
```

```python
class ThreadsRetryAfterItemReq(BaseReq):
    """Request to retry processing after a specific thread item."""

    type: Literal["threads.retry_after_item"] = "threads.retry_after_item"
    params: ThreadRetryAfterItemParams
```

```python
class ThreadRetryAfterItemParams(BaseModel):
    """Parameters specifying which item to retry."""

    thread_id: str
    item_id: str
```

```python
class ItemsFeedbackReq(BaseReq):
    """Request to submit feedback on specific items."""

    type: Literal["items.feedback"] = "items.feedback"
    params: ItemFeedbackParams
```

```python
class ItemFeedbackParams(BaseModel):
    """Parameters describing feedback targets and sentiment."""

    thread_id: str
    item_ids: list[str]
    kind: FeedbackKind
```

```python
class AttachmentsDeleteReq(BaseReq):
    """Request to remove an attachment."""

    type: Literal["attachments.delete"] = "attachments.delete"
    params: AttachmentDeleteParams
```

```python
class AttachmentDeleteParams(BaseModel):
    """Parameters identifying an attachment to delete."""

    attachment_id: str
```

```python
class AttachmentsCreateReq(BaseReq):
    """Request to register a new attachment."""

    type: Literal["attachments.create"] = "attachments.create"
    params: AttachmentCreateParams
```

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

### 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

```

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

### ChatKit Types Code Examples

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

All code examples from the source file, including class definitions, attribute declarations, and function definitions.

```python
class URLSource(SourceBase):
    """Source metadata for external URLs."""

    type: Literal["url"] = "url"
    url: str
    attribution: str | None = None

```

```python
class EntitySource(SourceBase):
    """Source metadata for entity references."""

    type: Literal["entity"] = "entity"
    id: str
    icon: IconName | None = None
    label: str | None = None
    """Optional label shown with the icon in the default entity hover header
    when no preview callback is provided.
    """
    inline_label: str | None = None
    """Optional label for the inline annotation view. When not provided, the icon is used instead."""
    interactive: bool = False
    """Per-entity toggle to wire client callbacks and render this entity as interactive."""
    data: dict[str, Any] = Field(default_factory=dict)
    """Additional data for the entity source that is passed to client entity callbacks."""

    preview: Literal["lazy"] | None = Field(
        default=None,
        deprecated=True,
        description="This field is ignored. Please use the entities.onRequestPreview ChatKit.js option instead.",
    )

```

```python
label: str | None = None

```

```python
inline_label: str | None = None

```

```python
interactive: bool = False

```

```python
data: dict[str, Any] = Field(default_factory=dict)

```

```python
def is_streaming_req(request: ChatKitReq) -> TypeIs[StreamingReq]:
    """Return True if the given request should be processed as streaming."""
    return isinstance(
        request,
        (
            ThreadsCreateReq,
            ThreadsAddUserMessageReq,
            ThreadsRetryAfterItemReq,
            ThreadsAddClientToolOutputReq,
            ThreadsAddStructuredInputReq,
            ThreadsCustomActionReq,
        ),
    )

```

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

### Two-phase AttachmentStore implementation

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

Example two-phase store issuing a multipart upload URL. Implements AttachmentStore.create_attachment to build an upload_url, persist Attachment metadata via Store.save_attachment, and return it; also implements delete_attachment to delete stored bytes.

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

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

### 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

```

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

### Implement ChatKitServer.transcribe with OpenAI Audio API

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

Implement `ChatKitServer.transcribe` to accept audio input and return a `TranscriptionResult` with the final text. The client sends audio in one of the MIME types `audio/webm;codecs=opus`, `audio/mp4`, or `audio/ogg;codecs=opus`; use `audio_input.media_type` to get the base media type. Raises HTTP 400 for unexpected formats.

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

```

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

### 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,
        }
    )
```

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

### start_workflow

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

Function signature and implementation for start_workflow.

```python
start_workflow(workflow: Workflow) -> None

```

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

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

### get_stream_options

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

Method to get stream-level runtime options, allowing the user to cancel the stream by default. Override to customize behavior.

```APIDOC
## get_stream_options

### Description
Return stream-level runtime options. Allows the user to cancel the stream by default. Override this method to customize behavior.

### Method
Function

### Signature
```python
def get_stream_options(
    thread: ThreadMetadata, context: TContext
) -> StreamOptions
```

### Parameters
- **thread** (ThreadMetadata) - Required - The thread metadata.
- **context** (TContext) - Required - Arbitrary per-request context provided by the caller.

### Returns
- **StreamOptions** - Stream-level runtime options.

### Source Code
```python
def get_stream_options(
    self, thread: ThreadMetadata, context: TContext
) -> StreamOptions:
    """
    Return stream-level runtime options. Allows the user to cancel the stream by default.
    Override this method to customize behavior.
    """
    return StreamOptions(allow_cancel=True)
```
```

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

### 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"})
```

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

### 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 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} />;
}

```

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

### Convert tags to model input with ThreadItemConverter.tag_to_message_content

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

Override ThreadItemConverter.tag_to_message_content to convert structured tags into model-readable content. This example wraps article tag details in custom <ARTICLE_TAG> markup, loading a summary via fetch_article_summary(tag.id).

```python
from chatkit.agents import ThreadItemConverter
from chatkit.types import UserMessageTagContent
from openai.types.responses import ResponseInputTextParam


class MyThreadItemConverter(ThreadItemConverter):
    async def tag_to_message_content(
        self, tag: UserMessageTagContent
    ) -> ResponseInputTextParam:
        if tag.type == "article":
            # Load or unpack the entity the tag refers to
            summary = await fetch_article_summary(tag.id)
            return ResponseInputTextParam(
                type="input_text",
                text=(
                    "<ARTICLE_TAG>\n"
                    f"ID: {tag.id}\n"
                    f"Title: {tag.text}\n"
                    f"Summary: {summary}\n"
                    "</ARTICLE_TAG>"
                ),
            )
```

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

### 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.
```

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

### 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} />
    </>
  );
}
```

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

### Convert attachments to model input with ThreadItemConverter

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

Subclass ThreadItemConverter and implement attachment_to_message_content to convert attachments into ResponseInputContentParam values before calling Runner.run_streamed. The example uses a blob fetch helper and converts ImageAttachment to ResponseInputImageParam and PDFs to ResponseInputFileParam using data URLs. For other text formats, check API support before sending as ResponseInputFileParam.

```python
from chatkit.agents import ThreadItemConverter
from chatkit.types import ImageAttachment
from openai.types.responses import ResponseInputFileParam, ResponseInputImageParam


async def read_bytes(attachment_id: str) -> bytes:
    ...  # fetch from your blob store


def as_data_url(mime: str, content: bytes) -> str:
    return "data:" + mime + ";base64," + base64.b64encode(content).decode("utf-8")


class MyConverter(ThreadItemConverter):
    async def attachment_to_message_content(self, attachment):
        content = await read_bytes(attachment.id)
        if isinstance(attachment, ImageAttachment):
            return ResponseInputImageParam(
                type="input_image",
                detail="auto",
                image_url=as_data_url(attachment.mime_type, content),
            )
        if attachment.mime_type == "application/pdf":
            return ResponseInputFileParam(
                type="input_file",
                file_data=as_data_url(attachment.mime_type, content),
                filename=attachment.name or "unknown",
            )
        # For other text formats, check for API support first before
        # sending as a ResponseInputFileParam.

```

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

### 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},
)
```

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

### 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,
        }),
      };
    }
  },
});
```

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

### start_workflow

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

Begins streaming a new workflow item. If the workflow type is not 'reasoning' and has no tasks, the added event is deferred until tasks are available.

```APIDOC
## start_workflow

### Description
Begin streaming a new workflow item.

### Method
Async Function

### Signature
```
async start_workflow(workflow: Workflow) -> None
```

### Parameters
- **workflow** (Workflow) - Required - The workflow to start.

### Returns
- **None** - This method does not return a value.

### Example
```python
await start_workflow(workflow)
```
```

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

### Prompt the model about mentions with system guidance

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

System prompt guidance to help the assistant understand the ARTICLE_TAG input item. Instructs the model to use the tag as trusted context, avoid verbatim restating, and call fetch_article when more detail is needed.

```markdown
- <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.
```

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

### 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

```

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

### 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!")],
            )
        )
```

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

### widget_to_input async method

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

Converts a WidgetItem into a user message containing a text description and JSON representation of the widget, excluding unset and None fields.

```python
async def widget_to_input(
    self, item: WidgetItem
) -> TResponseInputItem | list[TResponseInputItem] | None:
    """
    Convert a WidgetItem into input item(s) to send to the model.
    By default, WidgetItems converted to a text description with a JSON representation of the widget.
    """
    return Message(
        type="message",
        content=[
            ResponseInputTextParam(
                type="input_text",
                text=f"The following graphical UI widget (id: {item.id}) was displayed to the user:"
                + item.widget.model_dump_json(
                    exclude_unset=True, exclude_none=True
                ),
            )
        ],
        role="user",
    )

```

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

### 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")

```

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

### partial_image_index_to_progress implementation

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

Implementation of partial_image_index_to_progress method.

```python
def partial_image_index_to_progress(self, partial_image_index: int) -> float:
    """Convert a partial image index into a normalized progress value.

    Args:
        partial_image_index: The index of the partial image update.
    """
    if self.partial_images is None or self.partial_images <= 0:
        return 0.0

    return min(1.0, partial_image_index / self.partial_images)
```

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

### 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})),
      };
    }
  },
});
```

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

### sync_action

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

Handle a widget or client-dispatched action and return a SyncCustomActionResponse. This async method should be overridden.

```APIDOC
## sync_action

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

### Method
async

### Signature
```python
async def sync_action(
    self,
    thread: ThreadMetadata,
    action: Action[str, Any],
    sender: WidgetItem | None,
    context: TContext,
) -> SyncCustomActionResponse
```

### Parameters
- **thread** (ThreadMetadata) - Required - Metadata about the thread.
- **action** (Action[str, Any]) - Required - The action to handle.
- **sender** (WidgetItem | None) - Required - The widget item that dispatched the action, if any.
- **context** (TContext) - Required - Arbitrary per-request context provided by the caller.

### Returns
- **SyncCustomActionResponse** - The response to the sync action.

### Source Code
```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"
    )
```
```

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

### ChatKitServer class with Store and AttachmentStore

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

Defines the ChatKitServer abstract base class with a generic context type. The constructor accepts a required Store and an optional AttachmentStore; _get_attachment_store raises RuntimeError if no AttachmentStore is configured.

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

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

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

### generated_image_to_input implementation

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

Default implementation returns None if no image, otherwise constructs a user message with text and image parameters. Override to customize conversion.

```python
async def generated_image_to_input(
    self, item: GeneratedImageItem
) -> TResponseInputItem | list[TResponseInputItem] | None:
    """
    Convert a GeneratedImageItem into input item(s) to send to the model.
    Override this method to customize the conversion of generated images, such as when your
    generated image url is not publicly reachable.
    """
    if not item.image:
        return None

    return Message(
        type="message",
        content=[
            ResponseInputTextParam(
                type="input_text",
                text="The following image was generated by the agent.",
            ),
            ResponseInputImageParam(
                type="input_image",
                detail="auto",
                image_url=item.image.url,
            ),
        ],
        role="user",
    )

```

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

### 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,
)
```

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

### 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.",
      },
    ],
  },
});
```

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

### action

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

Handles a widget or client-dispatched action and yields response events. Must be overridden to react to actions.

```APIDOC
## action(thread, action, sender, context)

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

### Method
action

### Signature
```python
def action(
    self,
    thread: ThreadMetadata,
    action: Action[str, Any],
    sender: WidgetItem | None,
    context: TContext,
) -> AsyncIterator[ThreadStreamEvent]:
```

### Parameters
- **thread** (ThreadMetadata) - Required - Metadata for the thread being processed.
- **action** (Action[str, Any]) - Required - The action to handle.
- **sender** (WidgetItem | None) - Required - The widget item that sent the action, if any.
- **context** (TContext) - Required - Arbitrary per-request context provided by the caller.

### Returns
An async iterator that yields response events.
```

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

### 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",
  },
};

```

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

### widget_to_input (async)

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

Converts a WidgetItem into input item(s) to send to the model. By default, WidgetItems are converted to a text description with a JSON representation of the widget.

```APIDOC
## widget_to_input (async)

### Description
Converts a WidgetItem into input item(s) to send to the model. By default, WidgetItems are converted to a text description with a JSON representation of the widget.

### Method
async function

### Signature
widget_to_input(item: WidgetItem) -> TResponseInputItem | list[TResponseInputItem] | None

### Parameters
- **item** (WidgetItem) - Required - The widget item to convert.

### Returns
- TResponseInputItem | list[TResponseInputItem] | None - The converted input item(s) to send to the model.

### Example
```python
result = await agent.widget_to_input(item)
```
```

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

### sync_action

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

Handles a widget or client-dispatched action and returns a SyncCustomActionResponse. Must be overridden to react to sync actions.

```APIDOC
## sync_action(thread, action, sender, context)

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

### Method
sync_action

### Signature
```python
async def sync_action(
    self,
    thread: ThreadMetadata,
    action: Action[str, Any],
    sender: WidgetItem | None,
    context: TContext,
) -> SyncCustomActionResponse:
```

### Parameters
- **thread** (ThreadMetadata) - Required - Metadata for the thread being processed.
- **action** (Action[str, Any]) - Required - The action to handle.
- **sender** (WidgetItem | None) - Required - The widget item that sent the action, if any.
- **context** (TContext) - Required - Arbitrary per-request context provided by the caller.

### Returns
A SyncCustomActionResponse.
```

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

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

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

### Converting Thread Items to Agent Input

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

This code shows the implementation of methods that convert various thread item types into response input items for the agent. It includes handling for assistant messages, client tool calls, and end-of-turn items, with a dispatcher that routes each item type to its corresponding conversion method. The `to_agent_input` method processes a sequence of thread items, making a shallow copy to avoid mutation issues, and returns a list of response input items.

```python
async def assistant_message_to_input(
        self, item: AssistantMessageItem
    ) -> TResponseInputItem | list[TResponseInputItem] | None:
        return EasyInputMessageParam(
            type="message",
            content=[
                # content param doesn't support the assistant message content types
                cast(
                    ResponseInputContentParam,
                    ResponseOutputText(
                        type="output_text",
                        text=c.text,
                        annotations=[],  # TODO: these should be sent back as well
                    ).model_dump(),
                )
                for c in item.content
            ],
            role="assistant",
        )

    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),
            ),
        ]

    async def end_of_turn_to_input(
        self, item: EndOfTurnItem
    ) -> TResponseInputItem | list[TResponseInputItem] | None:
        # Only used for UI hints - you shouldn't need to override this
        return None

    async def _thread_item_to_input_item(
        self,
        item: ThreadItem,
        is_last_message: bool = True,
    ) -> list[TResponseInputItem]:
        match item:
            case UserMessageItem():
                out = await self.user_message_to_input(item, is_last_message) or []
                return out if isinstance(out, list) else [out]
            case AssistantMessageItem():
                out = await self.assistant_message_to_input(item) or []
                return out if isinstance(out, list) else [out]
            case ClientToolCallItem():
                out = await self.client_tool_call_to_input(item) or []
                return out if isinstance(out, list) else [out]
            case EndOfTurnItem():
                out = await self.end_of_turn_to_input(item) or []
                return out if isinstance(out, list) else [out]
            case WidgetItem():
                out = await self.widget_to_input(item) or []
                return out if isinstance(out, list) else [out]
            case WorkflowItem():
                out = await self.workflow_to_input(item) or []
                return out if isinstance(out, list) else [out]
            case TaskItem():
                out = await self.task_to_input(item) or []
                return out if isinstance(out, list) else [out]
            case StructuredInputItem():
                out = await self.structured_input_to_input(item) or []
                return out if isinstance(out, list) else [out]
            case HiddenContextItem():
                out = await self.hidden_context_to_input(item) or []
                return out if isinstance(out, list) else [out]
            case SDKHiddenContextItem():
                out = await self.sdk_hidden_context_to_input(item) or []
                return out if isinstance(out, list) else [out]
            case GeneratedImageItem():
                out = await self.generated_image_to_input(item) or []
                return out if isinstance(out, list) else [out]
            case _:
                assert_never(item)

    async def to_agent_input(
        self,
        thread_items: Sequence[ThreadItem] | ThreadItem,
    ) -> list[TResponseInputItem]:
        if isinstance(thread_items, Sequence):
            # shallow copy in case caller mutates the list while we're iterating
            thread_items = thread_items[:]
        else:
            thread_items = [thread_items]
        output: list[TResponseInputItem] = []
        for item in thread_items:
            output.extend(
                await self._thread_item_to_input_item(
                    item,
                    is_last_message=item is thread_items[-1],
                )
            )
        return 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,
      },
    ],
  },
});
```

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

### container_file_citation_to_annotation implementation

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

Implementation of container_file_citation_to_annotation method.

```python
async def container_file_citation_to_annotation(
    self, container_file_citation: AnnotationContainerFileCitation
) -> Annotation | None:
    """Convert a Responses API container file citation into an assistant message annotation."""
    filename = container_file_citation.filename
    if not filename:
        return None

    return Annotation(
        source=FileSource(filename=filename, title=filename),
        index=container_file_citation.end_index,
    )
```

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

### process

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

Async method to parse an incoming ChatKit request and route it to streaming or non-streaming handlers.

```APIDOC
## process (async)

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

### Method
Async Function

### Signature
```python
async def process(
    request: str | bytes | bytearray, context: TContext
) -> StreamingResult | NonStreamingResult
```

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

### Returns
- **StreamingResult | NonStreamingResult** - The result of processing the request, either streaming or non-streaming.

### Source Code
```python
async def process(
    self, request: str | bytes | bytearray, context: TContext
) -> StreamingResult | NonStreamingResult:
    """Parse an incoming ChatKit request and route it to streaming or non-streaming handlers."""
    parsed_request = TypeAdapter[ChatKitReq](ChatKitReq).validate_json(request)
    logger.info(f"Received request op: {parsed_request.type}")

    if is_streaming_req(parsed_request):
        return StreamingResult(self._process_streaming(parsed_request, context))
    else:
        return NonStreamingResult(
            await self._process_non_streaming(parsed_request, context)
        )
```
```
Tokens:
Raw