chainlit-utils

Reusable building blocks for Chainlit applications:

• versioned PostgreSQL migrations for Chainlit's official data layer;
• exclusion of UI-only and failed messages from model context;
• conversion of simple JSON Schema settings into Chainlit widgets;
• compact serialization of changed chat settings;
• OpenAI Responses rendering and attachment upload;
• durable human-in-the-loop Responses continuations;
• MCP discovery and client-side tool execution;
• secure OIDC login with encrypted, per-browser token delegation;
• retrieval of the authenticated Chainlit user identifier.

The package owns reusable protocol and Chainlit integration. Applications keep their model catalog, service URL layout, environment settings, login-mode policy, and OpenAI clients. Package-owned settings use the CHAINLIT_UTILS_ environment-variable prefix.

Install

uv add chainlit-utils

Install the sso extra only when the application uses the OIDC login or delegated-token modules:

uv add "chainlit-utils[sso]"

For local development before publishing, append [sso] only when those modules are needed:

uv add --editable /path/to/chainlit-utils
uv add --editable "/path/to/chainlit-utils[sso]"  # With SSO support.

PostgreSQL persistence

Chainlit uses DATABASE_URL to enable its native PostgreSQL data layer. Apply the matching schema before starting the application:

uv run --env-file .env chainlit-utils-migrate
uv run --env-file .env chainlit run app.py

Migrations are checksum-protected and serialized with a PostgreSQL advisory lock. Existing applications can keep their current migration history table:

CHAINLIT_UTILS_MIGRATIONS_TABLE=_my_app_chainlit_schema_migrations

Review Chainlit's migration guidance before widening the supported Chainlit version range. The bundled migrations target Chainlit 2.12 or newer within the 2.x series.

Chat helpers

import chainlit as cl

from chainlit_utils.chat.history import (
    mark_persisted_errors_excluded,
    send_ui_message,
    text_only_chat_messages,
)


@cl.on_chat_resume
async def on_chat_resume(thread):
    mark_persisted_errors_excluded(thread)


@cl.on_message
async def on_message(_message):
    messages = text_only_chat_messages()
    # Send messages to an OpenAI-compatible client.


async def report_error(error: Exception):
    await send_ui_message(f"Chat completion failed: {error}")

text_only_chat_messages uses Chainlit's native role/content projection. It is not a lossless tool-call ledger.

UI-only messages use the chainlit_utils.exclude_from_model_context metadata key by default. Override it for an existing application without changing call sites:

CHAINLIT_UTILS_MODEL_CONTEXT_EXCLUDED_KEY=my_app.exclude_from_model_context

Chat settings

from chainlit_utils.chat.settings import settings_widgets, serialize_settings

widgets = settings_widgets(json_schema, defaults, saved_values)
await cl.ChatSettings(widgets).send()

encoded = serialize_settings(defaults, selected_values, max_length=512)
metadata = {"my_runtime_settings": encoded} if encoded is not None else {}

The widget adapter intentionally supports only booleans, string enums, strings, and integers. An integer with both minimum and maximum becomes a slider; other integers become a number input. serialize_settings sends whole numbers from those widgets as integers. The receiving application remains responsible for full schema validation.

OpenAI Responses and Files

chainlit_utils.openai.responses converts Chainlit's text transcript to Responses input, separates final-answer text from commentary, creates clickable citation elements, and validates terminal response status. CommentaryTaskList renders streamed commentary as Chainlit tasks.

chainlit_utils.openai.tools selects client-owned function calls, builds their outputs, and assembles stateless continuation input. Its function_call_output helper keeps the original caller metadata required by programmatic tool calls.

Use chainlit_utils.openai.files when a chat profile accepts attachments:

from chainlit_utils.openai.files import file_upload_overrides, with_response_file_parts

profile = cl.ChatProfile(
    name="files",
    markdown_description="Analyze files",
    config_overrides=file_upload_overrides(enabled=True),
)

input_items = await with_response_file_parts(
    input_items,
    message,
    client=openai_client,
    extra_query={"provider": "my-files-provider"},
)

The helper uploads each current Chainlit element through the OpenAI Files API and adds input_file parts to the latest user item. The effective Chainlit chat profile must have spontaneous uploads enabled.

Human-in-the-loop Responses

chainlit_utils.openai.hitl validates and serializes exact Responses function-call batches. HitlWorkflow persists the continuation in a model-context-excluded Chainlit message with a custom element. It then returns, so the pending review is ordinary persisted UI rather than a socket-bound ask coroutine.

Configure one workflow with the application-owned tool, element, and action names plus small presentation and request callbacks:

import chainlit as cl

from chainlit_utils.chat.hitl import HitlWorkflow

hitl = HitlWorkflow(
    "human_review",
    action_name="human_review_submit",
    continue_response=continue_response,
    element_name="HumanReview",
    prompt=prompt_for_calls,
    publish_final=publish_final,
    review=review_props,
    validate_outputs=validate_outputs,
)


@cl.action_callback("human_review_submit")
async def submit_review(action: cl.Action):
    submission = ReviewSubmission.model_validate(action.payload)
    await hitl.submit(
        step_id=submission.step_id,
        element_id=submission.element_id,
        revision=submission.revision,
        outputs=submission.outputs,
    )

Call await hitl.publish(response, model_id=model_id) when the tool appears and await hitl.block_new_message(message) before starting a new request. The custom element submits its opaque step, element, and revision references through Chainlit's callAction; model IDs, response IDs, function calls, and the expected element ID are always read from trusted current-thread message metadata. Each accepted action advances one Responses transition. A later interrupt updates the same persisted form; a terminal response marks the ledger complete and removes it.

Chainlit natively restores the message and custom element when a persisted thread is opened. No on_chat_end cancellation, on_chat_resume recreation, session task ownership, reconnect timer, or user_session HITL cache is needed. The application callbacks still own the API request, final rendering, payload schema, review UI, and client credentials.

HitlWorkflow also normalizes the timestamp on a restored ledger message before updating it. Chainlit 2.12's official PostgreSQL layer hydrates createdAt without the trailing Z that its own update path requires; without this narrow compatibility fix, the UI can finish while the durable ledger remains pending.

OpenAI continuations by ID require a stored prior Response. Stateless client-tool loops instead use continuation_input to replay every output item in order.

The ledger is strict and versioned. It rejects unknown fields, unexpected tool names, duplicate call IDs, and incomplete output batches instead of guessing at state that may no longer be safe to resume.

MCP tools

McpTools keeps the MCP session and discovered tool schemas in the current Chainlit user session. An application chooses the server name and URL and wires the native Chainlit callbacks:

import chainlit as cl
from chainlit.config import config

from chainlit_utils.mcp import McpTools

mcp_tools = McpTools("company-tools")
config.features.mcp.servers = [
    mcp_tools.server(
        "https://mcp.example/tools",
        headers={"Authorization": f"Bearer {api_key}"},
    )
]


@cl.on_mcp_connect
async def on_mcp_connect(connection, session):
    await mcp_tools.connect(connection, session)


@cl.on_mcp_disconnect
async def on_mcp_disconnect(name, session):
    await mcp_tools.disconnect(name, session)

Pass mcp_tools.response_tools() to a Responses request. When the model returns a function call, await mcp_tools.execute(call) validates that the tool was advertised, executes it through the active MCP session, shows a Chainlit tool step, and returns a function_call_output item.

OIDC token delegation

This integration requires the optional sso dependencies:

uv add "chainlit-utils[sso]"

The SSO modules provide three explicit layers:

• OidcClient owns discovery validation, S256 PKCE, ID-token exchange, refresh, and revocation clients.
• OAuthTokenStore encrypts grants with Fernet, stores them in Chainlit's PostgreSQL pool, isolates concurrent browser sessions, and serializes refresh across workers.
• ChainlitOAuth owns the Chainlit login/logout routes and resolves delegated credentials for HTTP discovery and WebSocket chat callbacks.

from chainlit_utils.sso.chainlit import ChainlitOAuth
from chainlit_utils.sso.oidc import OidcClient, OidcConfig
from chainlit_utils.sso.tokens import OAuthTokenStore
from openai import AsyncOpenAI

oidc = OidcClient(
    OidcConfig(
        issuer="https://id.example",
        client_id="chainlit-client",
        client_secret=client_secret,
        scopes="openid offline_access llm:invoke",
        resource="https://llm.example/",
    )
)
tokens = OAuthTokenStore(
    oidc,
    lambda: encryption_keys,
    table_name="my_chainlit_oauth_sessions",
)
oauth = ChainlitOAuth(
    provider_id="generic",
    chainlit_url="https://chat.example",
    auth_secret=chainlit_auth_secret,
    oidc=oidc,
    provider_env=("OAUTH_CLIENT_ID", "OAUTH_CLIENT_SECRET", "OAUTH_ISSUER"),
    token_store=tokens,
)

await tokens.initialize()  # Run once during application startup.
oauth.configure(app)       # Run before mounting Chainlit.
openai_client = AsyncOpenAI(api_key=oauth.credential, base_url=openai_base_url)

Encryption keys are ordered: new grants use the first key and existing grants can still be read with later keys during rotation. OAuthLoginRequired is an OpenAIError, so a delegated-credential failure propagates through the OpenAI SDK without sending an unauthenticated request.

Omit token_store when OIDC is used only to log in and the downstream service uses a static credential. The application remains responsible for validating its URLs and secrets before constructing these services.

Development

The source modules are grouped by responsibility: chat/ owns history, settings, and Chainlit HITL lifecycle helpers; openai/ owns Responses rendering, function tools, Files, and protocol-level HITL integration; sso/ owns OIDC clients, Chainlit login, and delegated-token storage. mcp.py and auth.py own MCP tools and the authenticated-user identifier. db/schema.py owns PostgreSQL schema migrations and loads its bundled SQL from db/migrations/. Import helpers from their concrete modules.

just install
just check
just build

Run just --list for the focused test, PostgreSQL, formatting, and build recipes.

The regular suite includes provider-backed OIDC browser-flow tests using an in-process signing provider; it excludes only PostgreSQL tests. Run the persistence suite against a test database:

TEST_CHAINLIT_DATABASE_URL=postgresql://chainlit:chainlit@localhost:5432/chainlit \
  just test-postgres

Each test creates and removes its own schema. This suite covers encrypted grants, refresh concurrency, key rotation, independent browser sessions, and logout races. CI runs both suites; no live identity provider is needed.