quartermaster-nodes

Composable node types for building AI agent graphs.

Features

• 40 built-in node types across 6 categories: LLM, control flow, data, user interaction, memory, and utility
• Framework-agnostic via the NodeContext protocol -- integrates with any runtime
• Stateless class-based design -- all nodes use @classmethod with no instance state
• Chain-of-Responsibility pattern for composable LLM processing pipelines
• NodeCatalog with auto-discovery, version tracking, and catalog generation (formerly NodeRegistry — old name still works as an alias)
• Configurable flow behavior per node: traversal strategy, thought type, message type, error handling

Installation

pip install quartermaster-nodes

Dependencies: jinja2>=3.1, quartermaster-graph (enums), quartermaster-providers (LLM config)

Quick Start

Register and Discover Nodes

from quartermaster_nodes import NodeCatalog
from quartermaster_nodes.nodes import InstructionNodeV1, StartNodeV1, EndNodeV1, Decision1

# Manual registration
catalog = NodeCatalog()
catalog.register(StartNodeV1)
catalog.register(InstructionNodeV1)
catalog.register(Decision1)
catalog.register(EndNodeV1)

# Or auto-discover all built-in nodes
catalog = NodeCatalog()
count = catalog.discover("quartermaster_nodes.nodes")
print(f"Discovered {count} nodes")

# Look up a node by name
node_cls = catalog.get("InstructionNode")
print(node_cls.info().description)

# Generate a JSON catalog of all nodes
all_nodes = catalog.catalog_json()

Not the runtime registry. NodeCatalog is a design-time catalog of node class definitions. If you're wiring up FlowRunner, you want a runtime executor registry — use quartermaster_engine.SimpleNodeRegistry for that. NodeRegistry (the old name, still aliased here) collides with quartermaster_engine.nodes.NodeRegistry (a Protocol); passing the wrong one to FlowRunner now raises a clear TypeError instead of AttributeError.

Creating a Custom Node

from quartermaster_nodes import AbstractAssistantNode, AssistantInfo, FlowNodeConf
from quartermaster_nodes.enums import (
    AvailableTraversingIn,
    AvailableTraversingOut,
    AvailableThoughtTypes,
    AvailableMessageTypes,
)


class HttpRequestNode(AbstractAssistantNode):
    """Fetch data from an HTTP endpoint."""

    @classmethod
    def info(cls) -> AssistantInfo:
        info = AssistantInfo()
        info.version = cls.version()
        info.description = "Make an HTTP request and return the response"
        info.metadata = {"url": "", "method": "GET"}
        return info

    @classmethod
    def name(cls) -> str:
        return "HttpRequest"

    @classmethod
    def version(cls) -> str:
        return "1.0"

    @classmethod
    def flow_config(cls) -> FlowNodeConf:
        return FlowNodeConf(
            traverse_in=AvailableTraversingIn.AwaitFirst,
            traverse_out=AvailableTraversingOut.SpawnAll,
            thought_type=AvailableThoughtTypes.NewThought1,
            message_type=AvailableMessageTypes.Tool,
        )

    @classmethod
    def think(cls, ctx) -> None:
        import urllib.request

        url = cls.get_metadata_key_value(ctx, "url", "")
        method = cls.get_metadata_key_value(ctx, "method", "GET")

        req = urllib.request.Request(url, method=method)
        with urllib.request.urlopen(req) as response:
            body = response.read().decode()

        ctx.handle.append_text(body)

Implementing the NodeContext Protocol

Nodes are framework-agnostic. Implement the NodeContext protocol to integrate with your runtime:

from quartermaster_nodes.protocols import NodeContext


class MyRuntimeContext:
    """Adapts your runtime to the NodeContext protocol."""

    def __init__(self, node_config, thought):
        self._config = node_config
        self._thought = thought

    @property
    def node_metadata(self) -> dict:
        return self._config.metadata

    @property
    def flow_node_id(self):
        return self._config.id

    @property
    def thought_id(self):
        return self._thought.id if self._thought else None

    @property
    def thought(self):
        return self._thought

    @property
    def handle(self):
        return self._thought  # Must implement ThoughtHandle protocol

    @property
    def assistant_node(self):
        return self._config

    @property
    def chat_id(self):
        return None


# Execute a node
ctx = MyRuntimeContext(node_config, thought)
InstructionNodeV1.think(ctx)

Node Categories

LLM Nodes (9)

| AgentNodeV1 | Autonomous agent with agentic loop and tool orchestration | | Summarize1 | LLM-powered summarization | | Merge1 | Combine parallel branches via LLM |

Control Flow Nodes (6)

Data Nodes (9)

User Interaction Nodes (4)

Memory Nodes (5)

Utility Nodes (6)

API Reference

AbstractAssistantNode

Base class for all nodes. All methods are @classmethod.

AbstractLLMAssistantNode

Extends AbstractAssistantNode with LLM configuration.

Configurable metadata keys: llm_model, llm_provider, llm_temperature, llm_max_input_tokens, llm_max_output_tokens, llm_max_messages, llm_stream, llm_vision, llm_system_instruction, llm_thinking_level.

FlowNodeConf

Defines a node's flow behavior constraints.

NodeCatalog (formerly NodeRegistry)

NodeRegistry is kept as a backward-compatible alias for NodeCatalog.

Chain-of-Responsibility

LLM nodes use a composable handler chain for processing:

from quartermaster_nodes.chain import Chain
from quartermaster_nodes.chain.handlers import (
    ValidateMemoryID,
    PrepareMessages,
    ContextManager,
    TransformToProvider,
    GenerateStreamResponse,
    ProcessStreamResponse,
)

chain = (
    Chain()
    .add_handler(ValidateMemoryID())
    .add_handler(PrepareMessages(client, config))
    .add_handler(ContextManager(client, config, ctx_config))
    .add_handler(TransformToProvider(transformer))
    .add_handler(GenerateStreamResponse(client, config))
    .add_handler(ProcessStreamResponse())
)

result = chain.run({"memory_id": thought_id, "flow_node_id": node_id, "ctx": ctx})

Integration with Sibling Packages

With quartermaster-graph (graph schema)

Enums are re-exported from quartermaster-graph for consistency:

from quartermaster_nodes.enums import (
    AvailableNodeTypes,
)  # Alias for quartermaster_graph.enums.NodeType
from quartermaster_graph.enums import NodeType  # Canonical source

With quartermaster-engine (execution runtime)

quartermaster-nodes and quartermaster-engine operate on different axes:

• NodeCatalog (this package) — design-time catalog of node class definitions for discovery, introspection, and versioning.
• SimpleNodeRegistry (engine) — runtime registry of node executors that FlowRunner dispatches to.

For runtime execution with FlowRunner, register executors directly with the engine registry:

from quartermaster_engine import FlowRunner
from quartermaster_engine.nodes import SimpleNodeRegistry

registry = SimpleNodeRegistry()
registry.register("Instruction1", my_instruction_executor)
runner = FlowRunner(graph=graph, node_registry=registry)

The design-time catalog is still useful — to inspect available node classes before wiring executors:

from quartermaster_nodes import NodeCatalog

catalog = NodeCatalog()
catalog.discover("quartermaster_nodes.nodes")

node_cls = catalog.get("InstructionNode", "1.0")
print(node_cls.info().description)
node_cls.think(execution_context)  # direct invocation (no engine involved)

Contributing

See CONTRIBUTING.md for guidelines.

License

Apache License 2.0 -- see LICENSE for details.