Python

The Python SDK is for writing pugmark handlers. A handler is a process the runtime invokes on each new event in a session; it reads inputs from stdin, fetches object bodies over HTTP, and writes outputs back to stdout.

pip install pugmark

Requires Python 3.11+; zstandard is installed automatically. Optional OpenTelemetry trace-context propagation is available with the otel extra (pip install 'pugmark[otel]').

Minimal handler

A pugmark session is an append-only log of JSON events. On every invocation pugmark feeds the handler the full log on stdin. Walk it, decide one next event, append it.

import pugmark

log = list(pugmark.read())
last = log[-1].decode_json() if log else None
match last:
    case {"kind": "input", "text": text}:
        pugmark.write({"kind": "echo", "text": text.upper()})
    case _:
        pugmark.pause("nothing to echo")

Run it under the CLI:

pugmark start
pugmark push <<<'{"kind":"input","text":"hello"}'
pugmark run --once -- python handler.py
pugmark log -A           # input event, then echo event ({"text":"HELLO"})

Writing outputs

pugmark.write accepts either a pugmark.Object or a raw Python value. Content type is auto-detected unless overridden.

pugmark.write("hello", name="reply")                  # text/plain
pugmark.write({"answer": 42}, name="result")          # application/json
pugmark.write(b"\x00\x01", name="bin")                # application/octet-stream
pugmark.write("# Title", name="doc", content_type="text/markdown")

Explicit helpers when you’d rather not rely on type dispatch:

pugmark.write_text("hello", name="reply")
pugmark.write_json({"answer": 42}, name="result")
pugmark.write_bytes(b"\x00\x01", name="bin")

Reading inputs

pugmark.read() walks the full session log in order, including any inherited parent-session entries. This is the primary primitive for log-replay handlers; iterate it once per invocation to reconstruct everything you need.

for obj in pugmark.read():
    event = obj.decode_json()       # every entry is JSON
    fold(state, event)              # accumulate whatever the handler needs

For ad-hoc named lookups (slot-filling workflows where each name appears at most once), pugmark.state() returns a dict-like view of the session keyed by object name. Names are collapsing: only the most recently written object per name survives, so this isn’t appropriate for conversations or any log with repeated entry kinds.

state = pugmark.state()             # dict-like; one entry per unique name
if "config" in state:
    cfg = state["config"].decode_json()

Named loads with the right decoder built in:

text  = pugmark.load_text("greeting")
reply = pugmark.load_json("result")
blob  = pugmark.load_bytes("bin")

Object methods

Every object (local or remote) supports:

Method	Returns
`body()`	raw bytes (auto-decompressed)
`content_type()`	MIME type string
`name()`	optional name set by the writer
`metadata()`	`Dict[str, str]`
`decode_text()`	UTF-8 decoded `str`
`decode_json()`	parsed JSON value
`decode_data()`	raw bytes

Control flow

pugmark.pause("waiting for next user message")     # suspend until next event
pugmark.sleep(60, reason="rate-limited")           # suspend for 60 seconds

Events

Each handler invocation is triggered by an event. Iterate pugmark.events() to inspect:

for event in pugmark.events():
    match event:
        case pugmark.StartEvent():       ...    # new root session
        case pugmark.ForkEvent():        ...    # forked from a parent
        case pugmark.WakeEvent():        ...    # resumed from pause/sleep
        case pugmark.PushEvent() as e:   ...    # object was pushed; e.object is the Object

Handlers may ignore lifecycle events entirely. The replay-decide-append pattern needs no lifecycle awareness, since the log itself fully determines the next action. See the Handler Protocol spec for the full event semantics.