Session And Request Context

`Session`

Session is the main runtime context for chat and API interactions. If your handler needs to know “who is talking?”, “what happened earlier?”, or “where should I send the reply?”, the answer is usually on Session. In hosted apps, that identity is app-scoped. Session is not telling you about some global Capsule user. It is telling you about the current user of this deployed app.

Worked example

This is the kind of session-aware flow you see in a real app:

@app.message()
async def handle(session: cpsl.Session, msg: cpsl.Message):
    if "sync" in (msg.text or "").lower():
        handle = await sync_threads.submit(session=session)
        await session.show_task(handle, message="Mailbox sync started")
        return

    await session.notify("Thinking...")

    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "You are a support operations copilot."},
            *session.chat_messages(msg),
        ],
    )
    await session.reply(response.choices[0].message.content or "No response.")

In one handler, Session gives you:

the live conversation history for the model call
a reply target for text and structured blocks
a way to surface background work back into the chat UI

Core fields

Field	Meaning
`id`	Unique session id
`user`	`UserInfo` for the current app user
`channel`	`SessionChannel` describing the transport
`history`	Recent message history
`data`	Per-session persistent key-value store
`integrations`	Connected runtime credentials
`db`	Scoped collection access for the current session

Small example:

@app.message()
async def handle(session: cpsl.Session, msg: cpsl.Message):
    await session.reply(
        f"session={session.id[:8]} "
        f"user={session.user.email or 'anon'} "
        f"text={msg.text}"
    )

On hosted web apps, session.user comes from the app’s sign-in flow. External channels such as Telegram may provide a channel-scoped platform user id until that identity is explicitly linked to an app user. The same email can appear in another Capsule app too, but it is still a separate app user in that other app’s context.

Common methods

Method	Purpose
`await reply(text)`	Send a complete response
`await notify(text)`	Send a lightweight status update
`await stream(chunks)`	Pipe an async iterator of strings
`stream_reply()`	Open a streaming reply context manager
`await stream_reply_from(stream)`	Pipe a model stream and return the final text
`chat_messages(current, cls=None)`	Build merged user/assistant chat history
`await show(block)`	Render a structured block
`await show_task(handle_or_id, message=None)`	Show a live task card
`await show_integration(type, reason="")`	Show a non-blocking integration card
`get_integration(type_or_config)`	Return connected credentials if present
`await require_integration(type_or_config, reason="")`	Prompt when credentials are missing
`await show_browser(...)`	Show an embedded browser/live view block
`await show_image(source, alt="", width=None)`	Render an image inline
`await prompt_file(...)`	Block until the user uploads a file
`await prompt_integration(...)`	Block until the user connects an integration

Real LLM usage usually looks like this:

from openai import AsyncOpenAI

client = AsyncOpenAI()


@app.message()
async def handle(session: cpsl.Session, msg: cpsl.Message):
    response = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[
            {"role": "system", "content": "You are a concise Capsule assistant."},
            *session.chat_messages(msg),
        ],
    )
    await session.reply(response.choices[0].message.content or "No response.")

session.chat_messages(msg) is useful because it already merges Capsule history into the alternating user / assistant format most LLM APIs expect.

Streaming helpers

`stream_reply()`

async with session.stream_reply() as reply:
    reply.write("Starting...\n")
    reply.write("More text...\n")

The accumulated text is written to session history when the context exits, which means a streaming response still becomes part of the conversation record.

`stream_reply_from(stream)`

full_text = await session.stream_reply_from(model_stream)

This is the one-liner for async model streams such as BAML outputs. In practice it saves you from rewriting the same async with session.stream_reply() loop over and over.

Structured UI helpers

The Session object is also how chat handlers render richer UI.

`show_task(...)`

handle = await build_report.submit(session=session, report_id="daily")
await session.show_task(handle, message="Building report...")

`show_integration(...)`

await session.show_integration(
    cpsl.Integration.GITHUB,
    reason="Connect GitHub to unlock project analysis",
)

Use require_integration(...) when the current flow cannot continue without credentials:

github = await session.require_integration(
    cpsl.Integration.GITHUB,
    reason="Connect GitHub so I can inspect repositories.",
)
headers = {"Authorization": f"Bearer {github.access_token}"}

`show_browser(...)`

await session.show_browser(
    url="https://example.com",
    title="Browser session",
)

Use this when a chat or workflow should surface a live browser view back to the user.

`prompt_file(...)`

upload = await session.prompt_file(
    message="Upload a CSV export",
    accept=".csv,text/csv",
)
await session.reply(f"Received {upload.name}")

`RequestContext`

Use RequestContext in @app.data() and @app.endpoint() handlers when you need caller metadata outside chat.

Fields

Field	Meaning
`user`	`UserInfo` for the caller as an app user
`integrations`	Connected integrations available to the caller
`authenticated`	Boolean auth state
`request`	Underlying request object when available

Example:

@app.data("threads", access="authenticated")
async def get_threads(ctx: cpsl.RequestContext, status: str = ""):
    gmail = ctx.integrations.get(cpsl.Integration.GMAIL)
    if not gmail:
        return {"threads": [], "gmail_connected": False}

    rows = await threads.find(filter={"status": status} if status else {}, limit=50)
    return {"threads": rows, "gmail_connected": True}

If a page or endpoint needs to know who is calling it, RequestContext is the right abstraction. If a chat turn needs to know who is calling it, you already have Session.

Messages and attachments

`Message`

Field	Meaning
`text`	Message body
`sender`	`"user"` or `"app"`
`channel_type`	Transport type
`timestamp`	Unix epoch seconds
`attachments`	Optional list of `Attachment`

Example:

@app.message()
async def handle(session: cpsl.Session, msg: cpsl.Message):
    if msg.attachments:
        first = msg.attachments[0]
        await session.reply(f"First attachment: {first.name} ({first.content_type})")
        return

`Attachment`

Attachment represents a file attached to a user message. Useful fields:

name
content_type
url
size

Use await attachment.download(path) to save it locally.

Blocks and uploads

`Block`

Block(type, payload) is the low-level structured UI primitive used by helpers like show_task() and show_integration(). If you are writing normal app code, prefer the higher-level helpers first. Reach for raw Block(...) only when you need a custom block type.

`FileUpload`

Returned by prompt_file(...). Useful fields:

name
content_type
url
size
path when auto-downloaded

Use await upload.download(path) if you want to save it manually.

Identity types

`UserInfo`

Common fields:

id
email
org_id
owner_id property for owner-scoped runtime identity

`SessionChannel`

Describes the transport, such as chat or Telegram.

Exceptions

These are part of the public session-facing API:

IntegrationTimeout
IntegrationDeclined
FileUploadTimeout

`current_session()`

runtime_session = cpsl.current_session()

Returns the active runtime Session when one exists. In scheduled handlers and some background contexts, this may be a synthetic session with identity and integrations but no live reply target. Example:

@app.schedule("0 * * * *")
async def hourly_job():
    session = cpsl.current_session()
    owner = session.user.owner_id if session else "unknown"
    print(f"running for owner={owner}")

Python App API

CLI

Session And Request Context

`Session`

Worked example

Core fields

Common methods

Streaming helpers

`stream_reply()`

`stream_reply_from(stream)`

Structured UI helpers

`show_task(...)`

`show_integration(...)`

`show_browser(...)`

`prompt_file(...)`

`RequestContext`

Fields

Messages and attachments

`Message`

`Attachment`

Blocks and uploads

`Block`

`FileUpload`

Identity types

`UserInfo`

`SessionChannel`

Exceptions

`current_session()`

See also

​Session

​Worked example

​Core fields

​Common methods

​Streaming helpers

​stream_reply()

​stream_reply_from(stream)

​Structured UI helpers

​show_task(...)

​show_integration(...)

​show_browser(...)

​prompt_file(...)

​RequestContext

​Fields

​Messages and attachments

​Message

​Attachment

​Blocks and uploads

​Block

​FileUpload

​Identity types

​UserInfo

​SessionChannel

​Exceptions

​current_session()

​See also

`Session`

Worked example

Core fields

Common methods

Streaming helpers

`stream_reply()`

`stream_reply_from(stream)`

Structured UI helpers

`show_task(...)`

`show_integration(...)`

`show_browser(...)`

`prompt_file(...)`

`RequestContext`

Fields

Messages and attachments

`Message`

`Attachment`

Blocks and uploads

`Block`

`FileUpload`

Identity types

`UserInfo`

`SessionChannel`

Exceptions

`current_session()`

See also