agentsparty 0.2.0

Protocol-first orchestration for AI agents.
Declarative multiparty session protocols for AI agents

Inspired by Multiparty Session Types (MPST), this project focuses on declarative session protocols for AI agents; it is a practical protocol-oriented experiment, not a claim to implement MPST as a type system.

What it is

You describe one typed global conversation between named roles. agentsparty projects it onto every role, rejects the conversations a role could not follow, and runs the session. A language model fills a typed payload or picks a declared branch — it does not route the workflow, invent roles, or call arbitrary code. The protocol owns control; the model owns content.

Install

pip install agentsparty
pip install "agentsparty[openai]"   # optional OpenAI Responses backend

One agent

The smallest session is a request and a reply. ap.patterns.request_reply builds it, and ask plays the caller from your program — it sends the task and returns what comes back. Roles are declared with ap.roles(...): a typo is a NameError, not a new role.

import agentsparty as ap

Reader, Writer = ap.roles('Reader', 'Writer')
shape = ap.patterns.request_reply(Reader, Writer)
model = ap.model('openai:gpt-5.6-luna')

session = ap.Session(
    shape,
    {
        Writer: ap.Agent(model, 'Write three paragraphs.'),
    },
)

article = session.ask(
    Reader,
    shape.Ask('What does a multi-agent AI system mean?'),
    expect=shape.Answer,
)

print(article.text)

A session, end to end

A Requester hands a Writer a task. The Writer drafts an article and the Reviewer approves it or sends it back with a critique — a loop that ends the moment the Reviewer approves. ap.patterns.review_loop builds the shape. The cast is a dict: one participant spec per role. A dict comprehension is the bulk form when several roles share a model.

import agentsparty as ap

Requester, Writer, Reviewer = ap.roles(
    'Requester', 'Writer', 'Reviewer'
)
shape = ap.patterns.review_loop(Requester, Writer, Reviewer)
model = ap.model('openai:gpt-5.6-luna')
briefs = {
    Writer: 'Write the article. Revise it to address any critique.',
    Reviewer: (
        'Approve the draft once it is ready; otherwise Reject it with '
        'a short critique.'
    ),
}

session = ap.Session(
    shape,
    {
        role: ap.Agent(model, brief)
        for role, brief in briefs.items()
    },
)

article = session.ask(
    Requester,
    shape.Task('What does a multi-agent AI system mean?'),
    expect=shape.Final,
)
print(article.text)

The task travels as a message, never as the Writer's system prompt. On each rejected draft the Writer sends the Requester a notice, so the Requester can tell a still-running loop from the final article — the pattern inserts that notice and projects its shape before returning it.

ask only plays a role that authors exactly one message and then listens; a role that chooses what to send belongs in the cast. Which of the two a role is comes from the projection, not from a flag.

The same session, as a runnable file, is examples/online/quickstart.py:

export OPENAI_API_KEY=...
uv run python examples/online/quickstart.py

Declare your own conversation

A message is a class. A protocol is a list of steps: send, choose, loop / repeat, par, closed by Protocol.

import agentsparty as ap

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


class Question(ap.Message):
    """What the reader asks."""

    text: str


class Article(ap.Message):
    """The answer, three paragraphs."""

    text: str


protocol = ap.Protocol(
    ap.send(Reader, Writer, Question),
    ap.send(Writer, Reader, Article),
)

Protocol patterns

The shapes agent systems keep re-deriving are ready-made in ap.patterns. Each returns a Shape: a closed protocol plus the message classes it declares (shape.Answer is a real ap.Message subclass).

Pattern	What it is	Example
request_reply	a call and its answer	examples/online/hello.py
review_loop	draft, review, and a notice until accepted	examples/online/quickstart.py
pipeline	a relay chain, each stage handing work on	examples/online/pipeline.py
triage	a router that hands work to one of several desks	examples/online/help_desk.py
swarm	peers answer or hand work around a ring	examples/online/agent_swarm.py
best_of	several candidates collected and selected	examples/online/best_of.py

Patterns project their shape before returning it, so a pattern cannot be handed a conversation that refuses to run. See the patterns guide.

What it refuses to run

The same declarative surface rejects a conversation that cannot be carried out by the roles it names — before the first model call. Hand-write the review loop without the notice, so the Requester is never told the loop went round:

import agentsparty as ap

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


class Task(ap.Message):
    """The writing task."""

    text: str


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

    text: str


class Approve(ap.Message):
    """Accept the draft."""


class Reject(ap.Message):
    """Send the draft back."""

    text: str


class Final(ap.Message):
    """The finished article."""

    text: str


protocol = ap.Protocol(
    ap.send(Requester, Writer, Task),
    ap.loop(
        ap.send(Writer, Reviewer, Draft),
        ap.choose(
            Reviewer,
            Writer,
            {
                Approve: [
                    ap.send(Writer, Requester, Final)
                ],
                Reject: [ap.repeat()],
            },
        ),
    ),
)

print(ap.render(protocol))
try:
    ap.Session(
        protocol,
        {Writer: ap.Script([]), Reviewer: ap.Script([])},
    )
except ap.ProjectionError as error:
    print(error.role)
    print(error.where)
    print(error.recipe)

Projection fails, and the error names the role, both branches, and the fix — which is exactly the notice the working session sends. The fields .role / .where / .recipe carry the same diagnosis:

role 'Requester' cannot tell the branches of the alt Reviewer -> Writer apart:
  on 'Approve' it must receive Final from Writer (as Requester), on 'Reject' it must loop at 'loop.0'.
A role that behaves differently per branch must be told which branch was taken — add a message from Reviewer (or Writer) to Requester inside each branch.

No API key, no network, and no model call: the refusal is pure projection. A protocol built from ap.patterns cannot fail this way: every pattern inserts its synchronising messages and projects its shape before it is returned.

The full ladder of examples, from a two-message session to a coding harness, is in examples/README.md.

Deterministic runs

A test does not call a network. ap.replies is a model that returns scripted messages; ap.Script is a role that says known messages in order.

import agentsparty as ap

A, B = ap.roles('A', 'B')


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

    text: str


class Answer(ap.Message):
    """The reply."""

    text: str


protocol = ap.Protocol(
    ap.send(A, B, Ask), ap.send(B, A, Answer)
)
transcript = ap.Session(
    protocol,
    {
        A: ap.Script([Ask('q')]),
        B: ap.Agent(ap.replies([Answer('ok')]), ''),
    },
).run()
print(transcript.last(Answer) == Answer('ok'))

Embedding in a service

Session.start returns a live Run. A person outside the process decides through ap.Desk.

import agentsparty as ap

A, B = ap.roles('A', 'B')


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

    text: str


class Answer(ap.Message):
    """The reply."""

    text: str


async def open_ticket(text: str) -> ap.Run:
    desk = ap.Desk()
    protocol = ap.Protocol(
        ap.send(A, B, Ask), ap.send(B, A, Answer)
    )
    return ap.Session(protocol, {B: ap.Human(desk)}).start(
        A, Ask(text)
    )

When to use — and when not to

Use when

• the allowed interaction shape is known up front
• every role must only act on messages it actually receives
• you need projection to fail closed before the first model call
• tools are roles with a protocol surface, not free function-calling
• sessions must resume from recorded decisions without re-asking

Do not use when

• one agent with a free tool set is enough (use a simpler agent SDK)
• the route must be discovered at run time by the model
• you need a large catalogue of vendor integrations out of the box
• you require a stable API before 1.0 (this project is research / 0.x)
• you need multi-process or multi-machine transport (the runtime is in-process)

Links

• Documentation
• Examples
• Messages
• Protocol
• Sessions and runs
• Projection
• Protocol patterns
• A very gentle introduction to Multiparty Session Types
• Global Types for Agent Interaction Protocols
• Security
• Contributing

Status

Research framework at 0.2.x. The public surface is agentsparty.__all__ (68 names) plus the tier-2 submodules; see tests/public_api.txt. What agentsparty proves, checks at run time, and deliberately leaves to the application — including the non-guarantees (no deadlock-freedom, no liveness, no exactly-once) — is set out in the guarantee table. Exception types are stable; message text and journal formats are not.

Security

Untrusted payloads and web content are data, not instructions. User-written handlers run with the privileges of the host process — sandbox them, and validate paths or commands before any effect. Give a hand-built OpenAI client a finite transport timeout; ap.model('openai:...') already applies one. within= on a step, Allowance, and ap.metered bound branch windows, protocol steps, and token spend. Journals and tracers persist payloads and model output in plaintext.

Private reports and security guidance: SECURITY.md.

Development

uv sync --all-groups
just all    # or: uv run nox -t ci

Agent conventions for contributors live in AGENTS.md.

License

MIT — see LICENSE.