The public API

This is the documentation for unreleased changes on main. For the released library, read v0.1.

Everything a consumer calls is on the psych_runtime module. Anything reached through a submodule path is internal and moves without ceremony while the package is 0.x.

Psych is a library, not a framework. You call these; Psych calls back only through ports you supplied.

The short way in

psych_runtime.session() assembles a store, a tool registry, a Runtime and a running Worker, and stops them again on the way out. It is the wiring below with the parts a first-time reader has no opinion about yet already decided:

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

s.store, s.registry, s.runtime and s.worker are the real objects, and reaching past them is the expected path out rather than a failure of the helper. Move to the long form below as soon as your Workers run in their own process, each tenant needs its own model client, or you want a Worker fleet. There is one execution model underneath either way.

The calls

Every read above also takes an optional scope=. Pass it when you serve end users: a Run belonging to another tenant is then refused rather than returned, which is what keeps one leaked run id from being a readable conversation (DESIGN.md §14).

Plus what you construct rather than call, all exported from psych_runtime itself: Worker, which claims and executes Runs; Runtime, which tells the Worker how; ToolRegistry, HttpTransport, OpenAICompatibleClient, McpPool, McpTools and DEFAULT_PRICES.

Everything you can hold is nameable from psych_runtime too, which is what makes "the public API is this module" true rather than aspirational: the Spec models and their step types, every report and status type, the id aliases (RunId, VersionHash, ToolCallId), the enums (TerminalState, SuspendReason, ToolOutcome, Lifecycle), and the ports you implement (Store, BlobStore, ModelClient, Policy, SecretResolver, Telemetry, Sandbox, PriceResolver, EgressPolicy). A unit test asserts that no exported model has a field whose type is unreachable from the module, so this cannot quietly drift back. The two submodule imports that stay routine are the store adapter you chose and the test helpers.

One name is deliberately absent. psych_runtime.store.memory.InMemoryStore (the Store adapter) and psych_runtime.memory.port.MemoryStore (the durable-facts port) are different types with the same name, so neither is exported under it; import whichever you mean from its own module.

A worked example

Everything below runs. It is the shape of a real integration with the parts a consumer supplies made explicit.

import asyncio

import psych_runtime
from psych_runtime.store.postgres import PostgresStore

# 1. Register your tools. The functions stay yours; Psych keeps the names.
registry = psych_runtime.ToolRegistry()


@registry.register(annotations={"read-only"})
async def lookup_order(order_id: str) -> dict[str, str]:
    """Look up an order by its id."""
    return {"order_id": order_id, "status": "shipped"}


@registry.register(interruptible=False, annotations={"destructive"})
async def issue_refund(order_id: str, cents: int) -> str:
    """Refund an order. Not interruptible: a half-issued refund is worse than a
    slow stop."""
    return f"refunded {cents} on {order_id}"


# 2. Describe the agent. This is data. It has no callables in it.
spec = psych_runtime.AgentSpec(
    name="support",
    instructions="Help the customer with their order. Be brief.",
    model=psych_runtime.ModelRef(model="gpt-4o", temperature=0.2),
    tools=(
        psych_runtime.CodeTool(name="lookup_order"),
        psych_runtime.CodeTool(name="issue_refund", interruptible=False),
    ),
    limits=psych_runtime.Limits(max_turns=12, deadline_seconds=300),
)


async def main() -> None:
    async with PostgresStore(dsn="postgresql://localhost/psych") as store:
        await store.migrate()

        # 3. Publish. Validation happens here, never mid-conversation.
        version = await psych_runtime.publish(
            store,
            spec,
            context=psych_runtime.ValidationContext(registered_tools=registry.names),
        )

        # 4. Run a Worker. Your process, Psych's loop.
        transport = psych_runtime.HttpTransport()
        runtime = psych_runtime.Runtime(
            store=store,
            model=psych_runtime.OpenAICompatibleClient(
                base_url="http://localhost:4000",  # any compatible gateway
                transport=transport,
                scope=psych_runtime.Scope(tenant="acme"),
            ),
            registry=registry,
            approval_selectors=("@destructive",),
        )
        worker = psych_runtime.Worker(store, runtime)
        worker_task = asyncio.create_task(worker.run())

        # 5. Admit a Run. Your webhook handler or queue consumer does this.
        scope = psych_runtime.Scope(tenant="acme", principal="user-42")
        run = await psych_runtime.dispatch(
            store,
            version,
            scope,
            input={"message": "where is order A1?"},
            idempotency_key="webhook-evt-8891",
        )

        # 6. Stream it to your user. Reconnect with after=N and miss nothing.
        async for record in psych_runtime.stream(store, run.run_id):
            if record.type == "model_call_finished" and record.text:
                print(record.text)

        # 7. Say what it cost.
        report = await psych_runtime.report(store, run.run_id)
        print(report.terminal_state, report.totals.usage, report.totals.cost)

        worker.stop()
        await worker_task


asyncio.run(main())

Tools whose shape is known only at runtime

registry.register is the documented default, and it stays that way: the schema comes from the function's own type hints, so the schema and the function cannot drift apart. Some consumers do not have a signature to point it at -- a tool generated from a database table of integrations, a non-MCP catalogue proxied at runtime, a customer-authored form definition. For those, build the arguments model yourself with pydantic.create_model and hand it to register_dynamic instead:

from pydantic import ConfigDict, create_model

integrations = {
    "lookup_salesforce_account": ("account_id", "Look up a Salesforce account by id."),
    "lookup_zendesk_ticket": ("ticket_id", "Look up a Zendesk ticket by id."),
}


async def call_integration(name: str, **arguments: str) -> dict[str, str]:
    return {"tool": name, **arguments}


for tool_name, (field, description) in integrations.items():
    arguments_model = create_model(
        f"{tool_name}_Arguments",
        __config__=ConfigDict(extra="forbid"),  # required: see below
        **{field: (str, ...)},
    )
    registry.register_dynamic(
        tool_name,
        functools.partial(call_integration, tool_name),
        arguments_model,
        description=description,
        annotations={"read-only"},
    )

This produces the exact same RegisteredTool the decorator does -- the resolver, the access-narrowing intersection and the failure-streak guard read one shape and cannot tell which door a tool came through. Two things it holds to the same standard as the derived path, deliberately, rather than being the looser way in:

• A description is required. A model chooses tools by their descriptions, so a schema with none is refused exactly like a docstring-less function.
• extra="forbid" is required, and checked rather than silently added. create_model defaults to ignoring fields it does not recognise, which would turn a model's hallucinated argument into a silently dropped field instead of a validation failure the model is told about. A model built without ConfigDict(extra="forbid") is rejected at registration rather than rebuilt behind the scenes, because rebuilding it from bare field info would drop anything not representable that way -- a @field_validator, a @model_validator, a computed field.

What you supply, and what Psych refuses to

Psych takes these through ports because you already have them, and because owning them would make Psych a platform rather than a library:

And these it will not grow, whatever the request: an HTTP server, an auth system, orgs and roles, a scheduler, a UI, a prompt library, an eval framework, a vector store or RAG, budget enforcement, or a model router.

Calling it from synchronous code

Every entry point is a coroutine, and that is a decision rather than an omission. There is no psych_runtime.sync facade and there will not be one.

The reason is not purity. A sync facade does not solve the problem it exists for. asyncio.run() builds a fresh event loop per call, and an asyncpg pool binds to the loop that created it. A wrapper over the read paths would therefore either open a pool per call, which is catastrophic under load, or hand back a pool bound to a dead loop, which fails in a way that reads as a Psych bug. That moves the deadlock a consumer would have written themselves into the library, where it is harder to see and ours to support.

A facade over Worker would be worse and is refused outright. A Worker holds a lease and renews it on a timer; wrapping it synchronously is a lease that stops being renewed the moment the caller blocks.

You own the loop. Both frameworks people ask about have an established answer.

Django

asgiref ships with Django and already solves this. async_to_sync runs the coroutine in a loop it manages, and SyncToAsync goes the other way for ORM calls inside a tool.

from asgiref.sync import async_to_sync, sync_to_async

import psych_runtime


def start_chat(request):  # an ordinary sync view
    dispatched = async_to_sync(psych_runtime.dispatch)(
        STORE, VERSION, scope_for(request), input={"message": request.POST["message"]}
    )
    return JsonResponse({"run_id": dispatched.run_id})


@registry.register(annotations={"read-only"})
async def lookup_order(order_id: str) -> dict[str, str]:
    """Look up an order by its id."""
    order = await sync_to_async(Order.objects.get)(pk=order_id)
    return {"order_id": order_id, "status": order.status}

Run the Worker as a management command with asyncio.run(), in its own process. It is long-lived and owns its loop for its whole life, which is the shape it wants anyway.

Celery, and anything with a worker process

Do not call asyncio.run() per task. Give the process one loop and keep it:

import asyncio
import threading

_loop = asyncio.new_event_loop()
threading.Thread(target=_loop.run_forever, daemon=True).start()


def run_sync(coro):
    """Run one coroutine on the process's own loop, from sync code.

    One loop for the life of the process, so a connection pool created on it
    stays valid across every task rather than being rebuilt or, worse, reused
    after its loop is gone.
    """
    return asyncio.run_coroutine_threadsafe(coro, _loop).result()


@app.task
def handle_message(tenant: str, message: str) -> str:
    dispatched = run_sync(
        psych_runtime.dispatch(STORE, VERSION, Scope(tenant=tenant), input={"message": message})
    )
    return dispatched.run_id

Ten lines, in your process, where you can see it. That is the whole reason this is documented rather than shipped: written here it is inspectable and yours; shipped it is a second public surface to keep in step forever, and the failure mode above is one we would be answering support questions about.

psych new writes a main.py with asyncio.run(main()) already in the right place, so the first thing a stranger sees is the async model in a file they own.

Triggers: you own the clock

Psych stores no cron expressions, evaluates no schedules and runs no timers (DESIGN.md §20). It provides dispatch, and your existing scheduler calls it. The refusal is the feature: a scheduler inside a library is one you cannot see during an incident and cannot reconcile against the one you already run.

Three shapes, each five lines.

A cron job. Idempotency keyed on the window, so a retrying CronJob or a double-fired timer produces one Run:

window = datetime.now(UTC).strftime("%Y-%m-%dT%H")
await psych_runtime.dispatch(
    store,
    version,
    Scope(tenant=tenant),
    input={"message": "run the hourly digest"},
    idempotency_key=f"digest:{tenant}:{window}",
)

A webhook handler. Keyed on the provider's event id, which is what makes an at-least-once delivery safe:

async def on_webhook(event: dict) -> None:
    await psych_runtime.dispatch(
        store,
        version,
        Scope(tenant=event["tenant"]),
        input={"message": event["text"]},
        idempotency_key=f"stripe:{event['id']}",
    )

A queue consumer. Same idea, keyed on the message id, so a redelivery after a visibility timeout does not start a second Run:

async def on_message(msg: Message) -> None:
    await psych_runtime.dispatch(
        store,
        version,
        Scope(tenant=msg.tenant),
        input=msg.payload,
        idempotency_key=msg.message_id,
    )
    await msg.ack()

Delivery is at-least-once and exactly-once does not exist. The key is what makes that survivable; without one, every redelivery is a new Run and a second refund.

Suspension and approvals

A Run that needs a human suspends rather than blocking. It releases its lease, persists, and waits. Nothing is held in memory, so the process that resumes it need not be the one that asked.

state = await psych_runtime.state(store, run_id)
if state.suspended and state.suspend_reason == "approval":
    # state.pending_approval_call_id names the call awaiting a decision.
    await psych_runtime.resume(store, run_id, approved=True)

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

Interrupts

await psych_runtime.interrupt(store, run_id, reason="the user pressed stop")
# A message sent immediately after starts a new Run carrying it. Both are in the
# log, in order, and a client streaming with after=N sees both.

An abort is a Record with a sequence number, so "what arrived after the stop" is a question the log answers rather than a race the runtime has to win.

Talking to other people's agents (A2A)

psych_runtime.a2a speaks A2A v1.0: the data model, both HTTP bindings' envelopes, version and extension negotiation, Agent Cards and their signatures, and the mapping from a Run to a Task. It contains no server, because A2A is a transport and this list has already refused to own one.

from psych_runtime.a2a import agent_card, task_of, context_id_of, AgentInterface

card = agent_card(spec, interfaces=[...], version="2026.4.1")
task = task_of(
    await psych_runtime.status(store, run_id),
    context_id=context_id_of(run_id, chain),
    answer=await psych_runtime.answer(store, run_id),
)

The mapping is pure and total: a Run's RunId is the taskId, its continuation chain is the contextId, a question is INPUT_REQUIRED, and psych_runtime.answer() is an Artifact. Your routes call it and decide nothing about the protocol; examples/playground/backend/app/a2a/ is a complete set of such routes to copy.

The other direction is a tool source. Declare a peer on the Spec and its skills appear in the prompt, narrowed exactly as MCP tools are:

spec = psych_runtime.AgentSpec(
    name="support",
    model=psych_runtime.ModelRef(model="gpt-4o"),
    a2a_peers=(psych_runtime.A2APeer(name="research", url="https://research.example.com"),),
)
runtime = Runtime(store=store, model=model, registry=registry, a2a=A2ATools(pool))

A2APool keys by (scope, peer, credential) and never by URL, for the reason McpPool does.

Testing your own agents

from psych_runtime.testing.fake_model import FakeModel
from psych_runtime.store.memory import InMemoryStore

model = (
    FakeModel()
    .turn(text="Checking.", tool_calls=[("lookup_order", {"order_id": "A1"})])
    .turn(text="A1 has shipped.")
)

The fake scripts malformed tool calls, stalled streams and streams that abort mid-token, so your tests can see the cases a real provider will eventually produce. psych_runtime.testing.logs.LogBuilder builds record logs directly when you want to test a projection without running anything.