intelligence-builder-sdk 1.0.0

Official Python SDK for the Intelligence Builder API

Intelligence Builder SDK

Official Python SDK for the Intelligence Builder API. This SDK provides a fully-featured async client for building knowledge graphs, semantic search, and AI-powered query capabilities.

Installation

pip install intelligence-builder-sdk

Quick Start

import asyncio
from intelligence_builder_sdk import IntelligenceBuilderClient, ClientConfig

async def main():
    # Configure the client
    config = ClientConfig(
        base_url="http://localhost:8100",
        api_key="your-api-key",  # pragma: allowlist secret
        timeout=30,
        max_retries=3,
    )

    # Create and connect the client
    client = IntelligenceBuilderClient(config)
    await client.connect()

    try:
        # Create an entity
        entity = await client.create_entity({
            "entity_type": "aws_resource",
            "name": "prod-ec2-instance",
            "description": "Production EC2 instance",
            "metadata": {
                "region": "us-east-1",
                "instance_type": "t3.large",
            },
            "tags": ["aws", "ec2", "production"],
        })
        print(f"Created entity: {entity.entity_id}")

        # Query the knowledge graph
        response = await client.query(
            query="What are the dependencies for my production EC2 instance?",
            context={"domain": "cloud_optimizer"},
        )
        print(f"Answer: {response.answer}")

    finally:
        await client.disconnect()

asyncio.run(main())

Features

Authentication

• API key-based authentication with automatic JWT token management
• Token refresh on expiry

Entity Management

• create_entity() - Create a new entity in the knowledge graph
• create_entities_batch() - Batch create multiple entities
• get_entity() - Retrieve an entity by ID
• list_entities() - List entities with pagination and filtering
• search_entities() - Semantic search across entities

Relationship Management

• create_relationship() - Create relationships between entities
• create_relationships_batch() - Batch create relationships
• get_relationship() - Retrieve a relationship by ID
• list_relationships() - List relationships with filtering
• get_entity_relationships() - Get all relationships for an entity

Graph Operations

• traverse_graph() - Traverse the knowledge graph from a starting entity
• find_shortest_path() - Find the shortest path between two entities
• find_all_paths() - Find all paths between entities
• get_subgraph() - Extract a subgraph around an entity
• get_entity_neighbors() - Get immediate neighbors of an entity

Query & Search

• query() - Intelligent queries with context awareness
• search_graph() - Semantic search across the knowledge graph

Document Ingestion

• ingest_document() - Ingest documents for knowledge extraction

Reliability Features

• Circuit Breaker: Automatic failure detection and recovery
• Retry Logic: Configurable retry with exponential backoff
• Response Caching: Memory, Redis, or file-based caching
• Rate Limiting: Handles rate limit responses with retry-after

Configuration

from intelligence_builder_sdk import ClientConfig

config = ClientConfig(
    # Required
    base_url="http://localhost:8100",
    api_key="your-api-key",  # pragma: allowlist secret

    # Timeouts
    timeout=30,  # Request timeout in seconds

    # Retry Configuration
    max_retries=3,  # Maximum retry attempts

    # Caching
    enable_caching=True,
    cache_ttl=300,  # Cache TTL in seconds
    cache_backend="memory",  # Options: memory, redis, file
    redis_url="redis://localhost:6379/0",  # Required if cache_backend="redis"

    # Circuit Breaker
    enable_circuit_breaker=True,
    circuit_breaker_threshold=5,  # Failures before opening
    circuit_breaker_timeout=60,  # Timeout before attempting recovery
)

Exception Handling

from intelligence_builder_sdk import (
    IntelligenceBuilderError,
    AuthenticationError,
    RateLimitError,
    ServiceUnavailableError,
    ValidationError,
    CircuitBreakerOpenError,
)

try:
    response = await client.query("my query")
except AuthenticationError:
    print("Invalid API key or token expired")
except RateLimitError as e:
    print(f"Rate limited. Retry after {e.retry_after} seconds")
except ServiceUnavailableError:
    print("Service temporarily unavailable")
except ValidationError:
    print("Invalid request parameters")
except CircuitBreakerOpenError:
    print("Circuit breaker is open - too many failures")
except IntelligenceBuilderError:
    print("General SDK error")

Context Manager Support

async with IntelligenceBuilderClient(config) as client:
    entity = await client.create_entity({...})

Models

Entity

from intelligence_builder_sdk import Entity

entity: Entity
print(entity.entity_id)  # UUID
print(entity.name)
print(entity.entity_type)
print(entity.description)
print(entity.metadata)  # Dict[str, Any]
print(entity.tags)  # List[str]
print(entity.created_at)  # datetime

Relationship

from intelligence_builder_sdk import Relationship

rel: Relationship
print(rel.relationship_id)  # UUID
print(rel.from_entity_id)
print(rel.to_entity_id)
print(rel.relationship_type)
print(rel.weight)  # float 0.0-1.0
print(rel.confidence)  # float 0.0-1.0
print(rel.properties)  # Dict[str, Any]

GraphPath

from intelligence_builder_sdk import GraphPath

path: GraphPath
print(path.entities)  # List[UUID]
print(path.relationships)  # List[UUID]
print(path.total_weight)  # float
print(path.length)  # int

Requirements

• Python 3.9+
• httpx
• pydantic
• pybreaker
• cachetools

License

MIT License - Copyright 2025 Intelligence Builder

Links

• GitHub Repository
• API Documentation
• Issue Tracker