graffitty-graph-protocol

Build custom Python nodes for Graffitty's agent graph. Connect your scripts, data transformations, and integrations to other nodes with typed inputs and outputs.

Declare ports, decorate a handler, and return an output. The runtime handles registration, heartbeats, forwarding, acknowledgements, and shutdown. Pydantic models validate messages and port declarations; handlers can be synchronous or async.

Installation

Requires Python 3.11+ and a running Graffitty host. This package is an SDK, not a standalone graph server or an installer for Graffitty.

pip install graffitty-graph-protocol

This is an early release; APIs and the host protocol may change.

Quick start

Save this as node.py:

from graffitty_graph_protocol import Event, Node, Output

node = Node("Uppercase", inputs=["in"], outputs=["out"])

@node.on("in")
def uppercase(message: Event):
    return Output(port="out", text=message.text.upper())

node.run()

Run it inside a Graffitty terminal pane, using the environment where you installed the package:

python node.py

In Graffitty's graph, connect Manual Prompt → Uppercase → Payload Inspector, then send hello graph. The inspector receives HELLO GRAPH.

Graffitty supplies GRAFFITTY_SOCKET and GRAFFITTY_TOKEN to the pane. Running this script in an unrelated terminal without a host connection will not work. Keep the token private.

Handlers and data

• @node.on("input_id") selects an advertised input. Duplicate handlers and undeclared ports fail early. Every input needs a handler.
• Handlers may be ordinary def or async def. Return Output, a list of outputs for multiple branches, or None to consume a message without forwarding.
• Event and the port/output models are Pydantic models. message.payload is JSON; use your own BaseModel.model_validate(message.payload) for application schemas.
• Output(port="out") forwards the original content unchanged. Set text, payload, content, or renderings to replace just those fields. An explicit payload=None clears it. The host preserves run IDs, session references, and provenance; the handler cannot forge those through Output.
• message.sessions can be empty. Each reference has an optional path and checkpoint; Read permission controls whether the transcript path is exposed.
• Generic nodes register as kind="custom" without a provider or synthetic conversation ID. Manual Prompt has sessions: []. Session-reference provider is optional; a real integration can set Node(..., provider="telegram").
• String output IDs declare automatic Send ports with solid wires. Configured OutputPort instances default to Send and line_style=None, using the host's mode-specific style. Set line_style="solid" for automatic outputs, or "dashed" when sending requires a later agent decision.

Complete port declarations

from graffitty_graph_protocol import InputPort, Node, OutputPort, SvgIcon, TextIcon

node = Node("Judge", icon=TextIcon(text="⚖", font="Symbols Nerd Font Mono"),
    inputs=[InputPort(id="in", name="Evaluate", color="#8ecaff", icon="↓")],
    outputs=[
        OutputPort(id="approved", name="Approved", icon="✓", line_style="solid"),
        OutputPort(id="history", mode="read"),
        OutputPort(id="done", mode="turn_completed", icon=SvgIcon(svg='<svg xmlns="http://www.w3.org/2000/svg"/>')),
    ])

Node and port icons accept text/emoji, TextIcon, SvgIcon, or their JSON object forms. Each direction allows 0–32 ports with unique IDs of 1–64 ASCII letters, digits, underscores or hyphens. Node and port names allow 128 UTF-8 bytes; icon text allows 32 Unicode characters, font families 128, and SVG 16,384. Pydantic checks these bounds, colors, modes and styles. Manifest/port application metadata survives round trips, including null values. Manifest.to_wire() omits unset presentation fields while preserving application data. Output overrides cannot add routing or provenance fields.

Handlers may return outputs only on Send ports. Read grants transcript access; Turn completed routes Stop lifecycle events and also grants Read. These modes do not send a second copy of every handler result automatically.

await node.serve() integrates with an existing asyncio application. node.run() is the standalone entry point and handles Ctrl+C/SIGTERM. Heartbeats continue while a synchronous handler works in a worker thread. node.stop() asks the event loop to stop after the current event; call it on that loop.

Applications with their own loop

Use Session without adopting the handler runtime:

from graffitty_graph_protocol import Manifest, OutputPort, Session

async def run_ui():
    async with Session() as session:
        await session.register(Manifest(name="My UI", icon="⚙", inputs=[],
            outputs=[OutputPort(id="done", mode="turn_completed")]))
        # Run your UI here, calling await session.heartbeat() regularly.
        await session.event("Stop", {"text": "finished", "completion_id": "turn-1"})
    # SessionEnd is sent on normal exit and exceptions.

The session uses inbox transport; heartbeats send at most once per five seconds. Node uses the same session internally, available as node.session. register, take, forward, emit, ack, progress and event mirror the Rust session API. emit(Output(...), run_id=..., outcome=...) starts a message on a Send port, using a null payload when omitted. Matching connections are required. Give each Stop a distinct completion_id. For Read access to a custom node's actual transcript, declare session_id and transcript_path with session.event("Ready", {...}) before reporting completion.

await session.call(method, params) and Client.call expose every RPC, including graph.list/adjacent, connect/disconnect, read/spawn, queue, and graph.run.default/limit/reset. The host enforces permissions and run limits. graph.list returns nodes and their manifests; use array lengths for node and input/output counts.

Checkpoint recovery

Node(..., restorable=True) lets a Graffitty checkpoint relaunch the node with its own interpreter, arguments and directory. With Session, pass restore=Restore.current_process() (or an explicit Restore(command=..., cwd=..., env=...)). await session.report_state({...}) keeps up to 64 KiB of JSON for the relaunch, and the relaunched process reads it with restored_state(). Never put secret values in the command, environment or state: keep a reference, such as a credential name in secrets.toml, and resolve the secret again.

Delivery behavior

Handlers run one event at a time. The runtime validates all returned outputs, forwards them, then acknowledges the input. On failure it reports failed progress, leaves the input pending, and stops. Restarting resumes the inbox. It does not promise exactly-once delivery: a crash or a partially completed fan-out can leave some outputs delivered. Make external side effects idempotent using event_id; check what completed before retrying. No hidden retries are made.

The client reads GRAFFITTY_SOCKET and GRAFFITTY_TOKEN from its pane. It supports Unix sockets (including ipc:// paths) and authenticated loopback TCP. Windows named-pipe endpoints use the installed gft rpc transport. stdin and stdout remain available to your program. Context credentials are not included in model data.

Development

From a repository checkout, run the tests:

uv run --directory python/libs/graffitty-graph-protocol pytest

The shared manifest cases in tests/fixtures/graph-manifests.json are checked by both SDKs. For the real socket/graph integration tests, build gft, the graffitty-workspace-core example graph_fixture, and the Rust uppercase example, then run pytest with GRAFFITTY_SDK_INTEGRATION=1 (requires Godot).

License

MIT.