Chat Sessions

Overview

Tasks (run_tasks_until_done) are fire-and-forget, not interactive. A chat session is a long-lived conversation with an agent that keeps its sandbox, its workspace files, and its message history across turns.

NOTE: The chat sessions API are only available for Kosmos users. It is subject to change as Kosmos evolves over time.

from edison_client import EdisonClient

client = EdisonClient()  # reads EDISON_API_KEY

Chat lives on ChatMethods, mixed into EdisonClient, and is backed by the /v0.1/chat and /v0.1/conversations endpoints. Every method has an a-prefixed async twin (asend_chat_message, aget_conversation, …).

Method	Purpose
send_chat_message(project_id, message, ...)	Send a turn; opens a session if session_id is omitted
get_conversation(session_id, ...)	Read message history for one session
get_conversations(...)	List sessions the caller can see
queue_chat_message(session_id, project_id, message)	Inject a message into a turn already running
recover_chat_session(session_id, project_id, ...)	Warm a session back up after its sandbox expired
store_conversation_message(session_id, message, ...)	Append a message to history without dispatching a turn

send_chat_message returns as soon as the turn is dispatched to a sandbox — a session_id and a pending status — not the agent's answer.

first = client.send_chat_message(
    project_id,
    "Which of the attached patent figures contain R-group variability?",
    job_name="job-futurehouse-data-analysis-crow-high",
)
first.session_id   # UUID(...) — the handle for every call below
first.status       # ChatDispatchStatus.PENDING

Starting a session

A chat session lives inside a project. Create one, or reuse an existing project_id:

project_id = client.create_project("Markush triage", description="Patent figures")

Omit session_id on the first message and the platform opens a session for you; the response tells you which one. Pass that session_id back on later messages and the same sandbox is reused, with its TTL reset — the agent keeps its workspace and history.

session_id = first.session_id

client.send_chat_message(
    project_id,
    "Transcribe the three you flagged and give me CXSMILES for each.",
    session_id=session_id,
)

send_chat_message parameters

Parameter	Type	Default	Notes
project_id	UUID	required	Project to chat within
message	str	required	User message content
session_id	UUID | None	None	Existing session, or None to open a new one
job_name	str | None	None	Which deployed agent to talk to — a built-in JobNames entry or one of your own deployments (see configuration.md)
data_storage_ids	list[str | UUID] | None	None	Data storage entries to attach to this turn
ttl_seconds	int | None	None	Sandbox idle TTL override, 60–86400
timeout	int | None	None	Per-turn execution timeout override, 30–172800
channel	str	"app"	Tags message provenance; use "api" for SDK traffic

ttl_seconds and timeout are validated against those bounds client-side, so an out-of-range value raises before the request is sent.

Waiting for a reply

There is no completion signal over REST and no wait_for_reply in the SDK. Poll get_conversation and either treat "an assistant message with no pending tool calls" as the end of a turn -- or calling the submit_answer tool. An example is shown here:

import time
from uuid import UUID

from edison_client.models.chat import ConversationMessage


def wait_for_reply(
    session_id: UUID,
    *,
    after: int = 0,
    timeout_s: float = 600,
    poll_s: float = 3,
) -> ConversationMessage:
    """Block until the agent's next completed turn, then return it.

    Args:
        session_id: Session to watch.
        after: Message count already seen; only later messages count.
        timeout_s: Give up after this long.
        poll_s: Delay between polls.
    """
    deadline = time.monotonic() + timeout_s
    while time.monotonic() < deadline:
        conversation = client.get_conversation(session_id, limit=None)
        for message in conversation.messages[after:]:
            if message.role == "assistant" and not message.tool_calls:
                return message
        time.sleep(poll_s)
    raise TimeoutError(f"No reply on session {session_id} within {timeout_s}s")


reply = wait_for_reply(session_id)
print(reply.content)

To wait on a follow-up turn, pass the count of messages you have already seen:

seen = len(client.get_conversation(session_id, limit=None).messages)
client.send_chat_message(project_id, "Now check figure 5.", session_id=session_id)
print(wait_for_reply(session_id, after=seen).content)

Attaching data

data_storage_ids accepts bare UUIDs or data_entry:<uuid> URIs from the data storage service (the data_entry: prefix is stripped for you). The entries are mounted into the agent's workspace for that turn.

figures = client.store_file_content(
    "US1234567 figures",
    "US1234567.pdf",
    project_id=project_id,
)

client.send_chat_message(
    project_id,
    "Extract every structure figure from this patent.",
    session_id=session_id,
    data_storage_ids=[figures.data_storage.id],
)

See File management for the upload side.

Steering a turn that is already running

send_chat_message on a busy session waits its turn. queue_chat_message injects into the running rollout instead — the agent picks the message up at its next step. This is the "actually, also check X" affordance while the agent is mid-work.

pending = client.queue_chat_message(
    session_id,
    project_id,
    "Skip figure 4 — we already have it.",
)
pending.id         # pending message ID
pending.target_id  # the session it was queued against

Reading history

limit=None fetches the whole conversation, following pagination and reassembling messages the API had to split into content blocks.

from edison_client.models.chat import ConversationDetail

full: ConversationDetail = client.get_conversation(session_id, limit=None)

for message in full.messages:
    if message.tool_calls:                    # assistant asking for a tool
        for call in message.tool_calls:
            print(f"→ {call['function']['name']}({call['function']['arguments']})")
    elif message.role == "tool":              # the tool's result
        print(f"← {message.name}: {str(message.content)[:120]}")
    else:
        print(f"{message.role}: {str(message.content)[:120]}")

Or one page at a time, most recent first by sort_order:

recent = client.get_conversation(session_id, limit=20)
recent.has_more             # older messages exist beyond this page
recent.continuation_cursor  # cursor into a split content block, if any

limit defaults to 50. offset requires limit — passing offset with limit=None raises ValueError.

Messages convert to aviary types when you want to replay them into an LLM:

messages = [m.to_aviary_message() for m in full.messages]

List sessions the caller can see (limit default 25, max 200):

for conversation in client.get_conversations(limit=50).conversations:
    print(conversation.session_id, conversation.created_at)

Sandbox knobs

Defaults ususally work for interactive use, though raising timeouts may be necessary for long autonomous turns.

client.send_chat_message(
    project_id,
    "Run the full corpus and report when finished.",
    session_id=session_id,
    ttl_seconds=8 * 60 * 60,   # keep the sandbox alive 8h idle (60–86400)
    timeout=4 * 60 * 60,       # allow one turn to run 4h (30–172800)
    job_name="job-futurehouse-data-analysis-crow-high",
    channel="api",
)

These override the agent profile's deploy-time timeouts for this session only,

Chatting with an agent you deployed yourself is the same call — job_name takes any job you can reach:

client.send_chat_message(
    project_id,
    "Which of these figures have R-group variability?",
    job_name="acme-assay-analysis",
)

A deployment with MIN_REPLICAS=0 scales to zero, so the first message to an idle custom agent pays a cold start. Deploy with at least one replica for anything interactive.

Response models

Import from edison_client.models.chat.

ChatMessageResponse

Field	Type	Notes
session_id	UUID	New or existing session
status	ChatDispatchStatus	Dispatch lifecycle status
claim_name	str | None	Kubernetes SandboxClaim name; internal users only
pod_name	str | None	Pod serving the sandbox; internal users only
logs_url	str | None	Console logs URL; internal users only

ChatDispatchStatus is a StrEnum: pending, dispatched, processing, completed, failed. It describes dispatch, not the agent's progress — a completed dispatch means the turn was accepted, not answered.

ConversationDetail

Field	Type	Notes
messages	list[ConversationMessage]	Ordered oldest → newest
session_id	UUID	Use for follow-up messages
has_more	bool	Older messages exist beyond this page
continuation_cursor	ConversationContinuationCursor | None	message_id + block_index

ConversationMessage

Field	Type	Notes
role	str	user, assistant, tool, system
content	str | list[dict]	Text, or content blocks
tool_calls	list[dict] | None	Present on assistant tool-request messages
tool_call_id	str | None	Present on tool-response messages
name	str | None	Tool name on tool-response messages
info	dict | None	Aviary Message.info — carries channel and data_storage_ids
metadata	dict | None	Caller-supplied metadata
message_id	UUID | None	Stable ID
content_block_start / _end / _total	int | None	Set when the API split one message across pages

to_aviary_message() converts to Message / ToolRequestMessage / ToolResponseMessage.

QueueMessageResponse

Field	Type
id	UUID — pending message ID
target_id	UUID — session (chat) or trajectory (run)

Errors

Writes raise ChatError; reads raise ChatFetchError (a subclass). Connection errors are retried three times with exponential backoff before surfacing.

from edison_client.clients.chat_methods import ChatError

try:
    client.send_chat_message(project_id, "…", session_id=session_id)
except ChatError as exc:
    print(f"chat turn rejected: {exc}")

Async: fan out across sessions

Each session gets its own sandbox, so parallel sessions genuinely run in parallel — the useful shape for batch work from a notebook or service.

import asyncio
from uuid import UUID

from edison_client.models.chat import ConversationDetail


async def triage(patents: list[str]) -> list[UUID]:
    """Open one session per patent and return their session IDs."""
    responses = await asyncio.gather(*[
        client.asend_chat_message(
            project_id,
            f"Transcribe every Markush structure in {patent}.",
            job_name="job-futurehouse-data-analysis-crow-high",
        )
        for patent in patents
    ])
    return [r.session_id for r in responses]


async def collect(session_id: UUID) -> ConversationDetail:
    return await client.aget_conversation(session_id, limit=None)


sessions = asyncio.run(triage(["US1234567", "US7654321", "EP9876543"]))

Stopping and resetting

queue_chat_message steers a turn; these two stop it.

client.interrupt_chat_session(session_id)   # halt the turn in flight
client.reset_chat_sandbox(project_id)       # discard the sandbox itself

interrupt_chat_session halts the current turn and leaves the session usable — the sandbox, workspace, and history survive, so the next send_chat_message continues normally.

reset_chat_sandbox is the heavier hammer and is keyed on the project, not the session: it discards the chat sandbox entirely. Conversation history is durable, so the session survives and recovers on the next message, but anything the agent left in an ephemeral workspace is gone. Reach for it when a sandbox is wedged rather than when a turn went wrong.