Embed and stream a live session

Start a Run inside a service, stream events, and receive a later human decision through Desk.

Use Session.start(role, message) when a service must acknowledge work before the conversation finishes. It returns a live Run; the host keeps that handle, exposes its transcript and events, and owns the surrounding request state.

Use run() or ask() when the caller can wait for the final result. A live run adds status, cancellation, persistence, and reconnection work.

Start and observe

import asyncio

import agentsparty as ap

Reporter, Analyst = ap.roles('Reporter', 'Analyst')


class Incident(ap.Message):
    """The incoming service request."""

    text: str


class Report(ap.Message):
    """The completed report."""

    text: str


async def main() -> None:
    session = ap.Session(
        ap.Protocol(
            ap.send(Reporter, Analyst, Incident),
            ap.send(Analyst, Reporter, Report),
        ),
        {
            Analyst: ap.Agent(
                ap.replies([Report('ready')]),
                'summarise',
            )
        },
        allowance=ap.Allowance(steps=8),
    )
    run = session.start(Reporter, Incident('high latency'))
    signals = []
    async for event in run.events():
        signals.append(type(event.signal).__name__)
    transcript = await run.result()


asyncio.run(main())

Run.done reports completion, Run.transcript exposes delivered envelopes so far, and await run.result() returns the final transcript or raises the typed failure. await run.cancel() stops remaining protocol work; it does not undo an external effect that already completed.

Connect a human with Desk

Desk is a queue-backed HumanIo. A service can show desk.offered in a status response and later call desk.submit(message) when an operator chooses. The submitted class must be on the current offer.

import asyncio

import agentsparty as ap

Reporter, Approver = ap.roles('Reporter', 'Approver')


class Ticket(ap.Message):
    """A ticket waiting for approval."""

    text: str


class Accept(ap.Message):
    """Accept the ticket."""


async def main() -> None:
    desk = ap.Desk()
    session = ap.Session(
        ap.Protocol(
            ap.send(Reporter, Approver, Ticket),
            ap.choose(Approver, Reporter, {Accept: []}),
        ),
        {Approver: ap.Human(desk)},
    )
    run = session.start(Reporter, Ticket('timeout'))
    while not desk.offered:
        await asyncio.sleep(0.001)
    await desk.submit(Accept())
    await run.result()


asyncio.run(main())

Service boundary

A typical HTTP adapter has three operations: start a run, read its current status, and submit an offered human decision. Store the Run and Desk under an application-owned identifier. Authentication, durable ticket storage, idempotency, process recovery, and cross-host coordination stay outside the protocol runtime.

For recovery, persist the application id, SessionId, protocol version, and journal location. An in-memory Run object is not a durable queue record.

Sessions, execution, and limits: choose run, ask, or start.
Journals, replay, and tracing: persist decisions and inspect events.
Roles, participants, and models: use Human and Desk.
Guarantees and limitations: review the in-process boundary.

Embed and stream a live session

Start and observe

Connect a human with Desk

Service boundary

Related

On this page