Introducing roomkit-graph: Workflow Graphs for RoomKit

roomkit-graph is a workflow graph engine for RoomKit. Define multi-step business processes as directed graphs — with AI agents, human approvals, parallel execution, and conditional branching — and run them as RoomKit rooms. It's now available on PyPI as v0.1.0a1.

Why a Library, Not a Platform

Existing workflow engines like n8n, Airflow, and Temporal are full platforms with their own servers, UIs, and deployment requirements. That's the right choice when you need a standalone orchestration system. But when you're already building on RoomKit and want to add "when X happens, do A, then based on the result do B or C, wait for approval, then notify" — you don't need another platform. You need a library.

roomkit-graph is that library. One dependency (roomkit), no external services, embed directly in your app.

Graphs Are Data, Not Code

Every graph is fully serializable to JSON. Build it in Python, export it as a dict, store it in a database, load it in another process, render it in a UI. This is a deliberate design choice: workflows should be inspectable, versionable, and portable.

from roomkit_graph import Graph, Node, Edge, Condition, WebhookTrigger

graph = Graph(
    id="bug-triage",
    name="Bug Triage",
    trigger=WebhookTrigger(source_type="github"),
)

graph.add_nodes(
    Node("start", type="start"),
    Node("triage", type="agent", config={
        "agent_id": "triage-agent",
    }),
    Node("escalate", type="notification", config={
        "channel": "slack",
    }),
    Node("assign", type="agent", config={
        "agent_id": "labeler-agent",
    }),
    Node("end", type="end"),
)

graph.add_edges(
    Edge("start", "triage"),
    Edge("triage", "escalate",
         condition=Condition.field("triage.output.severity").equals("critical")),
    Edge("triage", "assign",
         condition=Condition.otherwise()),
    Edge("escalate", "end"),
    Edge("assign", "end"),
)

The entire graph — nodes, edges, conditions, triggers — round-trips cleanly through graph.to_dict() and Graph.from_dict(data).

Key Features

Fluent Condition Builder

The condition DSL reads like English and serializes to JSON. Field comparisons, logical combinators (all_, any_, not_), and an otherwise() catch-all. Edges are evaluated in definition order — first match wins.

Condition.field("triage.output.severity").equals("critical")
Condition.field("extract.output.amount").gt(1000)
Condition.all_(
    Condition.field("triage.output.severity").in_(["critical", "high"]),
    Condition.field("triage.output.team").equals("backend"),
)
Condition.otherwise()

Human-in-the-Loop

The engine supports pause, persist, and resume. When a handler returns status="waiting", the engine pauses. Serialize the entire state to a database with engine.to_dict(), restore it later in another process with WorkflowEngine.from_dict(), and resume with human input.

# Pause — serialize to DB
state = engine.to_dict()
save_to_db(state)

# Later, in another process
engine = WorkflowEngine.from_dict(graph, load_from_db(), handlers=handlers)
await engine.resume("review", {"action": "approve"})

Parallel Execution

The built-in ParallelHandler executes child nodes concurrently using asyncio.TaskGroup. Define children with a parent reference, and the handler fans out, executes all children, and aggregates results.

Node("checks", type="parallel", config={"join": "all"})
Node("security", type="agent", parent="checks")
Node("compliance", type="agent", parent="checks")

# After execution:
# ctx.get("security.output")   → individual result
# ctx.get("checks.output")     → {"security": ..., "compliance": ...}

Pluggable Handlers

roomkit-graph ships built-in handlers for start, end, function, condition, switch, log, and parallel nodes. You provide handlers for your app-specific node types — agent, human, notification, orchestration — by implementing the NodeHandler interface.

from roomkit_graph import NodeHandler, NodeResult, WorkflowEngine

class AgentHandler(NodeHandler):
    async def execute(self, node, context, engine):
        result = await call_agent(node.config["agent_id"])
        return NodeResult(output=result, status="completed")

engine = WorkflowEngine(graph, handlers={
    "agent": AgentHandler(),
})
ctx = await engine.run()

Template Resolver

Steps reference previous outputs with {{node_id.output.field}} templates, resolved at runtime from the workflow context. When a template is exactly one placeholder, the raw value passes through without stringification — dicts, lists, and numbers keep their types.

Getting Started

pip install roomkit-graph

The project is open source under the MIT license. Check out the GitHub repository for full documentation, examples, and the API reference.

This is v0.1.0a1 — a pre-alpha release. The API is not stable yet, but the core engine, condition system, and handler model are solid and tested. We'd love your feedback.