Calls
The functions a consumer calls. This is the whole verb surface.
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.