Protocols and composition

Protocol describes the global conversation before any participant takes a role. Its constructors assemble interactions, choices, loops, and concurrency into a closed session type; projection then derives the endpoint visible to each role, and Session binds a participant to every endpoint.

import agentsparty as ap

Author, Reviewer = ap.roles('Author', 'Reviewer')


class Draft(ap.Message):
    """The initial draft."""

    text: str


class Feedback(ap.Message):
    """Review comments."""

    text: str


protocol = ap.Protocol(
    ap.send(Author, Reviewer, Draft),
    ap.send(Reviewer, Author, Feedback),
)
print(ap.render(protocol))

Author -> Reviewer : Draft(str) :: 'The initial draft.'
Reviewer -> Author : Feedback(str) :: 'Review comments.'
end

One conversation, three layers

Keep these values separate while designing a session:

The global value is the contract. Projection derives a local endpoint from it, so there is no second workflow that can drift. The runtime checks the participant against that endpoint; a model or handler cannot make an invalid topology valid by returning a clever value.

Steps

An empty continuation list ends the protocol on that branch. One list level is flattened: Protocol(send(...), [send(...), send(...)]). A protocol fragment is an ordinary function that returns ap.Step or list[ap.Step].

Read a protocol as a decision tree

Read send(A, B, M) as an obligation: A authors an instance of M and B receives it. Read choose(A, B, branches) as two obligations at once: A authors one branch label and B learns which continuation became active. Every role not named in the choice must still distinguish the resulting duties; if it cannot, projection rejects the protocol.

Loops add a return edge, not an instruction to run forever. A branch that omits repeat() leaves the loop. Allowance supplies the runtime ceiling for a loop that keeps repeating, while the protocol describes the possible exits.

When to write a custom protocol

Start with a pattern when its topology already matches the work. Write a Protocol when messages carry business data, branches have different owners, or a human and a service share a decision. Keep the first version small: one opening message, one clear reply, then a choice or loop only when it earns its place. A protocol that is hard to render is usually hard to review.

Before binding participants, ask these questions:

• Which role authors each message, and which role receives it?
• Which roles must observe a branch before their next duty changes?
• Where does each loop stop, and which allowance limits its unfolding?
• Which side effect is represented by a message, and which remains application code?

choose with branches

choose sends the selected class to the hearer and follows only that class's continuation. A branch with an empty list ends that branch.

import agentsparty as ap

Requester, Reviewer, Archive = ap.roles('Requester', 'Reviewer', 'Archive')


class Yes(ap.Message):
    """Continue with a report."""


class No(ap.Message):
    """Stop the exchange."""


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

    text: str


protocol = ap.Protocol(
    ap.choose(
        Requester,
        Reviewer,
        {
            Yes: [ap.send(Requester, Archive, Report)],
            No: [],
        },
    ),
)
rendered = ap.render(protocol)
print(rendered)

par and loops

par runs independent branches side by side. Branches must own disjoint roles; there is no join, so putting another step after par(...) is a composition error. repeat() belongs inside the nearest loop; outside one, construction raises ProtocolError rather than creating an ambiguous continuation.

Loops describe possible repetition; they do not promise progress. A branch that omits repeat() exits the nearest loop, while Allowance.unfoldings bounds a run whose participant keeps selecting a repeating branch.

par is intentionally terminal. If later work needs results from two branches, represent that aggregation as another explicit message exchange rather than an implicit scheduler join.

import agentsparty as ap

Planner, Writer, Auditor, Publisher = ap.roles(
    'Planner', 'Writer', 'Auditor', 'Publisher'
)


class Ping(ap.Message):
    """A ping."""


class Pong(ap.Message):
    """A pong."""


parallel = ap.Protocol(
    ap.par(
        ap.send(Planner, Writer, Ping),
        ap.send(Auditor, Publisher, Pong),
    ),
)
ap.render(parallel)
try:
    ap.Protocol(ap.repeat())
except ap.ProtocolError:
    print('repeat needs a loop')

A function is a fragment

import agentsparty as ap

Planner, Workspace, Client = ap.roles('Planner', 'Workspace', 'Client')


class Looking(ap.Message):
    """Still reading."""


class Listing(ap.Message):
    """Paths under a directory."""

    paths: list[str]


class List(ap.Message):
    """A directory to list."""

    path: str


def still_reading() -> list[ap.Step]:
    return [ap.send(Planner, Client, Looking), ap.repeat()]


protocol = ap.Protocol(
    ap.loop(
        ap.choose(
            Planner,
            Workspace,
            {
                List: [
                    ap.send(Workspace, Planner, Listing),
                    *still_reading(),
                ],
            },
        ),
    ),
)
ap.render(protocol)

Definition and load

Protocol.id fingerprints the shape. Two protocols are equal when their shapes are structurally the same. ap.definition / ap.load round-trip the portable data, which is useful when a protocol is stored beside a journal.

import agentsparty as ap

Author, Editor = ap.roles('Author', 'Editor')


class Note(ap.Message):
    """A note."""

    text: str


protocol = ap.Protocol(ap.send(Author, Editor, Note))
rebuilt = ap.load(ap.definition(protocol).to_dict())

From protocol to session

Projection happens while Session is constructed; the cast then supplies one participant for every projected role.

import agentsparty as ap

Reader, Writer = ap.roles('Reader', 'Writer')


class Ask(ap.Message):
    """A question."""

    text: str


class Answer(ap.Message):
    """An answer."""

    text: str


session = ap.Session(
    ap.Protocol(
        ap.send(Reader, Writer, Ask),
        ap.send(Writer, Reader, Answer),
    ),
    {
        Reader: ap.Script([Ask('why')]),
        Writer: ap.Agent(ap.replies([Answer('because')]), ''),
    },
)
session.run()

Related

• agentsparty.Protocol: construct closed protocol values.
• Messages: define the classes that steps carry.
• Projection and branch visibility: understand the role-local checks.
• Sessions, execution, and limits: bind participants and execute.