sekha-python-sdk 0.2.0

Python SDK for Sekha AI Memory System

Sekha Python SDK

Official Python Client for Sekha Memory System

---

What is Sekha Python SDK?

Official Python client library for interacting with Sekha AI Memory System, providing unified access to Controller, MCP, and Bridge services.

Features:

• ✅ Unified Client Interface - Single client for all services (Controller, MCP, Bridge)
• ✅ Full Type Safety - Complete type hints with runtime validation
• ✅ Async/Await Support - Built on httpx with connection pooling
• ✅ Streaming Support - Server-sent events for LLM completions
• ✅ Automatic Retries - Exponential backoff with jitter
• ✅ Rate Limiting - Built-in token bucket rate limiter
• ✅ 90+% Test Coverage - Comprehensive test suite (2,000+ lines)
• ✅ Complete API Coverage - 19 Controller + 4 Bridge + 2 MCP endpoints
• ✅ 5 Convenience Workflows - High-level methods for common patterns

---

📚 Documentation

Complete SDK docs: docs.sekha.dev/sdks/python-sdk

• Python SDK Guide
• API Reference
• Code Examples
• Getting Started
• Changelog

---

🚀 Quick Start

Installation

# From PyPI
pip install sekha-python-sdk

# From source
git clone https://github.com/sekha-ai/sekha-python-sdk.git
cd sekha-python-sdk
pip install -e .

Basic Usage - Unified Client (Recommended)

from sekha import SekhaClient

# Initialize unified client with all services
client = SekhaClient(
    controller_url="http://localhost:8080",
    api_key="sk-your-api-key-here",
    bridge_url="http://localhost:5001",  # Optional
)

# ===== CONTROLLER: Memory Operations =====
await client.controller.create_conversation({
    "label": "My Conversation",
    "messages": [
        {"role": "user", "content": "Hello Sekha!"},
        {"role": "assistant", "content": "Hello! I'll remember this."}
    ]
})

# ===== BRIDGE: LLM Completions =====
response = await client.bridge.complete(
    messages=[
        {"role": "user", "content": "Explain quantum computing"}
    ],
    model="gpt-4",
    temperature=0.7
)
print(response["choices"][0]["message"]["content"])

# ===== BRIDGE: Streaming Completions =====
async for chunk in await client.bridge.stream_complete(
    messages=[{"role": "user", "content": "Tell me a story"}],
    model="gpt-4"
):
    print(chunk["choices"][0]["delta"].get("content", ""), end="")

# ===== MCP: Memory Statistics =====
stats = await client.mcp.memory_stats({
    "labels": ["important"],
    "start_date": "2026-01-01T00:00:00Z"
})
print(f"Total conversations: {stats['total_conversations']}")

# ===== MCP: Memory Search =====
results = await client.mcp.memory_search({
    "query": "project architecture",
    "limit": 5,
    "labels": ["technical"]
})
for result in results["results"]:
    print(f"{result['label']}: {result['content']}")

Unified Workflow Methods (NEW in v0.2.0)

High-level convenience methods that coordinate multiple services:

# 1. Store conversation and immediately search
results = await client.store_and_query(
    messages=[
        {"role": "user", "content": "Discussed project timeline"},
        {"role": "assistant", "content": "2 week sprint cycle"}
    ],
    query="timeline",
    label="Planning"
)

# 2. Assemble context from memory + generate LLM completion
response = await client.complete_with_context(
    prompt="Continue our architecture discussion",
    context_query="architecture decisions",
    model="gpt-4",
    context_budget=4000
)

# 3. Search memory + use results in LLM prompt
response = await client.complete_with_memory(
    prompt="Summarize our past discussions about:",
    search_query="architecture microservices",
    model="gpt-4",
    limit=5
)

# 4. Stream LLM response with assembled context
async for chunk in await client.stream_with_context(
    prompt="Explain our deployment strategy",
    context_query="deployment docker kubernetes",
    model="gpt-4"
):
    print(chunk["choices"][0]["delta"].get("content", ""), end="")

# 5. Health check all services concurrently
health = await client.health_check()
print(f"Controller: {health['controller']['status']}")
print(f"Bridge: {health['bridge']['status']}")

Basic Usage - Memory Controller Only

from sekha import MemoryController

# Direct controller client (no Bridge/MCP)
client = MemoryController(
    base_url="http://localhost:8080",
    api_key="sk-your-api-key-here",
    timeout=30.0,
    max_retries=3
)

# Store a conversation
conversation = await client.create_conversation({
    "label": "My First Conversation",
    "folder": "/personal",
    "messages": [
        {"role": "user", "content": "Hello Sekha!"},
        {"role": "assistant", "content": "Hello! I'll remember this."}
    ]
})

# Search semantically
results = await client.query(
    query="What did we discuss?",
    limit=5
)

# Assemble context for next LLM call
context = await client.assemble_context(
    query="Continue our conversation",
    context_budget=4000,
    preferred_labels=["important"]
)

Async Context Manager

from sekha import SekhaClient

async with SekhaClient(
    controller_url="http://localhost:8080",
    api_key="sk-your-api-key",
    bridge_url="http://localhost:5001"
) as client:
    # All clients automatically close on exit
    await client.controller.create_conversation({...})
    await client.bridge.complete(messages=[...])
    await client.mcp.memory_stats({})

Factory Function

from sekha import create_sekha_client

# Convenient factory function
client = create_sekha_client(
    controller_url="http://localhost:8080",
    api_key="sk-your-api-key",
    bridge_url="http://localhost:5001",
    timeout=60.0
)

---

📋 Complete API Coverage

Controller (Memory Operations) - 19 Endpoints

Conversation Management (9 endpoints)

• ✅ create_conversation - Store new conversations with messages
• ✅ get_conversation - Retrieve conversation by ID
• ✅ list_conversations - List with filtering and pagination
• ✅ update_label - Update label and folder
• ✅ update_folder - Move to different folder
• ✅ pin_conversation - Pin important conversations
• ✅ archive_conversation - Archive old conversations
• ✅ delete_conversation - Permanently delete
• ✅ count_conversations - Get total count

Search & Query (3 endpoints)

• ✅ query - Semantic search using vector similarity
• ✅ full_text_search - SQLite FTS5 full-text search
• ✅ rebuild_embeddings - Trigger embedding rebuild

Memory Orchestration (5 endpoints)

• ✅ assemble_context - Intelligent context assembly for LLMs
• ✅ summarize - Generate hierarchical summaries
• ✅ prune_dry_run - Get pruning suggestions
• ✅ prune_execute - Execute pruning operations
• ✅ suggest_labels - AI-powered label suggestions

Health & Metrics (2 endpoints)

• ✅ health - Health check endpoint
• ✅ metrics - Prometheus metrics

Bridge (LLM Integration) - 4 Endpoints

• ✅ complete - Generate chat completions (OpenAI-compatible)
• ✅ stream_complete - Streaming chat completions with SSE
• ✅ embed - Generate text embeddings
• ✅ health - Bridge service health check

MCP (Model Context Protocol) - 2 Endpoints

• ✅ memory_stats - Get memory statistics with filtering
• ✅ memory_search - Semantic memory search with pagination

Unified Workflows - 5 Convenience Methods

• ✅ store_and_query - Store conversation and immediately search
• ✅ complete_with_context - Assemble context + generate completion
• ✅ complete_with_memory - Search memory + use in prompt
• ✅ stream_with_context - Stream completion with context
• ✅ health_check - Check all services concurrently

Complete API Reference

---

🎯 Type Safety

The SDK provides comprehensive type safety:

from sekha.types import (
    # Core Models
    Message, MessageContent, ContentPart,
    Conversation, ConversationStatus,
    MessageRole,
    
    # Request Types
    CreateConversationRequest,
    QueryRequest,
    ContextAssembleRequest,
    PruneRequest,
    
    # Response Types
    QueryResponse,
    SearchResult,
    PruneResponse,
    SummaryResponse,
    
    # Enums
    SummaryLevel,
    PruneRecommendation,
)

from sekha.type_guards import (
    is_string_content,
    is_multi_modal_content,
    extract_text,
    extract_image_urls,
    has_images,
    has_text,
)

Type Guards provide runtime validation:

from sekha.type_guards import is_valid_role, extract_text

if is_valid_role("user"):
    # TypeScript-style type narrowing
    message = {"role": "user", "content": "Hello"}

# Extract text from multi-modal content
text = extract_text(message["content"])

---

🔧 Configuration

SekhaConfig (Unified Client)

from sekha import SekhaConfig, SekhaClient

# Full configuration options
config = SekhaConfig(
    controller_url="http://localhost:8080",
    api_key="sk-controller-key",          # Required
    bridge_url="http://localhost:5001",   # Optional
    bridge_api_key="bridge-key",          # Optional
    mcp_url="http://localhost:8080",      # Optional (defaults to controller)
    mcp_api_key="sk-mcp-key",            # Optional
    timeout=30.0,
    max_retries=3,
)

client = SekhaClient(config)

ClientConfig (Individual Clients)

from sekha.types import ClientConfig
from sekha import MemoryController

config = ClientConfig(
    base_url="http://localhost:8080",
    api_key="sk-your-api-key-here",         # Required, min 32 chars
    timeout=30.0,                           # Request timeout in seconds
    max_retries=3,                          # Max retry attempts
    default_label="MyApp",                  # Default conversation label
    rate_limit_requests=1000,               # Max requests per window
    rate_limit_window=60.0,                 # Rate limit window in seconds
)

client = MemoryController(config)

---

🛡️ Error Handling

The SDK provides specific error types:

from sekha import (
    SekhaError,              # Base error
    SekhaAPIError,           # API errors (4xx, 5xx)
    SekhaAuthError,          # Authentication failures (401)
    SekhaConnectionError,    # Connection/timeout errors
    SekhaNotFoundError,      # Resource not found (404)
    SekhaValidationError,    # Invalid input (400)
)

try:
    await client.controller.get_conversation(conversation_id)
except SekhaNotFoundError:
    print("Conversation not found")
except SekhaAuthError:
    print("Invalid API key")
except SekhaConnectionError:
    print("Controller unreachable")
except SekhaError as e:
    print(f"Unexpected error: {e}")

---

🧪 Development

# Clone repository
git clone https://github.com/sekha-ai/sekha-python-sdk.git
cd sekha-python-sdk

# Install dev dependencies
pip install -e ".[dev]"

# Run tests (unit + integration)
pytest

# Run only unit tests
pytest tests/ -m "not integration"

# Run with coverage
pytest --cov=sekha --cov-report=html --cov-report=term

# Type checking
mypy sekha/

# Linting
ruff check .
black --check .

# Format code
black .
ruff check --fix .

Running Integration Tests

Integration tests run against a real Sekha controller:

# Start controller locally (see sekha-controller docs)
docker compose up -d

# Set environment variables
export SEKHA_INTEGRATION_TESTS=1
export SEKHA_BASE_URL=http://localhost:8080
export SEKHA_API_KEY=your-test-key

# Run integration tests
pytest tests/test_all_endpoints.py

---

📁 Project Structure

sekha-python-sdk/
├── sekha/
│   ├── __init__.py           # Public API exports
│   ├── client.py             # MemoryController (main client)
│   ├── unified.py            # SekhaClient (unified interface)
│   ├── types.py              # Type definitions (dataclasses)
│   ├── models.py             # Legacy Pydantic models
│   ├── type_guards.py        # Runtime type validation
│   ├── errors.py             # Exception hierarchy
│   └── utils.py              # Utilities (rate limiter, validators)
├── tests/
│   ├── conftest.py                  # Pytest fixtures
│   ├── test_client_complete.py      # Controller tests
│   ├── test_bridge_client.py        # Bridge tests (528 lines)
│   ├── test_mcp_client.py           # MCP tests (442 lines)
│   ├── test_unified_workflows.py    # Workflow tests (616 lines)
│   ├── test_type_guards.py          # Type guard tests
│   ├── test_unified.py              # Unified client tests
│   ├── test_utils_coverage.py       # Utils tests
│   └── test_all_endpoints.py        # Integration tests
├── pyproject.toml            # Project config
├── CHANGELOG.md              # Version history
└── README.md                 # This file

---

🗺️ Roadmap

• Batch operations for bulk creates/updates
• Connection pooling optimizations
• WebSocket support for real-time updates
• Enhanced caching layer

---

🔗 Links

• Main Repo: sekha-controller
• Docs: docs.sekha.dev
• Website: sekha.dev
• Discord: discord.gg/sekha
• Changelog: CHANGELOG.md

---

📄 License

AGPL-3.0 - License Details

---

🤝 Contributing

Contributions are welcome! Please see CONTRIBUTING.md for guidelines.

1. Fork the repository
2. Create a feature branch (git checkout -b feature/amazing-feature)
3. Commit your changes (git commit -m 'Add amazing feature')
4. Push to the branch (git push origin feature/amazing-feature)
5. Open a Pull Request

Please ensure:

• All tests pass (pytest)
• Code is formatted (black .)
• Type checks pass (mypy sekha/)
• Coverage remains above 90%

---

📝 Release Notes

v0.2.0 (Current Release)

Major Features:

• ✅ Complete BridgeClient implementation (LLM completions, embeddings, streaming)
• ✅ Complete MCPClient implementation (memory stats, search)
• ✅ 5 unified workflow convenience methods
• ✅ Comprehensive test suite (2,000+ lines)
• ✅ Full async/await support across all clients
• ✅ Streaming support for LLM completions

Full Changelog