Overview

A multiparty conversation is a typed value: named roles, labelled messages, branches, and loops. agentsparty projects that value onto each role, refuses a protocol a role cannot observe, and runs the session in-process. Participants fill declared payloads and pick declared branches.

pip install 'agentsparty[openai]'
export OPENAI_API_KEY=...

A session

Declare the conversation first. render prints it before anyone runs.

import agentsparty as ap
from agentsparty.protocol import msg, render

Writer, Reader = ap.roles('Writer', 'Reader')
Note = ap.Text('Note')
protocol = msg[Writer, Reader](Note)
print(render(protocol))

Writer -> Reader : Note(str)
end

Bind gpt-5.6-luna to each role and pass the participants to AgentRuntime:

import asyncio

from openai import AsyncOpenAI
import agentsparty as ap
from agentsparty.protocol import msg

Writer, Reader = ap.roles('Writer', 'Reader')
Note = ap.Text('Note')
protocol = msg[Writer, Reader](Note)

model = ap.OpenAIModel(
    'gpt-5.6-luna',
    AsyncOpenAI(max_retries=0, timeout=30.0),
)
writer = ap.Agent(
    model,
    Writer,
    'Send a concise note.',
    protocol,
)
reader = ap.Agent(
    model,
    Reader,
    'Receive the note.',
    protocol,
)


async def main() -> None:
    runtime = ap.AgentRuntime(protocol, [writer, reader])
    envelope = (await runtime.run())[0]
    print(
        f'{envelope.sender.name} -> {envelope.receiver.name}: '
        f'{envelope.label}({envelope.payload!r})'
    )


asyncio.run(main())

AgentRuntime projects the protocol and requires exactly one participant per role. The writer authors Note; the reader takes the envelope.

Combinators

msg[S, R](payload) is one edge and >> sequences. alt[chooser, informed] offers labelled continuations, and rec / var bind a guarded loop. par runs tracks whose role sets are disjoint; it has no join value.

import agentsparty as ap
from agentsparty.protocol import alt, msg, par, rec, render, var

W, V, A, B, C, D = ap.roles('W', 'V', 'A', 'B', 'C', 'D')

review = msg[W, V](ap.Text('Draft')) >> alt[V, W](
    ap.Nothing('Approve'),
    ap.Text('Reject'),
)
poll = rec(
    'Poll',
    msg[A, B](ap.Text('Tick')) >> var('Poll'),
).close()
split = par(
    msg[A, B](ap.Text('L')),
    msg[C, D](ap.Text('R')),
).close()

print(render(review))
print(render(poll))
print(render(split))

Projection

A role whose next action changes across branches must observe the choice. project checks that before any participant is asked.

import agentsparty as ap
from agentsparty.protocol import alt, msg, project

A, B, C = ap.roles('A', 'B', 'C')
broken = alt[A, B](
    ap.Nothing('Yes') >> msg[A, C]('Y', ap.Text),
    ap.Nothing('No') >> msg[C, A]('N', ap.Text),
).close()
try:
    project(broken, C)
except ap.ProjectionError as err:
    print(type(err).__name__)

Send a branch-specific message to every role whose continuation differs. Knowledge of alt walks through the repair.

Payloads

A label names the operation; a codec names the value. Decode runs at the boundary, so handlers receive a value that already satisfies the predicate.

import agentsparty as ap
from agentsparty.kernel.errors import PayloadError

Positive = ap.Integer.where('positive', lambda n: n > 0)
print(Positive.decode(3))
try:
    Positive.decode(-1)
except PayloadError as err:
    print(type(err).__name__)

Draft = ap.record('Draft', title=str, words=int)
print(Draft.decode({'title': 'v1', 'words': 12}))

Text, Integer, Flag, list_of, one_of, and record compose. Use json_model when a schema and a parser already exist.

Cast

Cast is the same binding written as a pipeline. Each play attaches a factory to one projected endpoint; run_sync demands every role is played.

from openai import AsyncOpenAI
import agentsparty as ap
from agentsparty.protocol import msg

Writer, Reader = ap.roles('Writer', 'Reader')
protocol = msg[Writer, Reader](ap.Text('Note'))
model = ap.OpenAIModel(
    'gpt-5.6-luna',
    AsyncOpenAI(max_retries=0, timeout=30.0),
)
trace = (
    ap
    .Cast(protocol)
    .play(Writer, ap.agent(model, 'Send a concise note.'))
    .play(Reader, ap.agent(model, 'Receive the note.'))
    .run_sync()
)
envelope = trace[0]
print(
    f'{envelope.sender.name} -> {envelope.receiver.name}: {envelope.label}({envelope.payload!r})'
)

Participants

Agent lets a model author, Human waits at a console, and Machine computes a Choice from the envelopes it has seen. service answers a request the protocol already typed.

import agentsparty as ap
from agentsparty.machine import machine
from agentsparty.protocol import msg

Lead, Worker = ap.roles('Lead', 'Worker')
Task = ap.Text('Task')
protocol = msg[Lead, Worker](Task)


def assign(view: ap.View) -> ap.Choice:
    return ap.says(Task, 'summarise the brief')


def unused(view: ap.View) -> ap.Choice:
    raise RuntimeError('Worker only receives')


trace = (
    ap
    .Cast(protocol)
    .play(Lead, machine(assign))
    .play(Worker, machine(unused))
    .run_sync()
)
print(trace[0].payload)

from openai import AsyncOpenAI
import agentsparty as ap
from agentsparty.protocol import alt

Asker, Tools = ap.roles('Asker', 'Tools')
Search = ap.Text('search')
Hits = ap.Text.many()('hits')
protocol = alt[Asker, Tools](
    Search >> alt[Tools, Asker](Hits),
)
model = ap.OpenAIModel(
    'gpt-5.6-luna',
    AsyncOpenAI(max_retries=0, timeout=30.0),
)


async def search(query: str) -> ap.Choice:
    return ap.reply(Hits, [query])


trace = (
    ap
    .Cast(protocol)
    .play(Asker, ap.agent(model, 'Search once, then stop.'))
    .play(Tools, ap.service(ap.tool_for(Search, search)))
    .run_sync()
)
print([envelope.label.name for envelope in trace])

A custom agentsparty.participant.Participant implements select, offer, recall, and cancel against the same projected endpoint.

Choreography

@choreography records the same session protocol from Python statements. c.say, c.decide, c.loop, c.times, and c.parallel are the operators.

import agentsparty as ap
from agentsparty.protocol import equal_session, msg, render

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


@ap.choreography
def note(c: ap.Chor) -> None:
    c.say(Writer, Reader, 'Note')


print(render(note()))
print(equal_session(note(), msg[Writer, Reader]('Note').close()))

Routines

A Routine is a named fragment with formal roles. do binds actual roles and returns a fragment you can sequence further.

import agentsparty as ap
from agentsparty.protocol import Routine, do, msg, render

Sender, Receiver = ap.roles('Sender', 'Receiver')
review = Routine(
    'review',
    (Sender, Receiver),
    msg[Sender, Receiver](ap.Text('Draft')),
)
Writer, Editor = ap.roles('Writer', 'Editor')
print(render(do(review, Writer, Editor).close()))

Composition

compose(contract, components) builds one session protocol from components that fit a shared interface. Internal role sets must be disjoint.

import agentsparty as ap
from agentsparty.protocol import compose, msg, owning, render

Brand, Analyst, Photo = ap.roles('Brand', 'Analyst', 'Photo')
Product = ap.Text('Product')
Copy = ap.Text('Copy')
Post = ap.Text('Post')
text = owning(Analyst).defining(
    msg[Brand, Analyst](Product) >> msg[Analyst, Photo](Copy),
)
image = owning(Photo).defining(
    msg[Analyst, Photo](Copy) >> msg[Photo, Brand](Post),
)
contract = (
    msg[Brand, Analyst](Product)
    >> msg[Analyst, Photo](Copy)
    >> msg[Photo, Brand](Post)
).close()
print(render(compose(contract, [text, image])))

Resume

A journal stores authored decisions. Replay restores them instead of asking the model again. JsonlJournal and SqliteJournal persist to disk.

from pathlib import Path

from openai import AsyncOpenAI
import agentsparty as ap
from agentsparty.protocol import msg

Writer, Reader = ap.roles('Writer', 'Reader')
protocol = msg[Writer, Reader](ap.Text('Note')).close()
model = ap.OpenAIModel(
    'gpt-5.6-luna',
    AsyncOpenAI(max_retries=0, timeout=30.0),
)
journal = ap.JsonlJournal(Path('session.jsonl'), protocol)
trace = (
    ap
    .Cast(protocol)
    .play(Writer, ap.agent(model, 'Send a concise note.'))
    .play(Reader, ap.agent(model, 'Receive the note.'))
    .run_sync(journal=journal)
)
print(trace[0].label.name, len(journal.script().decisions))

Observation

A tracer records events for inspection. StreamTracer delivers them as they occur; events carry a facet (runtime, protocol, model, tool). debug.Report prints a protocol, a conversation, or counted facts to a Console.

from openai import AsyncOpenAI
import agentsparty as ap
from agentsparty.protocol import msg

Writer, Reader = ap.roles('Writer', 'Reader')
protocol = msg[Writer, Reader](ap.Text('Note'))
model = ap.OpenAIModel(
    'gpt-5.6-luna',
    AsyncOpenAI(max_retries=0, timeout=30.0),
)
tracer = ap.MemoryTracer()
(
    ap
    .Cast(protocol)
    .play(Writer, ap.agent(model, 'Send a concise note.'))
    .play(Reader, ap.agent(model, 'Receive the note.'))
    .run_sync(tracer=tracer)
)
print(len(tracer.events))

Budget

Allowance bounds protocol steps and recursion unfolds for one run. Replayed journal decisions do not consume it. The first failure cancels every bound participant.

from openai import AsyncOpenAI
import agentsparty as ap
from agentsparty.protocol import msg

Writer, Reader = ap.roles('Writer', 'Reader')
protocol = msg[Writer, Reader](ap.Text('Note'))
model = ap.OpenAIModel(
    'gpt-5.6-luna',
    AsyncOpenAI(max_retries=0, timeout=30.0),
)
try:
    ap.AgentRuntime(
        protocol,
        [
            ap.Agent(model, Writer, 'Send a note.', protocol),
            ap.Agent(model, Reader, 'Receive the note.', protocol),
        ],
        allowance=ap.Allowance(steps=0),
    ).run_sync()
except ap.StepLimitError as err:
    print(type(err).__name__)

Deadline closes the current waiting window.

Models

OpenAIModel speaks the OpenAI Responses API. Wrappers nest: Retrying retries ModelUnavailable, fallback tries the next model, Metered refuses the next call once receipts pass a cap.

from openai import AsyncOpenAI
import agentsparty as ap

client = AsyncOpenAI(max_retries=0, timeout=30.0)
primary = ap.OpenAIModel('gpt-5.6-luna', client)
backup = ap.OpenAIModel('gpt-5.6-luna', client)
model = ap.fallback(
    ap.Retrying(primary, attempts=2),
    ap.Metered(backup, tokens=8_000),
)
print(type(model).__name__)

Install and quickstart
Core Concepts — session protocols, combinators, projection, participants
Tutorials — build a coding-agent harness
What you can rely on — the 0.1.x boundary

Research framework at 0.1.x. Exception types are stable; message text and journal formats are not.