feat: validate graph connections during drag

README.md

@@ -71,6 +71,34 @@ flow
 
 For property schemas and richer editors, provide `node_types`/`edge_types` with `PropertySpec` and handle changes via `ReactFlow.on(...)`. `.on` callbacks receive the event payload as the first argument and can optionally accept the `ReactFlow` instance as a second argument.
 
+## Connection validation
+
+Connection checks are opt-in. For frontend-only validation, set `connection_validation` with any of `direction`, `types`, `capacity`, `duplicates`, or `cycles` set to `True`. `types` compares declared handle types case-insensitively and permits unknown types; `capacity` uses `maxConnections` on input handle dictionaries, with no limit when omitted. Existing handle connectability flags still apply.
+
+```python
+from panel_reactflow import NodeType, ReactFlow
+
+flow = ReactFlow(
+    node_types={
+        "source": NodeType(type="source", outputs=[{"id": "value", "type": "str"}]),
+        "sink": NodeType(type="sink", inputs=[{"id": "value", "type": "str", "maxConnections": 1}]),
+    },
+    connection_validation={"direction": True, "types": True, "capacity": True, "cycles": True},
+)
+```
+
+For application rules, register a Python validator. On each drag start, ReactFlow requests results for every candidate port and shows the returned reasons while dragging. The callback receives an edge-shaped payload with `source`, `target`, `sourceHandle`, and `targetHandle`; return `None` to allow or a string to reject. Hooks and frontend checks can be used together. Validate again in your `edge_added` handler before accepting the connection, since the graph may have changed after drag start.
+
+```python
+def validate_connection(edge, flow):
+    if edge["source"] == edge["target"]:
+        return "A node cannot connect to itself."
+
+flow.add_connection_validator(validate_connection)
+```
+
+Run the [connection validation demo](examples/connection_validation.py) with `PYTHONPATH=src pixi run panel serve examples/connection_validation.py --show` from the repository root.
+
 ## Development
 
 ```bash

docs/how-to/validate-connections.md

@@ -0,0 +1,82 @@
+# Validate Connections
+
+Use `connection_validation` to reject invalid connections as users drag between handles. The policies run in the browser; a Python validator can add application-specific rules. Both are opt-in and can be used together.
+
+## Configure browser policies
+
+Set only the checks your graph needs:
+
+| Policy | Rejects |
+|--------|---------|
+| `direction` | Connections that do not run from a declared output to a declared input. Users can still start a drag at either end. |
+| `types` | Connections between handles with different declared `type` values (case-insensitive). If either type is missing, the connection is allowed. |
+| `capacity` | Connections to an input handle that has reached its positive integer `maxConnections` limit. Inputs without a limit are unrestricted. |
+| `duplicates` | A second edge with the same source, target, source handle, and target handle. Different handle pairs can still connect the same nodes. |
+| `cycles` | Connections that would create a cycle, including a self-connection. |
+
+Policies default to off. Existing `NodeType` connectable flags still control whether a handle can start or receive a drag; see [Control Handle Connectivity](control-handle-connectivity.md).
+
+The following app uses handle types and capacity limits alongside a Python rule. Save it as `validation_app.py` and run `panel serve validation_app.py --show`:
+
+```python
+import panel as pn
+
+from panel_reactflow import NodeSpec, NodeType, ReactFlow
+
+pn.extension("jsoneditor")
+
+flow = ReactFlow(
+    nodes=[
+        NodeSpec(id="source", type="source", label="Source", position={"x": 0, "y": 0}),
+        NodeSpec(id="transform", type="transform", label="Transform", position={"x": 260, "y": 0}),
+        NodeSpec(id="publish", type="publish", label="Publish", position={"x": 520, "y": 0}),
+    ],
+    node_types={
+        "source": NodeType(type="source", outputs=[{"id": "text", "type": "Text"}]),
+        "transform": NodeType(
+            type="transform",
+            inputs=[{"id": "text", "type": "text", "maxConnections": 1}],
+            outputs=[{"id": "cleaned", "type": "Text"}],
+        ),
+        "publish": NodeType(type="publish", inputs=[{"id": "text", "type": "Text", "maxConnections": 1}]),
+    },
+    connection_validation={
+        "direction": True,
+        "types": True,
+        "capacity": True,
+        "duplicates": True,
+        "cycles": True,
+    },
+    height=350,
+    width=800,
+)
+
+
+def validate_connection(edge, flow):
+    if edge["source"] == "source" and edge["target"] == "publish":
+        return "Publish requires the Transform output."
+    return None
+
+
+flow.add_connection_validator(validate_connection)
+
+
+def on_edge_added(event, flow):
+    edge = event["edge"]
+    if validate_connection(edge, flow):
+        flow.remove_edge(edge["id"])
+
+
+flow.on("edge_added", on_edge_added)
+flow.servable()
+```
+
+Drag from **Source.text** to **Transform.text**, then from **Transform.cleaned** to **Publish.text**. A direct connection from Source to Publish is rejected by Python. Try connecting Source.text to Transform.cleaned (two outputs) to see the direction check, or connect an output to an already occupied input to see the capacity check.
+
+## Add application rules
+
+Register callbacks with `flow.add_connection_validator(callback)` and unregister them with `flow.remove_connection_validator(callback)`. Each callback accepts either `edge` or `(edge, flow)` and returns `None` to allow the connection or a reason string to reject it. The payload has `source`, `target`, `sourceHandle`, and `targetHandle` keys; handles without an explicit ID use `None`.
+
+When a user starts dragging from an output or input, Python checks every candidate handle on the opposite side. Validators run in registration order until one rejects a candidate. Browser policies and Python reasons are shown on handles while dragging; the browser prevents a rejected connection. Python results must arrive before the connection is dropped: pending requests and requests taking longer than three seconds block the connection. Exceptions also reject the candidate and are logged on the server. Keep validators fast.
+
+Validation is for interactive drags, not a constraint on `flow.edges` or `flow.add_edge()`. Python validators run at drag start, not when the edge is added. If the rule depends on graph state that may change during the drag, check it again in an `edge_added` handler before keeping or persisting the edge. The example above removes an edge if its application rule no longer holds.

docs/index.md

@@ -62,6 +62,7 @@ flow.servable()
 - [Define Nodes & Edges](how-to/define-nodes-edges.md)
 - [Declare Node & Edge Types](how-to/declare-types.md)
 - [Control Handle Connectivity](how-to/control-handle-connectivity.md) — restrict connections
+- [Validate Connections](how-to/validate-connections.md) — check types, capacity, cycles, and application rules
 - [Define Editors](how-to/define-editors.md) — node *and* edge editors
 - [Embed Views in Nodes](how-to/embed-views-in-nodes.md)
 - [Style Nodes & Edges](how-to/style-nodes-edges.md)

examples/connection_validation.py

@@ -0,0 +1,89 @@
+"""Try frontend rules and Python connection hooks in one graph.
+
+Run from the repository root with:
+
+    PYTHONPATH=src pixi run panel serve examples/connection_validation.py --show
+"""
+
+import panel as pn
+import panel_material_ui as pmui
+
+from panel_reactflow import NodeSpec, NodeType, ReactFlow
+
+pn.extension()
+
+
+class ConnectionValidationDemo(pn.viewable.Viewer):
+    def __init__(self, **params):
+        super().__init__(**params)
+        self._status = pn.pane.Markdown("No connections yet.")
+        self._flow = ReactFlow(
+            nodes=[
+                NodeSpec(id="source", type="source", label="Source", position={"x": 0, "y": 60}).to_dict(),
+                NodeSpec(id="number", type="number", label="Number", position={"x": 0, "y": 280}).to_dict(),
+                NodeSpec(id="transform", type="transform", label="Transform", position={"x": 290, "y": 60}).to_dict(),
+                NodeSpec(id="publish", type="publish", label="Publish", position={"x": 610, "y": 60}).to_dict(),
+                NodeSpec(id="monitor", type="monitor", label="Monitor", position={"x": 610, "y": 280}).to_dict(),
+            ],
+            node_types={
+                "source": NodeType(type="source", outputs=[{"id": "text", "type": "Text"}]),
+                "number": NodeType(type="number", outputs=[{"id": "value", "type": "Number"}]),
+                "transform": NodeType(
+                    type="transform",
+                    inputs=[{"id": "text", "type": "Text", "maxConnections": 1}],
+                    outputs=[{"id": "cleaned", "type": "Text"}],
+                ),
+                "publish": NodeType(type="publish", inputs=[{"id": "text", "type": "Text", "maxConnections": 1}]),
+                "monitor": NodeType(type="monitor", inputs=[{"id": "text", "type": "Text"}]),
+            },
+            connection_validation={
+                "direction": True,
+                "types": True,
+                "capacity": True,
+                "duplicates": True,
+                "cycles": True,
+            },
+            height=510,
+            sizing_mode="stretch_width",
+        )
+        self._flow.add_connection_validator(self._validate_connection)
+        self._flow.on("edge_added", self._on_edge_added)
+        self._page = pmui.Page(
+            title="Connection validation",
+            main=[
+                pmui.Container(
+                    pmui.Column(
+                        pn.pane.Markdown(
+                            "Drag **Source.text** to **Transform.text**, then **Transform.cleaned** to **Publish.text**. "
+                            "**Number.value** to **Transform.text** fails the frontend type check; "
+                            "**Source.text** to **Publish.text** is rejected by Python. "
+                            "Try connecting both text outputs to **Monitor.text**, or creating a cycle."
+                        ),
+                        self._flow,
+                        self._status,
+                        sizing_mode="stretch_width",
+                    ),
+                    width_option="lg",
+                ),
+            ],
+        )
+
+    def _validate_connection(self, edge, flow):
+        if edge["source"] == "source" and edge["target"] == "publish":
+            return "Publish requires text from Transform, not Source."
+
+    def _on_edge_added(self, event, flow):
+        edge = event["edge"]
+        reason = self._validate_connection(edge, flow)
+        if reason:
+            flow.remove_edge(edge["id"])
+            self._status.object = f"Rejected: {reason}"
+        else:
+            self._status.object = f"Connected: {edge['source']}.{edge['sourceHandle']} to {edge['target']}.{edge['targetHandle']}"
+
+    def __panel__(self):
+        return self._page
+
+
+demo = ConnectionValidationDemo()
+demo.servable()

src/panel_reactflow/base.py

@@ -269,8 +269,9 @@ class NodeType:
     inputs : list of str or dict, optional
         List of input port definitions. Each entry can be a plain string
         (the handle ID) or a dict with ``"id"`` and optional ``"label"``
-        and ``"type"`` keys. When a label and/or type is provided it renders
-        as a tooltip on hover, e.g. ``"Raw data input (DataFrame)"``.
+        and ``"type"`` keys. An optional ``"maxConnections"`` integer limits
+        connections when ``connection_validation["capacity"]`` is enabled.
+        Labels and types render as tooltips on hover.
     outputs : list of str or dict, optional
         List of output port definitions. Each entry can be a plain string
         (the handle ID) or a dict with ``"id"`` and optional ``"label"``
@@ -361,8 +362,8 @@ class NodeType:
     type: str
     label: str | None = None
     schema: Any = None
-    inputs: list[str | dict[str, str]] | None = None
-    outputs: list[str | dict[str, str]] | None = None
+    inputs: list[str | dict[str, Any]] | None = None
+    outputs: list[str | dict[str, Any]] | None = None
     input_connectable: bool = True
     input_connectable_start: bool = True
     input_connectable_end: bool = True
@@ -1525,6 +1526,8 @@ class ReactFlow(ReactComponent):
     edges = param.List(default=[], doc="Canonical list of edge dictionaries or Edge instances.")
     node_types = param.Dict(default={}, doc="Node type descriptors keyed by type name.")
     edge_types = param.Dict(default={}, doc="Edge type descriptors keyed by type name.")
+    connection_validation = param.Dict(default={}, doc="Opt-in frontend connection policies (direction, cycles, duplicates, types, capacity).")
+    has_connection_validators = param.Boolean(default=False, doc="Whether Python connection validators are registered.")
 
     node_editors = param.Dict(default={}, doc="Node editor factories keyed by type name.", precedence=-1)
     edge_editors = param.Dict(default={}, doc="Edge editor factories keyed by type name.", precedence=-1)
@@ -1645,6 +1648,7 @@ def __init__(self, **params: Any):
             params["edges"] = [ReactFlow._coerce_edge(edge) for edge in params["edges"]]
         super().__init__(**params)
         self._event_handlers: dict[str, list[Callable]] = {"*": []}
+        self._connection_validators: list[Callable] = []
         self.param.watch(self._sync_instance_flow_refs, ["nodes", "edges"])
         self.param.watch(self._normalize_nodes, ["nodes"])
         self.param.watch(self._normalize_edges, ["edges"])
@@ -2448,6 +2452,8 @@ def _handle_msg(self, msg: dict[str, Any]) -> None:
 
     def _process_msg(self, msg: dict[str, Any]) -> None:
         match msg.get("type"):
+            case "connection_validation_requested":
+                self._validate_connection_request(msg)
             case "sync":
                 nodes = msg.get("nodes")
                 edges = msg.get("edges")
@@ -2542,6 +2548,60 @@ def _process_msg(self, msg: dict[str, Any]) -> None:
             case _:
                 return
 
+    def add_connection_validator(self, callback: Callable) -> None:
+        """Register a validator returning None to allow or a reason to reject.
+
+        Callbacks accept either a connection payload or (payload, flow).
+        They run in registration order for each candidate during a drag.
+        """
+        if not callable(callback):
+            raise TypeError("Connection validator must be callable.")
+        self._connection_validators.append(callback)
+        self.has_connection_validators = True
+
+    def remove_connection_validator(self, callback: Callable) -> None:
+        """Unregister a previously added connection validator."""
+        self._connection_validators.remove(callback)
+        self.has_connection_validators = bool(self._connection_validators)
+
+    def _validate_connection_request(self, msg: dict[str, Any]) -> None:
+        node_id = msg["node_id"]
+        handle_id = msg["handle_id"]
+        handle_type = msg["handle_type"]
+        opposite_type = "target" if handle_type == "source" else "source"
+        port_key = "inputs" if opposite_type == "target" else "outputs"
+        results = []
+        for node in self.nodes:
+            candidate_id = self._node_id(node)
+            node_type = self.node_types.get(self._node_type(node), {})
+            handles = node_type.get(port_key)
+            if handles is None:
+                handles = [None]
+            for handle in handles:
+                candidate_handle = handle.get("id") if isinstance(handle, dict) else handle
+                if handle_type == "source":
+                    payload = {"source": node_id, "target": candidate_id, "sourceHandle": handle_id, "targetHandle": candidate_handle}
+                else:
+                    payload = {"source": candidate_id, "target": node_id, "sourceHandle": candidate_handle, "targetHandle": handle_id}
+                reason = None
+                for callback in self._connection_validators:
+                    try:
+                        if len(inspect.signature(callback).parameters) == 2:
+                            reason = callback(payload, self)
+                        else:
+                            reason = callback(payload)
+                        if reason is not None and not isinstance(reason, str):
+                            reason = "Connection validator must return None or a reason string."
+                        elif reason == "":
+                            reason = "Connection rejected."
+                    except Exception as exc:
+                        _LOGGER.exception("Connection validator failed")
+                        reason = str(exc) or "Connection validator failed."
+                    if reason is not None:
+                        break
+                results.append({"node_id": candidate_id, "handle_id": candidate_handle, "handle_type": opposite_type, "reason": reason})
+        self._send_msg({"type": "connection_validation_result", "request_id": msg["request_id"], "results": results})
+
     def _handle_client_error(self, msg: dict[str, Any]) -> None:
         """Log a client-side error reported by the frontend and re-emit it.
 

src/panel_reactflow/dist/css/reactflow.css

@@ -107,6 +107,35 @@
     opacity: 1;
 }
 
+.react-flow__handle.rf-handle-valid {
+    background: #16803c;
+    border-color: #16803c;
+}
+
+.react-flow__handle.rf-handle-pending {
+    background: #797d87;
+    border-color: #797d87;
+}
+
+.react-flow__handle.rf-handle-invalid {
+    background: #bb3b35;
+    border-color: #bb3b35;
+}
+
+.rf-validation-status {
+    position: absolute;
+    bottom: 12px;
+    left: 50%;
+    transform: translateX(-50%);
+    padding: 6px 10px;
+    border-radius: 4px;
+    background: var(--panel-background-color, #fff);
+    color: var(--panel-on-background-color, #222);
+    box-shadow: 0 2px 8px rgba(0, 0, 0, 0.2);
+    pointer-events: none;
+    z-index: 5;
+}
+
 .rf-context-menu {
     background: var(--xy-node-background-color, var(--panel-background-color));
     border: 1px solid var(--panel-border-color);