Functor Logo
GitHub

FunctorClient Reference

The FunctorClient is the main entry point for interacting with the DRIP/KG-RAG system. It provides a clean, namespaced interface for all operations with comprehensive configuration options.

Client Initialization

Basic Usage

from functor_sdk import FunctorClient


# Simple initialization with API key
client = FunctorClient(api_key="your-api-key-here")


# Using environment variables (recommended)
client = FunctorClient()  # Uses FUNCTOR_API_KEY from environment

Full Configuration

from functor_sdk import FunctorClient


client = FunctorClient(
    base_url="http://localhost:8000",     # API base URL
    api_key="your-api-key",               # API key authentication
    token="your-bearer-token",            # Alternative: Bearer token
    timeout=30,                          # Request timeout in seconds
    max_retries=3,                       # Retry attempts for failed requests
    verify_ssl=True                      # SSL certificate verification
)

Authentication Methods

API Key (Recommended)

# Direct API key
client = FunctorClient(api_key="your-api-key")


# Environment variable
# export FUNCTOR_API_KEY=your-api-key
client = FunctorClient()

Bearer Token

# Direct token
client = FunctorClient(token="your-bearer-token")


# Environment variable
# export FUNCTOR_TOKEN=your-bearer-token
client = FunctorClient()

Configuration Options

base_url

Sets the base URL for API requests. Defaults to http://localhost:8000.

# Local development
client = FunctorClient(base_url="http://localhost:8000")


# Production
client = FunctorClient(base_url="https://api.functor.ai")


# Environment variable
# export FUNCTOR_BASE_URL=https://api.functor.ai

timeout

Request timeout in seconds. Defaults to 30.

# Short timeout for quick operations
client = FunctorClient(timeout=10)


# Longer timeout for complex queries
client = FunctorClient(timeout=120)

max_retries

Number of retry attempts for failed requests. Defaults to 3.

# Disable retries
client = FunctorClient(max_retries=0)


# More retries for unreliable networks
client = FunctorClient(max_retries=5)

verify_ssl

Whether to verify SSL certificates. Defaults to True.

# Disable SSL verification (development only)
client = FunctorClient(verify_ssl=False)


# Custom CA bundle
client = FunctorClient(verify_ssl="/path/to/ca-bundle.pem")

Context Managers

Both sync and async context managers are supported for proper resource management:

Synchronous Context Manager

from functor_sdk import FunctorClient


with FunctorClient(api_key="your-key") as client:
    result = client.queries.execute("What is AI?")
    print(result.answer)
# Client automatically cleaned up

Asynchronous Context Manager

import asyncio
from functor_sdk import FunctorClient


async def main():
    async with FunctorClient(api_key="your-key") as client:
        result = await client.queries.execute_async("What is AI?")
        print(result.answer)
    # Client automatically cleaned up


asyncio.run(main())

Namespaces

The client organizes functionality into logical namespaces:

client.queries

Execute natural language queries

execute()execute_async()

client.ingestion

Upload and process data

upload_file()upload_url()

client.knowledge_graphs

Manage knowledge graphs

list()get()

client.sources

Source management operations

list_for_kg()get_stats()delete()

client.visualizations

Access visualizations

list()get_metadata()

client.health

System health monitoring

check()check_detailed()

Environment Variables

Configure the client via environment variables:

.env
# Authentication
FUNCTOR_API_KEY=your-api-key-here
FUNCTOR_TOKEN=your-bearer-token


# Connection
FUNCTOR_BASE_URL=http://localhost:8000


# Optional: Custom timeout and retries
FUNCTOR_TIMEOUT=30
FUNCTOR_MAX_RETRIES=3

Advanced Configuration

Custom HTTP Client

import httpx
from functor_sdk import FunctorClient


# Custom HTTP client with specific settings
custom_client = httpx.Client(
    timeout=httpx.Timeout(60.0),
    limits=httpx.Limits(max_keepalive_connections=20)
)


# Note: FunctorClient manages its own HTTP client internally
# This is for reference only

Logging Configuration

import logging


# Enable debug logging
logging.basicConfig(level=logging.DEBUG)


# The SDK will log HTTP requests and responses
client = FunctorClient(api_key="your-key")

Error Handling

All client methods raise appropriate exceptions:

from functor_sdk import (
    FunctorClient,
    FunctorAPIError,
    FunctorAuthenticationError,
    FunctorNotFoundError,
    FunctorTimeoutError,
    FunctorConnectionError
)


client = FunctorClient()


try:
    result = client.queries.execute("test query")
except FunctorAuthenticationError:
    print("Authentication failed - check your API key")
except FunctorNotFoundError:
    print("Resource not found")
except FunctorTimeoutError:
    print("Request timed out")
except FunctorConnectionError:
    print("Connection failed")
except FunctorAPIError as e:
    print(f"API error: {e.status_code} - {e.message}")

Best Practices

1. Use Environment Variables

# Good: Use environment variables
client = FunctorClient()


# Avoid: Hardcoded credentials
client = FunctorClient(api_key="hardcoded-key")

2. Use Context Managers

# Good: Automatic cleanup
with FunctorClient() as client:
    result = client.queries.execute("query")


# Also good: Explicit cleanup
client = FunctorClient()
try:
    result = client.queries.execute("query")
finally:
    client.close()

3. Handle Errors Gracefully

from functor_sdk import FunctorClient, FunctorAPIError


client = FunctorClient()


try:
    result = client.queries.execute("query")
    print(result.answer)
except FunctorAPIError as e:
    print(f"Query failed: {e.message}")
    # Handle error appropriately

4. Configure Timeouts Appropriately

# Quick operations
client = FunctorClient(timeout=10)


# Complex queries
client = FunctorClient(timeout=120)


# Batch operations
client = FunctorClient(timeout=300)

Next Steps