TopologyValidator API Reference

Overview

TopologyValidator provides static analysis and security enforcement for dynamically generated graphs. It prevents self-modifying agents from creating unsafe topologies like infinite loops, unauthorized tool calls, or malformed structures.

Class Signature

class TopologyValidator:
    def __init__(self, allowed_tools: List[callable] = None)

Parameters

Parameter	Type	Required	Description
`allowed_tools`	`List[callable]`	No	Allowlist of Python functions that can be used in `ToolNode` instances. If `None`, all tools are allowed (insecure).

Methods

`validate(graph_spec: Dict[str, Any]) -> None`

Validates a JSON graph specification for safety and correctness.

Raises: - SecurityError: If the graph violates safety policies

Checks Performed: 1. Cycle Detection: Ensures the graph is a DAG (Directed Acyclic Graph) 2. Tool Allowlisting: Verifies all ToolNode functions are in the allowed list 3. Structural Integrity: Validates all next_node references exist 4. Type Safety: Ensures node types are valid (LLMNode, ToolNode, etc.)

Security Guarantees

1. Infinite Loop Prevention

TopologyValidator uses cycle detection to ensure no execution path can loop forever:

validator = TopologyValidator()

# This will raise SecurityError
invalid_graph = {
    "nodes": {
        "node_1": {"next_node": "node_2"},
        "node_2": {"next_node": "node_1"}  # Cycle!
    },
    "start_node": "node_1"
}

validator.validate(invalid_graph)  # ❌ SecurityError: Cycle detected

2. Tool Restriction

Only pre-approved tools can be invoked:

def safe_tool():
    return "Safe operation"

def dangerous_tool():
    import os
    os.system("rm -rf /")  # Malicious!

# Create validator with allowlist
validator = TopologyValidator(allowed_tools=[safe_tool])

# This will pass
safe_graph = {
    "nodes": {
        "node_1": {
            "type": "ToolNode",
            "config": {"tool_name": "safe_tool"},
            "next_node": null
        }
    },
    "start_node": "node_1"
}
validator.validate(safe_graph)  # ✅ Passes

# This will fail
malicious_graph = {
    "nodes": {
        "node_1": {
            "type": "ToolNode",
            "config": {"tool_name": "dangerous_tool"},
            "next_node": null
        }
    },
    "start_node": "node_1"
}
validator.validate(malicious_graph)  # ❌ SecurityError: Tool not in allowlist

3. Reference Integrity

All next_node references must point to existing nodes:

broken_graph = {
    "nodes": {
        "node_1": {"next_node": "node_999"}  # node_999 doesn't exist!
    },
    "start_node": "node_1"
}

validator.validate(broken_graph)  # ❌ SecurityError: Invalid reference

Example Usage

Basic Validation

from lar.dynamic import TopologyValidator, SecurityError

def approved_search(query):
    return f"Results for {query}"

def approved_filter(results):
    return [r for r in results if "important" in r]

# Create validator
validator = TopologyValidator(allowed_tools=[approved_search, approved_filter])

# Graph generated by LLM
llm_generated_graph = {
    "nodes": {
        "search": {
            "type": "ToolNode",
            "config": {
                "tool_name": "approved_search",
                "input_keys": ["query"],
                "output_key": "raw_results"
            },
            "next_node": "filter"
        },
        "filter": {
            "type": "ToolNode",
            "config": {
                "tool_name": "approved_filter",
                "input_keys": ["raw_results"],
                "output_key": "filtered_results"
            },
            "next_node": None
        }
    },
    "start_node": "search"
}

try:
    validator.validate(llm_generated_graph)
    print("✅ Graph is safe to execute")
except SecurityError as e:
    print(f"❌ Validation failed: {e}")

Integration with DynamicNode

from lar import DynamicNode, TopologyValidator

# Define safe tools
def fetch_wikipedia(topic):
    # Simulated API call
    return f"Wikipedia summary of {topic}"

def summarize_text(text):
    # Text processing
    return text[:100]

# Create validator
validator = TopologyValidator(allowed_tools=[fetch_wikipedia, summarize_text])

# Create metacognitive node
research_agent = DynamicNode(
    llm_model="gpt-4",
    prompt_template="""
    Design a research graph for: {research_topic}

    Available tools:
    - fetch_wikipedia
    - summarize_text

    Return JSON graph spec.
    """,
    validator=validator,  # Validation happens automatically
    context_keys=["research_topic"]
)

Validation Algorithm

Cycle Detection (DFS-based)

def has_cycle(graph):
    visited = set()
    rec_stack = set()

    def dfs(node_id):
        if node_id in rec_stack:
            return True  # Cycle found
        if node_id in visited:
            return False

        visited.add(node_id)
        rec_stack.add(node_id)

        next_node = graph["nodes"][node_id].get("next_node")
        if next_node and dfs(next_node):
            return True

        rec_stack.remove(node_id)
        return False

    return dfs(graph["start_node"])

Complexity: O(V + E) where V = nodes, E = edges

Security Best Practices

1. Principle of Least Privilege

Only allow the minimum set of tools needed:

# ❌ Too permissive
validator = TopologyValidator(allowed_tools=None)  # All tools allowed!

# ✅ Restrictive
validator = TopologyValidator(allowed_tools=[read_only_function])

2. Layered Validation

Combine TopologyValidator with additional checks:

class CustomValidator(TopologyValidator):
    def validate(self, graph_spec):
        # Call parent validation
        super().validate(graph_spec)

        # Add custom checks
        if len(graph_spec["nodes"]) > 10:
            raise SecurityError("Graph too large (max 10 nodes)")

        for node in graph_spec["nodes"].values():
            if node["type"] == "LLMNode":
                # Ensure no expensive models
                if "gpt-4" in node["config"].get("model_name", ""):
                    raise SecurityError("GPT-4 not allowed (cost control)")

3. Audit Logging

Always log validation results:

try:
    validator.validate(graph_spec)
    logger.info(f"Graph validated successfully: {graph_spec}")
except SecurityError as e:
    logger.error(f"Validation failed: {e}, Graph: {graph_spec}")
    raise

Common Validation Errors

Error	Cause	Fix
`Cycle detected in graph topology`	Node A → Node B → Node A	Remove cyclic reference
`Tool 'X' not in allowlist`	Using unauthorized function	Add tool to `allowed_tools` or remove from graph
`start_node 'X' not found in nodes`	Invalid entry point	Ensure `start_node` exists in `nodes` dict
`Invalid node type: 'X'`	Unsupported node class	Use only: LLMNode, ToolNode, RouterNode

Compliance

EU AI Act Article 13 (Transparency)

TopologyValidator enables compliance by: - Providing deterministic validation (non-AI, rule-based) - Logging exact rejection reasons - Creating audit trail of what was blocked

OWASP Recommendations

Follows OWASP secure coding practices: - ✅ Input validation (graph spec) - ✅ Allow listing (tools) - ✅ Fail-safe defaults (reject invalid graphs) - ✅ Defense in depth (multiple checks)

Performance

Validation Time: O(V + E) - Linear in graph size
Memory: O(V) for visited tracking
Recommended Max Nodes: 100 (for performance, not safety)

For very large graphs (>100 nodes), consider: 1. Caching validation results 2. Incremental validation 3. Graph simplification