LangGraph Integration with Sequrity Control API

This tutorial demonstrates how to integrate LangGraph with Sequrity's Control API using the langgraph.run method. We'll build a SQL agent with conditional routing that showcases LangGraph's powerful workflow capabilities while maintaining security through Sequrity.

Prerequisites

Before starting, ensure you have:

• Sequrity API Key: Sign up at Sequrity to get your API key
• LLM Provider API Key: This example uses OpenRouter, but you can use any supported provider

Set your API keys as environment variables:

export SEQURITY_API_KEY="your-sequrity-api-key"
export OPENROUTER_API_KEY="your-openrouter-api-key"

Download Tutorial Script

• Sequrity Client version

Installation

Install the required packages:

pip install sequrity

Using with LangGraph Integration

If you want to use the direct LangGraph integration (not shown in this tutorial), you need to install the langgraph dependency group:

# Using pip
pip install sequrity[langgraph]

# Using uv (recommended for development)
uv sync --group langgraph

For this tutorial with rich output formatting:

pip install sequrity rich

Dependency Groups

The Sequrity package includes optional dependency groups for different integrations:

• langgraph: For LangGraph integration (install with pip install sequrity[langgraph])
• agents: For OpenAI Agents SDK integration (install with pip install sequrity[agents])

These are only needed if you're using the direct integration helpers like create_sequrity_langgraph_client() or create_sequrity_openai_agents_sdk_client(). The core Sequrity functionality works without these.

Why Use Sequrity with LangGraph?

When building LangGraph workflows, you often need:

1. Security Controls: Ensure your agent doesn't leak sensitive data or perform unauthorized actions
2. Execution Monitoring: Track and audit workflow execution for compliance
3. LLM Integration: Secure interaction with external LLM providers
4. Policy Enforcement: Apply fine-grained security policies to tool calls and data flow

Sequrity's langgraph.run method provides all of this out-of-the-box, allowing you to focus on building your workflow logic while Sequrity handles security.

Tutorial Overview: SQL Agent with Conditional Routing

We'll build a SQL agent that:

1. Lists available database tables
2. Retrieves schema information
3. Generates SQL queries based on user input
4. Conditionally validates complex queries
5. Executes the query and returns results

The workflow includes conditional routing: simple queries execute directly, while complex queries go through a validation step first.

Step 1: Define the State Schema

LangGraph workflows are built around a typed state that flows through nodes. Let's define our SQL agent's state:

# Define the state schema for the SQL agent
class SQLAgentState(TypedDict):
    """State for SQL agent workflow

    Attributes:
        query: User's natural language query
        tables: Available database tables
        schema: Database schema information
        sql_query: Generated SQL query
        result: Query execution result
        needs_validation: Flag indicating if query needs validation
    """

    query: str
    tables: str
    schema: str
    sql_query: str
    result: str
    needs_validation: bool

Each node in the workflow can read from and update this state.

Step 2: Implement Node Functions

External Data Retrieval Nodes

These nodes represent external operations (e.g., database queries). In a real application, they would interact with actual databases:

# Define node functions for the workflow
def list_tables(state: SQLAgentState) -> dict:
    """List available database tables

    In a real application, this would query the database metadata.
    For this demo, we return a static list of tables.
    """
    rprint("📋 Listing available tables...")
    return {"tables": "users, orders, products"}


def get_schema(state: SQLAgentState) -> dict:
    """Get schema information for tables

    In a real application, this would retrieve actual schema from the database.
    For this demo, we return a simplified schema.
    """
    rprint(f"🔍 Getting schema for tables: {state['tables']}")
    schema_info = f"Schema for {state['tables']}: users(id, name, email), orders(id, user_id, total, date), products(id, name, price)"
    return {"schema": schema_info}

Query Generation Node

This node generates SQL based on the user's natural language query:

def generate_query(state: SQLAgentState) -> dict:
    """Generate SQL query based on user question

    In a real application, this might use an LLM to generate the query.
    For this demo, we create a simple query based on the user's input.
    """
    rprint(f"⚙️  Generating SQL for query: {state['query']}")

    # Simple query generation logic (in production, you'd use an LLM here)
    query_lower = state["query"].lower()
    if "user" in query_lower or "customer" in query_lower:
        sql = "SELECT * FROM users WHERE name LIKE '%recent%'"
    elif "order" in query_lower:
        sql = (
            "SELECT u.name, o.total, o.date FROM users u JOIN orders o ON u.id = o.user_id WHERE o.date > '2024-01-01'"
        )
    else:
        sql = "SELECT * FROM users"

    # Determine if query needs validation (complex queries)
    needs_validation = len(sql) > 50 or "JOIN" in sql

    return {"sql_query": sql, "needs_validation": needs_validation}

Validation and Execution Nodes

These nodes handle query validation and execution:

def validate_query(state: SQLAgentState) -> dict:
    """Validate and potentially modify the generated SQL query

    This adds safety measures like query limits.
    """
    rprint("✅ Validating SQL query...")
    validated_sql = state["sql_query"]

    # Add LIMIT if not present
    if "LIMIT" not in validated_sql.upper():
        validated_sql += " LIMIT 100"

    return {"sql_query": validated_sql, "needs_validation": False}


def execute_query(state: SQLAgentState) -> dict:
    """Execute the SQL query

    In a real application, this would execute against a real database.
    For this demo, we simulate execution and return mock results.
    """
    rprint(f"🚀 Executing SQL: {state['sql_query']}")

    # Simulate query execution
    result = f"Query executed successfully!\n\nSQL: {state['sql_query']}\n\nResults: Found 3 matching records:\n  1. John Doe (john@example.com)\n  2. Jane Smith (jane@example.com)\n  3. Bob Johnson (bob@example.com)"

    return {"result": result}

Conditional Routing Function

This function determines the next node based on the current state:

def route_validation(state: SQLAgentState) -> Literal["validate_query", "execute_query"]:
    """Conditional routing: decide whether query needs validation

    This function determines the next node based on the state.
    If the query needs validation, route to validate_query node.
    Otherwise, proceed directly to execute_query.
    """
    if state.get("needs_validation", False):
        rprint("⚠️  Query requires validation, routing to validate_query")
        return "validate_query"
    else:
        rprint("✓ Query looks safe, routing directly to execute_query")
        return "execute_query"

Routing Functions Must Be Included

This routing function must be included in the node_functions dictionary when calling compile_and_run_langgraph, even though it's not a node. It's referenced by the conditional edge and needs to be available during execution.

Step 3: Build the LangGraph Workflow

Now we construct the graph by connecting nodes and edges:

# Build the LangGraph workflow
def build_sql_agent_graph():
    """Construct the SQL agent workflow graph"""
    graph = StateGraph(SQLAgentState)  # ty: ignore[invalid-argument-type]

    # Add all workflow nodes
    graph.add_node("list_tables", list_tables)
    graph.add_node("get_schema", get_schema)
    graph.add_node("generate_query", generate_query)
    graph.add_node("validate_query", validate_query)
    graph.add_node("execute_query", execute_query)

    # Build the workflow edges
    graph.add_edge(START, "list_tables")
    graph.add_edge("list_tables", "get_schema")
    graph.add_edge("get_schema", "generate_query")

    # Conditional edge: route based on needs_validation
    graph.add_conditional_edges(
        "generate_query", route_validation, {"validate_query": "validate_query", "execute_query": "execute_query"}
    )

    graph.add_edge("validate_query", "execute_query")
    graph.add_edge("execute_query", END)

    return graph

Understanding the Workflow

1. START → list_tables: Begin by listing available tables
2. list_tables → get_schema: Retrieve schema for those tables
3. get_schema → generate_query: Generate SQL based on user input and schema
4. Conditional Branch:
   • If needs_validation=True: → validate_query → execute_query → END
   • If needs_validation=False: → execute_query → END

This demonstrates LangGraph's powerful conditional routing capabilities.

Step 4: Execute with Sequrity Control API

Initialize the Sequrity Client

# Initialize Sequrity client
openrouter_api_key = os.getenv("OPENROUTER_API_KEY", "your-openrouter-api-key")
sequrity_key = os.getenv("SEQURITY_API_KEY", "your-sequrity-api-key")
base_url = os.getenv("SEQURITY_BASE_URL", None)

assert openrouter_api_key != "your-openrouter-api-key", "Please set your OPENROUTER_API_KEY environment variable."
assert sequrity_key != "your-sequrity-api-key", "Please set your SEQURITY_API_KEY environment variable."

client = SequrityClient(api_key=sequrity_key, base_url=base_url)

Configure Security Settings

Set up security features, policies, and configurations:

    # Configure security features — only features is needed to select dual-llm arch.
    # security_policy is optional (server uses defaults).
    features = FeaturesHeader.dual_llm()
    fine_grained_config = FineGrainedConfigHeader(fsm=FsmOverrides(max_n_turns=10, disable_rllm=True))

• Features: Use Dual-LLM mode for enhanced security
• Security Policy: Apply default security policies (you can customize these with SQRT)
• Fine-Grained Config: Limit execution to 10 turns

Prepare Initial State and Node Functions

    # Define initial state
    initial_state: dict = {
        "query": "Find all users with recent orders",
        "tables": "",
        "schema": "",
        "sql_query": "",
        "result": "",
        "needs_validation": False,
    }

    # Prepare node functions dictionary
    # Note: Include routing functions (route_validation) as well as node functions
    node_functions = {
        "list_tables": list_tables,
        "get_schema": get_schema,
        "generate_query": generate_query,
        "validate_query": validate_query,
        "execute_query": execute_query,
        "route_validation": route_validation,  # Routing function for conditional edges
    }

Include Routing Functions

When your graph uses conditional edges (like add_conditional_edges), you must include the routing function in the node_functions dictionary. In this example, route_validation is the routing function that determines whether to go to validate_query or execute_query. Even though it's not a "node" in the traditional sense, it needs to be accessible during execution.

Execute the Graph

Finally, call compile_and_run_langgraph to execute your workflow securely:

    # Execute the graph with Sequrity
    result = client.control.langgraph.run(
        model="openai/gpt-5-mini",
        llm_api_key=openrouter_api_key,
        graph=graph,
        initial_state=initial_state,
        provider="openrouter",
        node_functions=node_functions,
        max_exec_steps=30,
        features=features,
        fine_grained_config=fine_grained_config,
    )

Understanding the Parameters

• model: The LLM model to use (can specify separate models for PLLM and QLLM)
• llm_api_key: API key for your LLM provider
• graph: Your LangGraph StateGraph instance
• initial_state: Starting state for the workflow
• provider: LLM provider — LlmServiceProvider enum or string literal (e.g., "openrouter", "openai")
• node_functions: Dictionary mapping node names to their functions
• max_exec_steps: Maximum execution steps (prevents infinite loops)
• features: Security features configuration
• security_policy: Security policies in SQRT
• fine_grained_config: Additional configuration options

Running the Example

Execute the example script and you should see output similar to:

======================================================================
🤖 SQL Agent with LangGraph + Sequrity Control API
======================================================================

📝 User query: Find all users with recent orders

🔄 Running workflow with Sequrity Control API...

📋 Listing available tables...
🔍 Getting schema for tables: users, orders, products
⚙️  Generating SQL for query: Find all users with recent orders
✓ Query looks safe, routing directly to execute_query
🚀 Executing SQL: SELECT * FROM users WHERE name LIKE '%recent%'

======================================================================
✨ Workflow Completed!
======================================================================

📊 Final State:
  • Tables: users, orders, products
  • SQL Query: SELECT * FROM users WHERE name LIKE '%recent%'
  • Result:
Query executed successfully!

SQL: SELECT * FROM users WHERE name LIKE '%recent%'

Results: Found 3 matching records:
  1. John Doe (john@example.com)
  2. Jane Smith (jane@example.com)
  3. Bob Johnson (bob@example.com)

✅ Success! The SQL agent workflow executed securely through Sequrity.

Additional Customizations

Custom Security Policies

You can define custom SQRT policies to restrict specific operations:

security_policy = SecurityPolicyHeader.dual_llm(
    codes=r"""
    // Prevent DELETE operations
    tool "execute_query" {
        hard deny when sql_query.value matching r".*DELETE.*";
    }

    // Tag sensitive table access
    let sensitive_tables = {"users", "payments"};
    tool "get_schema" {
        when tables.value overlaps sensitive_tables -> @tags |= {"sensitive"};
    }
    """
)

Multi-LLM Configuration

Use different models for planning and execution:

# Format: "pllm_model,qllm_model"
model = "openai/gpt-5-mini,anthropic/claude-sonnet-4.5"

Error Handling

The workflow automatically handles errors, but you can add custom error handling in your nodes:

def execute_query(state: SQLAgentState) -> dict:
    try:
        # Execute query logic
        result = execute_sql(state["sql_query"])
        return {"result": result}
    except Exception as e:
        return {"result": f"Error executing query: {str(e)}"}