MARP Python SDK

Typed, dependency-free, synchronous implementation of Minimal Assistant Room Protocol (MARP) 0.9.0.

Conforms to MARP 0.9 Core, Direct JSON Profile, and Function Calling Profile. Multi-reply profile: first by default (also supports all and nonblocking).

This library is unrelated to the Marp Markdown-presentation ecosystem.

The approved distribution and import name is marp.

pip install marp

The canonical protocol specification is at https://github.com/marprotocol/marp/blob/main/PROTOCOL.md.

from marp import RoomState

room = RoomState()
room.join("u1")
turn = room.accept_direct({"messages": [{"to": ["u1"], "text": "Hello"}], "reply_from": ["u1"]})
room.present_next()
assert room.capture_message("u1", "Hi") == "waiting"
assert room.complete_presentation() == "advanced"

Private delivery is routing metadata only: room operators and model providers processing a room can access delivery contents.

accept_event, join, and leave return "accepted". capture_message automatically adds the active pending reply_to when omitted and returns "advanced", "waiting", or "accepted"; inspect normalized_history() for the server-created event. Present a turn, then call complete_presentation() to make its completion observable. accept_calls() is atomic and returns ordered queued or typed rejected acknowledgements.

For function-calling providers, retain the original provider assistant item and append function_history(assistant_item, acknowledgements); this produces one correlated tool result per call ID and does not duplicate normalized turns in provider history.

Copy-paste examples

Parse

from marp import parse_event

event = parse_event({"event": "message", "user": "u1", "text": "Hello"})

Validate

from marp import parse_turn, validate_turn

turn = validate_turn(parse_turn({"messages": [], "reply_from": []}), ["u1"])

Ordered calls

from marp import RoomState, ToolCall

room = RoomState()
acks = room.accept_calls([ToolCall("c1", {"messages": [], "reply_from": []})])
assert acks[0].turn == 1

Present

from marp import RoomState

room = RoomState()
room.join("u1")
room.accept_direct({"messages": [], "reply_from": []})
assert room.present_next().turn == 1
assert room.complete_presentation() == "advanced"

Capture

from marp import RoomState

room = RoomState()
room.join("u1")
room.accept_direct({"messages": [], "reply_from": ["u1"]})
room.present_next()
assert room.capture_message("u1", "answer") == "waiting"
assert room.complete_presentation() == "advanced"

Replay

from marp import RoomState

room = RoomState()
first = room.accept_direct({"messages": [], "reply_from": []}, response_id="r1")
assert room.accept_direct({"messages": [], "reply_from": []}, response_id="r1") == first

Conformance

# From python/marp:
# python -m pytest tests/test_conformance.py