Pugmark Python SDK

Python SDK for writing Pugmark handlers. A handler is a process the Pugmark 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.

Install

pip install pugmark

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 Pugmark 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. Note that 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:

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

OpenTelemetry

The SDK ships optional OpenTelemetry trace context propagation. Install with the otel extra to enable:

pip install 'pugmark[otel]'

Requirements

• Python 3.11+
• zstandard (installed automatically)

License

Apache License 2.0.