Intelligent Search

Intelligent Search uses Claude Agent with specialized graph tools to answer natural language questions about your knowledge graph. It combines semantic search, graph traversal, and reasoning to provide comprehensive answers with supporting evidence.

Overview

Intelligent Search is an AI-powered search feature that allows you to query your context graph using natural language. Unlike traditional keyword search, it:

Understands Intent: Interprets what you’re actually asking
Multi-Turn Reasoning: Can explore the graph across multiple steps
Tool Use: Leverages semantic search, graph queries, and traversal
Conversational: Supports follow-up questions in sessions
Evidence-Based: Returns both natural language answers and structured graph data

Intelligent Search is powered by Claude and uses an agentic approach with access to graph exploration tools. It’s ideal for complex queries that require reasoning across multiple nodes and relationships.

Quick Start

from kubiya import ControlPlaneClient

# Initialize the client
client = ControlPlaneClient(api_key="your-api-key")

# Perform intelligent search
result = client.graph.intelligent_search(
    keywords="Find all production databases with high availability"
)

# Get natural language answer
print(result["answer"])
# Output: "I found 3 production databases with high availability..."

# Access structured results
for node in result["nodes"]:
    print(f"Database: {node['properties']['name']}")

Core Concepts

How It Works

Query Understanding: Claude analyzes your natural language query
Tool Selection: Agent chooses appropriate graph tools (semantic search, traversal, etc.)
Iterative Exploration: Makes multiple tool calls to gather comprehensive information
Answer Synthesis: Combines findings into a natural language answer with evidence

Available Tools

The Claude Agent has access to:

Semantic Search: Vector-based search for finding relevant nodes
Graph Traversal: Navigate relationships between nodes
Property Filtering: Filter nodes by labels and properties
Cypher Queries: (Optional) Execute custom graph queries

Sessions

Sessions enable conversational follow-up:

Each search creates a session with a unique session_id
Continue conversations by providing the session_id
Sessions remember context from previous questions
Clean up sessions when done to free resources

Basic Usage

Simple Search

from kubiya import ControlPlaneClient

client = ControlPlaneClient(api_key="your-api-key")

# Ask a question
result = client.graph.intelligent_search(
    keywords="What EC2 instances are running in us-east-1?"
)

print("Answer:", result["answer"])
print(f"Found {len(result['nodes'])} nodes")
print(f"Confidence: {result['confidence']}")

Example Response

{
  "answer": "I found 5 EC2 instances running in the us-east-1 region...",
  "nodes": [
    {
      "id": "i-1234567890abcdef0",
      "labels": ["EC2Instance", "AWS"],
      "properties": {
        "name": "prod-web-01",
        "region": "us-east-1",
        "instance_type": "t3.large",
        "state": "running"
      }
    }
  ],
  "relationships": [],
  "tool_calls": [
    {
      "tool": "semantic_search",
      "query": "EC2 instances us-east-1",
      "results": 5
    }
  ],
  "turns_used": 2,
  "confidence": "high",
  "suggestions": [
    "Would you like to see the security groups for these instances?",
    "Should I check the CPU utilization?"
  ],
  "session_id": "session-abc123"
}

Search with Filters

from kubiya import ControlPlaneClient

client = ControlPlaneClient(api_key="your-api-key")

# Filter by integration and label
result = client.graph.intelligent_search(
    keywords="Find all databases",
    integration="AWS",
    label_filter="RDS"
)

print(f"Found {len(result['nodes'])} RDS databases in AWS")

Control Exploration Depth

from kubiya import ControlPlaneClient

client = ControlPlaneClient(api_key="your-api-key")

# Allow more turns for complex queries
result = client.graph.intelligent_search(
    keywords="Trace the dependency chain from the frontend service to all databases",
    max_turns=10  # Allow deeper exploration
)

print(f"Agent used {result['turns_used']} turns to explore the graph")

Adjust Model Creativity

from kubiya import ControlPlaneClient

client = ControlPlaneClient(api_key="your-api-key")

# Lower temperature for more focused answers
result = client.graph.intelligent_search(
    keywords="List all production services",
    temperature=0.3  # More deterministic (0.0-2.0)
)

# Higher temperature for more creative exploration
result = client.graph.intelligent_search(
    keywords="What patterns do you see in our infrastructure?",
    temperature=1.2  # More creative
)

Multi-Turn Conversations

Basic Session

from kubiya import ControlPlaneClient

client = ControlPlaneClient(api_key="your-api-key")

# Initial question
result1 = client.graph.intelligent_search(
    keywords="What Kubernetes clusters do we have?"
)
session_id = result1["session_id"]

print("Initial answer:", result1["answer"])

# Follow-up question (same session)
result2 = client.graph.intelligent_search(
    keywords="How many pods are running in each cluster?",
    session_id=session_id
)

print("Follow-up answer:", result2["answer"])

# Clean up session when done
client.graph.delete_search_session(session_id)

Session Management

from kubiya import ControlPlaneClient

client = ControlPlaneClient(api_key="your-api-key")

# List all active sessions
sessions = client.graph.list_search_sessions()
print(f"Active sessions: {len(sessions)}")

# Get session details
session_info = client.graph.get_search_session(session_id="session-abc123")
print(f"Turns used: {session_info['turns_used']}/{session_info['max_turns']}")
print(f"Status: {session_info['status']}")

# Delete specific session
client.graph.delete_search_session(session_id="session-abc123")

# List including expired
all_sessions = client.graph.list_search_sessions(include_expired=True)

Advanced Features

Enable Cypher Queries

from kubiya import ControlPlaneClient

client = ControlPlaneClient(api_key="your-api-key")

# Allow agent to use Cypher for complex queries
result = client.graph.intelligent_search(
    keywords="Find all paths between the frontend and database services",
    enable_cypher_queries=True  # Use with caution!
)

# Review what queries were executed
for tool_call in result["tool_calls"]:
    if tool_call["tool"] == "cypher_query":
        print(f"Executed: {tool_call['query']}")

Enabling Cypher queries gives the agent more power but should be used carefully in production environments. Consider performance and security implications.

Disable Semantic Search

from kubiya import ControlPlaneClient

client = ControlPlaneClient(api_key="your-api-key")

# Use only graph traversal (no vector search)
result = client.graph.intelligent_search(
    keywords="Show me the organizational hierarchy",
    enable_semantic_search=False  # Rely on graph structure only
)

Use Specific Model

from kubiya import ControlPlaneClient

client = ControlPlaneClient(api_key="your-api-key")

# Use specific LiteLLM model
result = client.graph.intelligent_search(
    keywords="Analyze our infrastructure security posture",
    model="anthropic/claude-opus-4"  # Use more powerful model
)

Practical Examples

1. Infrastructure Audit

Perform comprehensive infrastructure audit:

from kubiya import ControlPlaneClient

def audit_infrastructure(client: ControlPlaneClient):
    """Perform multi-step infrastructure audit using intelligent search."""

    # Start audit session
    result1 = client.graph.intelligent_search(
        keywords="What AWS resources do we have in production?",
        max_turns=5
    )
    session_id = result1["session_id"]

    print("=== Production Resources ===")
    print(result1["answer"])
    print()

    # Follow up on security
    result2 = client.graph.intelligent_search(
        keywords="Which of these resources have public internet access?",
        session_id=session_id
    )

    print("=== Public Access Analysis ===")
    print(result2["answer"])
    print()

    # Check compliance
    result3 = client.graph.intelligent_search(
        keywords="Are all resources properly tagged?",
        session_id=session_id
    )

    print("=== Tagging Compliance ===")
    print(result3["answer"])

    # Clean up
    client.graph.delete_search_session(session_id)

    return {
        "resources": result1["nodes"],
        "public_access": result2["nodes"],
        "compliance": result3["answer"]
    }

# Usage
client = ControlPlaneClient(api_key="your-api-key")
audit_results = audit_infrastructure(client)

2. Dependency Analysis

Trace service dependencies:

from kubiya import ControlPlaneClient

def analyze_service_dependencies(
    client: ControlPlaneClient,
    service_name: str
):
    """Analyze dependencies for a service."""

    # Find the service
    result1 = client.graph.intelligent_search(
        keywords=f"Find the service named {service_name}",
        label_filter="Service",
        max_turns=3
    )

    if not result1["nodes"]:
        print(f"Service {service_name} not found")
        return None

    session_id = result1["session_id"]

    # Get downstream dependencies
    result2 = client.graph.intelligent_search(
        keywords="What services and databases does this service depend on?",
        session_id=session_id,
        max_turns=5
    )

    print(f"=== Dependencies for {service_name} ===")
    print(result2["answer"])
    print(f"\nFound {len(result2['nodes'])} dependent resources")

    # Get upstream consumers
    result3 = client.graph.intelligent_search(
        keywords="What services consume or call this service?",
        session_id=session_id
    )

    print("\n=== Consumers ===")
    print(result3["answer"])

    # Clean up
    client.graph.delete_search_session(session_id)

    return {
        "service": result1["nodes"][0] if result1["nodes"] else None,
        "dependencies": result2["nodes"],
        "consumers": result3["nodes"]
    }

# Usage
client = ControlPlaneClient(api_key="your-api-key")
deps = analyze_service_dependencies(client, "auth-service")

3. Cost Analysis

Analyze infrastructure costs:

from kubiya import ControlPlaneClient

def analyze_costs(client: ControlPlaneClient, integration: str = "AWS"):
    """Analyze resource costs using intelligent search."""

    result1 = client.graph.intelligent_search(
        keywords=f"What are the most expensive resources in {integration}?",
        integration=integration,
        max_turns=7
    )

    session_id = result1["session_id"]

    print("=== Cost Analysis ===")
    print(result1["answer"])

    # Follow up on optimization
    result2 = client.graph.intelligent_search(
        keywords="Which resources are underutilized and could be downsized?",
        session_id=session_id
    )

    print("\n=== Optimization Opportunities ===")
    print(result2["answer"])

    # Get recommendations
    result3 = client.graph.intelligent_search(
        keywords="What cost-saving recommendations do you have?",
        session_id=session_id
    )

    print("\n=== Recommendations ===")
    print(result3["answer"])

    # Suggestions for next steps
    if result3.get("suggestions"):
        print("\n=== Suggested Follow-ups ===")
        for suggestion in result3["suggestions"]:
            print(f"  - {suggestion}")

    # Clean up
    client.graph.delete_search_session(session_id)

    return result3

# Usage
client = ControlPlaneClient(api_key="your-api-key")
cost_analysis = analyze_costs(client, integration="AWS")

4. Security Investigation

Investigate security concerns:

from kubiya import ControlPlaneClient

def investigate_security(
    client: ControlPlaneClient,
    concern: str
):
    """Investigate security concerns using intelligent search."""

    result1 = client.graph.intelligent_search(
        keywords=f"Investigate: {concern}",
        max_turns=10,
        enable_cypher_queries=True,  # Allow deep investigation
        temperature=0.5  # More focused
    )

    session_id = result1["session_id"]

    print(f"=== Security Investigation: {concern} ===")
    print(result1["answer"])
    print(f"\nConfidence: {result1['confidence']}")
    print(f"Turns used: {result1['turns_used']}")

    # Analyze blast radius
    if result1["nodes"]:
        result2 = client.graph.intelligent_search(
            keywords="What other resources could be affected if these resources are compromised?",
            session_id=session_id
        )

        print("\n=== Blast Radius ===")
        print(result2["answer"])

    # Get remediation
    result3 = client.graph.intelligent_search(
        keywords="What steps should be taken to remediate this?",
        session_id=session_id
    )

    print("\n=== Remediation Steps ===")
    print(result3["answer"])

    # Clean up
    client.graph.delete_search_session(session_id)

    return {
        "findings": result1["nodes"],
        "affected_resources": result2.get("nodes", []),
        "remediation": result3["answer"],
        "tool_calls": result1["tool_calls"] + result2.get("tool_calls", []) + result3.get("tool_calls", [])
    }

# Usage
client = ControlPlaneClient(api_key="your-api-key")
investigation = investigate_security(
    client,
    "Security groups allowing unrestricted SSH access"
)

5. Compliance Checker

Check compliance across infrastructure:

from kubiya import ControlPlaneClient
from typing import List, Dict

def check_compliance(
    client: ControlPlaneClient,
    compliance_rules: List[str]
):
    """Check infrastructure compliance using intelligent search."""

    results = []

    for rule in compliance_rules:
        result = client.graph.intelligent_search(
            keywords=f"Check compliance: {rule}",
            max_turns=5,
            temperature=0.3  # Precise
        )

        status = "✅ PASS" if result["confidence"] == "high" and "compliant" in result["answer"].lower() else "❌ FAIL"

        print(f"\n{status} {rule}")
        print(f"Finding: {result['answer']}")

        results.append({
            "rule": rule,
            "status": status,
            "answer": result["answer"],
            "evidence": result["nodes"],
            "confidence": result["confidence"]
        })

    # Summary
    passed = sum(1 for r in results if "PASS" in r["status"])
    failed = len(results) - passed

    print(f"\n=== Compliance Summary ===")
    print(f"Total Rules: {len(results)}")
    print(f"Passed: {passed}")
    print(f"Failed: {failed}")
    print(f"Score: {(passed / len(results) * 100):.1f}%")

    return results

# Usage
client = ControlPlaneClient(api_key="your-api-key")

rules = [
    "All S3 buckets have encryption enabled",
    "All EC2 instances have required tags",
    "No security groups allow 0.0.0.0/0 on port 22",
    "All RDS instances have automated backups enabled"
]

compliance_results = check_compliance(client, rules)

Error Handling

from kubiya import ControlPlaneClient
from kubiya.resources.exceptions import GraphError

client = ControlPlaneClient(api_key="your-api-key")

# Handle search errors
try:
    result = client.graph.intelligent_search(
        keywords="Complex query",
        max_turns=20
    )
except GraphError as e:
    if "timeout" in str(e).lower():
        print(f"Search timed out: {e}")
        # Consider reducing max_turns or simplifying query
    else:
        print(f"Search failed: {e}")

# Handle session errors
try:
    session_info = client.graph.get_search_session(session_id="invalid-id")
except GraphError as e:
    if "not found" in str(e).lower():
        print("Session expired or not found")
    else:
        print(f"Session error: {e}")

# Handle invalid parameters
try:
    result = client.graph.intelligent_search(
        keywords="test",
        max_turns=100  # Exceeds maximum
    )
except GraphError as e:
    print(f"Invalid parameters: {e}")

Best Practices

1. Clean Up Sessions

from kubiya import ControlPlaneClient

client = ControlPlaneClient(api_key="your-api-key")

session_id = None
try:
    result = client.graph.intelligent_search(keywords="query")
    session_id = result["session_id"]

    # Use session for follow-ups...

finally:
    if session_id:
        client.graph.delete_search_session(session_id)

2. Use Appropriate max_turns

# Simple queries - use fewer turns
result = client.graph.intelligent_search(
    keywords="List all databases",
    max_turns=3  # Quick answer
)

# Complex queries - allow more exploration
result = client.graph.intelligent_search(
    keywords="Analyze security posture across all cloud providers",
    max_turns=15  # Deep analysis
)

3. Monitor Tool Usage

result = client.graph.intelligent_search(keywords="query")

# Review what tools were used
print(f"Tools used: {len(result['tool_calls'])}")
for tool_call in result["tool_calls"]:
    print(f"  - {tool_call['tool']}: {tool_call.get('results', 'N/A')} results")

# Optimize based on tool usage

4. Handle Low Confidence

result = client.graph.intelligent_search(keywords="ambiguous query")

if result["confidence"] == "low":
    print(f"Low confidence answer: {result['answer']}")
    print("Suggested follow-ups:")
    for suggestion in result.get("suggestions", []):
        print(f"  - {suggestion}")

API Reference