Calls

session

function

def session(model: 'ModelClient', *, store: 'Store | None' = None, registry: 'ToolRegistry | None' = None, tools: 'Sequence[Callable[..., Any]]' = (), tenant: 'str' = 'local', principal: 'str | None' = None, concurrency: 'int' = 4, **runtime_options: 'Any') -> 'AsyncIterator[Session]'

Assemble a store, a registry, a Runtime and a running Worker.

async with psych_runtime.session(model, tools=[lookup_order]) as s:
    view = await s.ask(spec, "where is order A1?")
    print(view.text)

The Worker runs for the life of the block and is stopped and awaited on the way out, including when the block raises. A Run still executing when the block ends is not lost: the Worker returns it RUNNABLE and the log is intact, so another Worker -- or the next session() over the same store -- picks it up where it stopped (DESIGN.md §8).

Args: model: the ModelClient. Required, deliberately: see the module docstring. FakeModel for a run that must not touch a network. store: where Runs live. Defaults to InMemoryStore(), which is a real Store and an entirely in-process one: nothing survives the process, so it is right for a first run, a test and a script, and wrong for anything whose Runs need to outlive it. Pass a PostgresStore, MySQLStore or DynamoDBStore for that, already migrated. registry: an existing ToolRegistry. Defaults to a fresh one. tools: plain functions to register into it, as @registry.register would. The schema comes from each function's type hints and its description from its docstring, so a function with neither is refused here rather than reaching the model undescribed.

Registration here passes no annotations, and an unannotated tool classifies as write (DESIGN.md §10.9). That matters the moment approvals are in play: a function registered through this argument never matches an @destructive selector, so a tool you meant to gate runs ungated and nothing says so. Anything carrying annotations, interruptible=False or safe_to_retry wants its own registry.register(...) call and a registry= here, which is what a real integration writes anyway. This argument is for the plain case, and the plain case only. tenant: the Scope every call in this Session defaults to. "local" is a placeholder that suits a single-tenant script; a consumer serving end users passes the real one per call to ask(), since one Session serves any number of tenants. principal: who is acting, recorded on every Record and passed to Policy. Psych never interprets it. concurrency: how many Runs the Worker executes at once. runtime_options: anything else Runtime takes -- policy, approval_selectors, telemetry, sandbox, memory, mcp, a2a, blob, http, prices. Passed straight through, so this helper never becomes the thing that decides which Runtime options exist.

publish

function

async def publish(store: 'Store', spec: 'Spec', *, context: 'ValidationContext | None' = None) -> 'Version'

Validate a Spec, content-hash it, and store it as an immutable Version.

Republishing an identical Spec returns the existing Version rather than creating a duplicate, which is what makes redeploying on every boot harmless.

Validation happens here and never at run time (DESIGN.md §4). A customer waiting on a response is not the right place to discover a typo, so a Spec naming an unregistered tool or a dangling skill link is refused now.

Args: store: where the Version is kept. spec: the Spec to publish. context: what exists at publish time: registered tool names, known models, reachable MCP servers. Omitted means structural checks only, which is right for an offline validation pass and wrong for a real deployment.

Returns: The Version. The same one every time for the same Spec.

Raises: SpecValidationError: carrying every problem found, each naming its path.

dispatch

function

async def dispatch(store: 'Store', version: 'Version | VersionHash', scope: 'Scope', *, input: 'dict[str, Any] | None' = None, idempotency_key: 'str | None' = None, deadline_seconds: 'float | None' = None, continues: 'RunId | None' = None) -> 'Dispatched'

Admit a Run, exactly once per idempotency key.

The only trigger Psych provides. The consumer owns the clock: their CronJob, their queue consumer, their webhook handler calls this (DESIGN.md §20).

Args: continues: pass the Run a second chat message follows to make this new Run see that earlier exchange (DESIGN.md §23.3). Refused with AccessDenied when that Run belongs to a different Scope -- a continuation is a new way to reach another Run's content and gets the same scrutiny as the MCP pool key (DESIGN.md §10.4).

stream

function

def stream(store: 'Store', run_id: 'RunId', *, after: 'int' = 0, scope: 'Scope | None' = None) -> 'AsyncIterator[Record]'

Every Record after after, then tail until the Run settles.

A client reconnecting passes the highest sequence it already has and misses nothing, including across an interrupt. The log is the stream, so there is no separate stream state that could break (DESIGN.md §12).

Args: scope: whose read this is. Passing it refuses a Run belonging to another tenant -- see psych_runtime.records.

Raises: RunNotFound: no such Run. Raised before the first record rather than tailing an id nobody ever admitted, which is what this used to do: a typo'd id produced a stream that never yielded and never ended. AccessDenied: scope names a different tenant than the Run's.

stream_text

function

def stream_text(store: 'Store', run_id: 'RunId', *, after: 'int' = 0, scope: 'Scope | None' = None) -> 'AsyncIterator[str]'

Just the assistant's words, for a chat UI that only renders words.

psych_runtime.stream() stays the complete truth and this is a projection over it, exactly as answer() is a projection over the same log. It ships for the reason answer(), status() and thread() ship: every consumer writes this loop, and every one of them gets the same three things wrong the first time. Assistant text arrives on model_call_finished rather than on a record named for text; an abort is a Record rather than an exception; and the iterator ends when the Run settles rather than when the model stops talking.

It refuses to swallow what it does not understand. A Run that aborts raises, a Run that fails raises, and a Run that settles any way other than COMPLETED raises. Yielding nothing and returning cleanly would be the bug this exists to prevent: a UI showing a blank reply and no error, for a Run whose log says exactly what went wrong. Someone who wanted only the words gets the words; a Run that goes wrong stays impossible to miss.

async for delta in psych_runtime.stream_text(store, run_id, scope=scope):
    yield f"data: \{delta}

"

Args: after: the highest sequence already seen, for a reconnect. Text before it is not replayed, which is what a client resuming a rendered stream wants; a client that needs the whole reply from the start passes 0 or reads answer() instead. scope: whose read this is. Passing it refuses a Run belonging to another tenant, the same as every other read here.

Raises: RunNotFound: no such Run. AccessDenied: scope names a different tenant than the Run's. RunAborted: the Run was interrupted or passed its deadline, carrying the terminal state so a caller can tell a user's stop from a timeout. RunFailed: the Run settled FAILED, carrying the failure from the log.

status

function

async def status(store: 'Store', run_id: 'RunId', *, scope: 'Scope | None' = None) -> 'RunStatus'

What a Run is doing right now, shaped for a person to be shown.

psych_runtime.state returns the reducer's own working dataclass, which carries bookkeeping that exists to make the next fold cheap (open_tool_calls, repeat_counts, approval_decisions) and is not JSON. Every consumer with a UI therefore wrote a projection of it, and the one in this repository's own example had to decide field by field what a status even is. This is that projection, as a frozen Pydantic model that serialises directly: the lifecycle, what it is waiting on and the exact call awaiting approval, the turn and step counts against their budgets, and the head sequence a stream should reconnect from.

state

function

async def state(store: 'Store', run_id: 'RunId', *, scope: 'Scope | None' = None) -> 'RunStateView'

Fold a Run's log into its current state.

Cheaper than a report and enough for "is it done, and what is it waiting on". psych_runtime.status is the same answer shaped for a UI; the report is the full projection.

Args: scope: whose read this is. Passing it refuses a Run belonging to another tenant -- see psych_runtime.records.

Raises: RunNotFound: no such Run. AccessDenied: scope names a different tenant than the Run's. CorruptLog: the log could not have been produced by the protocol.

answer

function

async def answer(store: 'Store', run_id: 'RunId', *, scope: 'Scope | None' = None) -> 'AnswerView'

One Run split into what it concluded and how it got there.

The other way to read a Run. psych_runtime.thread() gives the conversation in order, which is what a transcript is; this gives the answer with the work behind it, which is what a person who asked a question wants. Both project the same log, so neither can show something the other denies.

The split is derived rather than decided: the answer is the turn the loop itself finished on, and the work is everything before it (see psych_runtime.core.answer). Nothing asks the model to classify its own question, which is a judgement it is bad at and does not need to make.

Raises: RunNotFound: no such Run. AccessDenied: scope names a different tenant than the Run's.

thread

function

async def thread(store: 'Store', run_id: 'RunId', *, scope: 'Scope | None' = None, limit: 'int | None' = None) -> 'ThreadView'

One whole conversation, across every Run in its chain.

A second message starts a new Run (psych_runtime.runtime.thread argues why), so one Run's messages are one exchange rather than the conversation a person sees. This walks continues_run_id back to the Run that opened the thread and projects each one, oldest first, with every message carrying its Run id, sequence and timestamp.

Distinct from psych_runtime.runtime.thread.load_thread_history, which answers a different question: what to replay into the model, bounded by Limits.max_history_records because replaying a long chat is the dominant cost of a chat agent. A person scrolling back expects their whole conversation, not the slice the model was shown.

Args: scope: whose read this is. Every Run in the chain must share its tenant; the walk stops rather than crossing one (DESIGN.md §10.4 calls a link that can silently cross a tenant boundary the mistake that ends the project, and a control enforced at write time but not at read time is worse than none). limit: at most this many Runs, counting back from run_id. None walks the whole chain.

Raises: RunNotFound: no such Run. AccessDenied: scope, or an ancestor's own Scope, names a different tenant.

report

function

async def report(store: 'Store', run_id: 'RunId', *, child_depth: 'int' = 0, scope: 'Scope | None' = None) -> 'RunReport'

Everything a Run did, as a typed object.

The Spec version and hash, the system prompt as sent, every step, tool call and model call in order, usage split by cache state, computed cost, the latency breakdown, every suspension and resume, and the terminal state.

This ships with Psych rather than being left to consumers because it is the most visible thing they get on day one, and if every consumer writes their own projection they will each get the token arithmetic wrong in a different way (DESIGN.md §13.4).

Args: store: where the log lives. run_id: the Run to report on. child_depth: how many levels of nested Run to include. 0 reports this Run only. Bounded rather than unlimited so a deep delegation tree cannot read the whole database by accident. scope: whose read this is. Passing it refuses a Run belonging to another tenant -- see psych_runtime.records.

records

function

async def records(store: 'Store', run_id: 'RunId', *, after: 'int' = 0, limit: 'int | None' = None, scope: 'Scope | None' = None) -> 'Sequence[Record]'

The raw log. For a consumer building their own projection.

Args: scope: whose read this is. Passing it refuses a Run belonging to another tenant; omitting it reads any Run by id, which is right for an operator's own tooling and wrong for anything serving end users.

resume

function

async def resume(store: 'Store', run_id: 'RunId', *, payload: 'dict[str, Any] | None' = None, approved: 'bool | None' = None, by: 'str | None' = None) -> 'None'

Deliver a decision or payload to a suspended Run.

Approvals, clarifying questions and webhook waits are one mechanism rather than three, because they all run through the same lease and log machinery (DESIGN.md §11).

Args: by: who decided, as the consumer identifies people. Recorded on the resumed record and interpreted by nothing (DESIGN.md §14 keeps identity out of the library) -- but an approval of a destructive call whose log cannot say who approved it is not an audit trail, so there has to be somewhere to put the answer.

Raises: RunNotFound: no such Run. RunNotSuspended: the Run is not waiting for anything. SuspensionExpired: the decision arrived after the suspension's own expiry. The Run is settled ABANDONED first, so a stale approval never executes (DESIGN.md §11).

send

function

async def send(store: 'Store', run_id: 'RunId', *, message: 'dict[str, Any] | str', queue: 'QueueKind' = <QueueKind.STEER: 'steer'>) -> 'str'

Put a message into a Run that is still executing.

DESIGN.md §9's three queues, finally reachable from outside a test: steer the turn running right now, follow up once it settles, or hand a message to whatever Run comes next -- see psych_runtime.runtime.dispatch.send for what each one is for and when each is refused. For "a second chat message after the first has already settled", this is not what you want; that is dispatch(continues=run_id, ...), which starts a fresh Run instead of trying to inject into one that is no longer there to receive it.

Returns: The queue entry's id.

interrupt

function

async def interrupt(store: 'Store', run_id: 'RunId', *, reason: 'str' = '', by: 'str | None' = None) -> 'None'

Stop a Run.

An abort is a Record, not a flag. A message sent immediately afterwards lands in the next-Run queue and both are visible in the log in order, which is what makes "stop and send another" a modelled transition rather than a race (DESIGN.md §9).

agent

function

def agent(name: 'str', *, registry: 'ToolRegistry | None' = None) -> 'AgentBuilder'

Start building an Agent Spec named name.

workflow

function

def workflow(name: 'str', *, registry: 'ToolRegistry | None' = None) -> 'WorkflowBuilder'

Start building a Workflow Spec named name.