LangGraph Integration

Integrate Shadow Executor with LangGraph agents (TypeScript and Python).

Overview

LangGraph is LangChain's framework for building stateful, multi-actor AI agents. Shadow Executor provides tool wrappers that enforce policies before tool execution.

Installation

TypeScript

npm install @shadow-executor/sdk @langchain/core

Python

Shadow Executor Python SDK coming in Milestone 2. For Milestone 1, copy the standalone integration file:

curl -O https://raw.githubusercontent.com/shadow-executor/shadow-executor/main/packages/sdk/src/langgraph/python.py

TypeScript Integration

Basic Usage

import { createShadowExecutorTool } from '@shadow-executor/sdk/langgraph';
import { ChatOpenAI } from '@langchain/openai';
import { StateGraph } from '@langchain/langgraph';

// Define your base tool
async function deleteDatabase(params: { instance_id: string; environment: string }) {
  // Delete database logic
  console.log(`Deleting database ${params.instance_id}`);
  return { success: true };
}

// Wrap with Shadow Executor
const protectedDeleteDatabase = createShadowExecutorTool({
  name: 'aws_rds_delete_db_instance',
  description: 'Delete an RDS database instance',
  schema: z.object({
    instance_id: z.string(),
    environment: z.string(),
  }),
  policyPath: './shadow-exec.policy.yaml',
  logSecret: process.env.SHADOW_EXEC_LOG_SECRET!,
  baseHandler: deleteDatabase,
});

// Use in LangGraph agent
const tools = [protectedDeleteDatabase];

const model = new ChatOpenAI({
  modelName: 'gpt-4',
}).bindTools(tools);

const graph = new StateGraph({
  channels: {
    messages: {
      value: (left: BaseMessage[], right: BaseMessage[]) => left.concat(right),
      default: () => [],
    },
  },
})
  .addNode('agent', async (state) => {
    const result = await model.invoke(state.messages);
    return { messages: [result] };
  })
  .addNode('tools', async (state) => {
    const lastMessage = state.messages[state.messages.length - 1];
    // Tool execution handled by LangGraph
    return { messages: [] };
  });

const app = graph.compile();

Wrapping Multiple Tools

import { wrapLangGraphTools } from '@shadow-executor/sdk/langgraph';

const tools = wrapLangGraphTools(
  [
    {
      name: 'aws_rds_delete_db_instance',
      description: 'Delete RDS instance',
      schema: z.object({ instance_id: z.string() }),
      func: async (params) => { /* ... */ },
    },
    {
      name: 'aws_s3_delete_bucket',
      description: 'Delete S3 bucket',
      schema: z.object({ bucket_name: z.string() }),
      func: async (params) => { /* ... */ },
    },
  ],
  {
    policyPath: './shadow-exec.policy.yaml',
    logSecret: process.env.SHADOW_EXEC_LOG_SECRET!,
    enableIPIDetection: true,
  }
);

// Use tools in your LangGraph agent
const model = new ChatOpenAI({ modelName: 'gpt-4' }).bindTools(tools);

Python Integration

Basic Usage

from langgraph.prebuilt import ToolExecutor
from shadow_executor import create_shadow_executor_tool
import os

# Define your base tool
def delete_database(instance_id: str, environment: str) -> dict:
    """Delete an RDS database instance."""
    print(f"Deleting database {instance_id}")
    return {"success": True}

# Wrap with Shadow Executor
protected_delete_database = create_shadow_executor_tool(
    name="aws_rds_delete_db_instance",
    description="Delete an RDS database instance",
    func=delete_database,
    policy_path="./shadow-exec.policy.yaml",
    log_secret=os.environ["SHADOW_EXEC_LOG_SECRET"],
    enable_ipi_detection=True,
)

# Use in LangGraph agent
tools = [protected_delete_database]
tool_executor = ToolExecutor(tools)

# Define your graph
from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated, Sequence
import operator

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]

graph = StateGraph(AgentState)

def call_model(state):
    # Model logic
    pass

def call_tool(state):
    # Tool execution (Shadow Executor intercepts here)
    last_message = state["messages"][-1]
    tool_executor.invoke(last_message.tool_calls[0])
    return {"messages": []}

graph.add_node("agent", call_model)
graph.add_node("action", call_tool)
graph.add_edge("agent", "action")
graph.add_edge("action", END)

app = graph.compile()

Wrapping Multiple Tools

from shadow_executor import wrap_langgraph_tools

tools = wrap_langgraph_tools(
    [
        {
            "name": "aws_rds_delete_db_instance",
            "description": "Delete RDS instance",
            "func": delete_database,
        },
        {
            "name": "aws_s3_delete_bucket",
            "description": "Delete S3 bucket",
            "func": delete_bucket,
        },
    ],
    policy_path="./shadow-exec.policy.yaml",
    log_secret=os.environ["SHADOW_EXEC_LOG_SECRET"],
    enable_ipi_detection=True,
)

tool_executor = ToolExecutor(tools)

Configuration

createShadowExecutorTool Options (TypeScript)

interface ShadowExecutorToolConfig {
  /** Tool name (used for policy matching) */
  name: string;

  /** Tool description */
  description: string;

  /** Zod schema for parameters */
  schema: z.ZodObject<any>;

  /** Path to policy YAML file */
  policyPath: string;

  /** HMAC secret for audit log signing */
  logSecret: string;

  /** Path to audit log (default: ~/.shadow-exec/audit.ndjson) */
  logPath?: string;

  /** Enable IPI detection (default: false) */
  enableIPIDetection?: boolean;

  /** Agent ID for tracking (default: 'langgraph-agent') */
  agentId?: string;

  /** Base tool handler function */
  baseHandler: (params: any) => Promise<any>;

  /** Custom tool name to AgentAction converter */
  convertToAgentAction?: (name: string, params: any) => AgentAction;
}

create_shadow_executor_tool Options (Python)

def create_shadow_executor_tool(
    name: str,
    description: str,
    func: Callable,
    policy_path: str,
    log_secret: str,
    log_path: str = "~/.shadow-exec/audit.ndjson",
    enable_ipi_detection: bool = False,
    agent_id: str = "langgraph-agent",
) -> BaseTool:
    """Create a Shadow Executor protected LangGraph tool."""

Policy Examples

Block Production Database Deletion

- id: LG-001
  name: Block production database deletion
  severity: CRITICAL
  action: BLOCK
  match:
    service: rds
    operation: DeleteDBInstance
    resource_tags:
      Environment: production

Require Approval for IAM Changes

- id: LG-002
  name: Require approval for IAM changes
  severity: HIGH
  action: REQUIRE_APPROVAL
  match:
    service: iam
    operation: [AttachUserPolicy, PutUserPolicy]

Block High IPI Score Operations

- id: LG-003
  name: Block suspected IPI attacks
  severity: CRITICAL
  action: BLOCK
  match:
    operation: "Delete*"
    ipi_score: ">= 0.7"

Error Handling

TypeScript

import { BlockedActionError } from '@shadow-executor/sdk/langgraph';

try {
  await protectedDeleteDatabase({
    instance_id: 'prod-db-01',
    environment: 'production',
  });
} catch (error) {
  if (error instanceof BlockedActionError) {
    console.error('Action blocked by policy');
    console.error('Reason:', error.decision.reason);
    console.error('Rule ID:', error.decision.matched_rule_id);
  } else {
    throw error;
  }
}

Python

from shadow_executor import BlockedActionError

try:
    protected_delete_database(
        instance_id="prod-db-01",
        environment="production",
    )
except BlockedActionError as e:
    print(f"Action blocked by policy: {e.decision.reason}")
    print(f"Rule ID: {e.decision.matched_rule_id}")
except Exception as e:
    raise e

Approval Workflow

When a policy action is REQUIRE_APPROVAL, Shadow Executor creates an approval request file and waits for manual approval.

Approval file: ~/.shadow-exec/approvals/{request-id}.json

{
  "id": "550e8400-e29b-41d4-a716-446655440000",
  "status": "PENDING",
  "requested_at": "2026-05-06T12:00:00.000Z",
  "timeout_minutes": 30,
  "decision": {
    "action": "REQUIRE_APPROVAL",
    "matched_rule_id": "LG-002",
    "agent_action": {
      "service": "iam",
      "operation": "AttachUserPolicy",
      "parameters": {
        "user_name": "admin",
        "policy_arn": "arn:aws:iam::aws:policy/AdministratorAccess"
      }
    }
  }
}

Approve:

echo '{"status": "APPROVED"}' > ~/.shadow-exec/approvals/{request-id}.json

Deny:

echo '{"status": "DENIED"}' > ~/.shadow-exec/approvals/{request-id}.json

LangGraph agent will poll every 5 seconds and proceed/abort based on status.

Testing

TypeScript Test

import { createShadowExecutorTool } from '@shadow-executor/sdk/langgraph';
import { z } from 'zod';

// Mock tool
const mockDelete = async (params: { instance_id: string }) => {
  console.log(`Would delete ${params.instance_id}`);
  return { success: true };
};

// Wrap with test policy
const testTool = createShadowExecutorTool({
  name: 'aws_rds_delete_db_instance',
  description: 'Test tool',
  schema: z.object({ instance_id: z.string() }),
  policyPath: './test-policy.yaml',
  logSecret: 'test-secret',
  baseHandler: mockDelete,
});

// Test
try {
  await testTool.invoke({ instance_id: 'prod-db' });
} catch (error) {
  console.log('Blocked as expected:', error.message);
}

Python Test

from shadow_executor import create_shadow_executor_tool

# Mock tool
def mock_delete(instance_id: str) -> dict:
    print(f"Would delete {instance_id}")
    return {"success": True}

# Wrap with test policy
test_tool = create_shadow_executor_tool(
    name="aws_rds_delete_db_instance",
    description="Test tool",
    func=mock_delete,
    policy_path="./test-policy.yaml",
    log_secret="test-secret",
)

# Test
try:
    test_tool.invoke({"instance_id": "prod-db"})
except Exception as e:
    print(f"Blocked as expected: {e}")

Full Examples

See repository for complete examples:

• examples/langgraph/typescript/demo.ts
• examples/langgraph/python/demo.py

Next Steps

• Policy Reference — Define policies for LangGraph tools
• Audit Logging — Query and verify audit logs
• IPI Detection — Understand IPI scoring