Python SDK

The official Cortex Python SDK provides a convenient way to integrate real-time web search and content validation into your Python applications.

📦 Installation

pip install cortex-ai

Requirements

• Python 3.8+
• requests >= 2.25.0
• typing-extensions >= 4.0.0

🚀 Quick Start

from cortex import CortexClient

# Initialize client
client = CortexClient(api_key="cortex_sk_your_api_key_here")

# Search the web
result = client.search("latest AI developments")
print(result.summary)

# Extract content from URL
content = client.extract("https://example.com/article")
print(content.text)

# Validate a claim
validation = client.validate("The iPhone 15 supports USB-C")
print(f"Validation: {validation.result}")

🔧 Configuration

Environment Variables

# Set API key via environment variable
export CORTEX_API_KEY="cortex_sk_your_api_key_here"

from cortex import CortexClient

# Client automatically reads CORTEX_API_KEY
client = CortexClient()

Configuration File

# ~/.cortex/config.json
{
  "api_key": "cortex_sk_your_api_key_here",
  "base_url": "https://api.usecortex.co/v1",
  "timeout": 30,
  "max_retries": 3
}

client = CortexClient.from_config("~/.cortex/config.json")

Advanced Configuration

from cortex import CortexClient

client = CortexClient(
    api_key="cortex_sk_your_api_key_here",
    base_url="https://api.usecortex.co/v1",
    timeout=30,
    max_retries=3,
    retry_delay=1.0,
    user_agent="MyApp/1.0",
    verify_ssl=True
)

🔍 Search API

Basic Search

# Simple search
result = client.search("quantum computing breakthrough")
print(result.summary)
print(f"Sources: {len(result.sources)}")

Advanced Search

result = client.search(
    query="Tesla stock analysis",
    max_results=10,
    language="en",
    country="us",
    recency="week",
    include_sources=True,
    domain_filter={
        "include": ["reuters.com", "bloomberg.com"],
        "exclude": ["reddit.com"]
    }
)

# Access results
print(f"Summary: {result.summary}")
print(f"Confidence: {result.confidence_score}")
print(f"Processing time: {result.metadata.processing_time}s")

# Iterate through sources
for source in result.sources:
    print(f"- {source.title} ({source.confidence:.2f})")
    print(f"  {source.url}")
    print(f"  {source.snippet}")

Search with Context

result = client.search(
    query="machine learning model performance",
    context={
        "domain": "technology",
        "intent": "research",
        "complexity": "technical"
    },
    options={
        "deep_search": True,
        "enable_cache": True,
        "timeout": 20
    }
)

Streaming Search

# Stream search results as they arrive
def handle_progress(event):
    if event.type == "source":
        print(f"Found: {event.data.title}")
    elif event.type == "summary":
        print(f"Summary progress: {event.data.progress}%")

result = client.search_stream(
    query="breaking news AI",
    on_progress=handle_progress
)

# Wait for completion
final_result = result.wait()
print(final_result.summary)

📄 Extract API

URL Extraction

# Extract content from URL
content = client.extract("https://example.com/article")

print(f"Title: {content.metadata.title}")
print(f"Author: {content.metadata.author}")
print(f"Word count: {content.metadata.word_count}")
print(f"Content: {content.text}")

Advanced Extraction

content = client.extract(
    url="https://example.com/article",
    format="markdown",
    include_metadata=True,
    extract_images=True,
    extract_links=True,
    clean_html=True
)

# Access structured content
print(f"Title: {content.metadata.title}")
print(f"Reading time: {content.metadata.reading_time}")
print(f"Quality score: {content.quality_score}")

# Access images
for image in content.images:
    print(f"Image: {image.url} - {image.alt}")

# Access links
print(f"Internal links: {len(content.links.internal)}")
print(f"External links: {len(content.links.external)}")

Batch Extraction

urls = [
    "https://example1.com/article",
    "https://example2.com/article", 
    "https://example3.com/article"
]

# Extract multiple URLs concurrently
results = client.extract_batch(urls, max_workers=5)

for url, content in results.items():
    if content.success:
        print(f"{url}: {content.metadata.word_count} words")
    else:
        print(f"{url}: Failed - {content.error}")

PDF Extraction

# Extract from PDF
content = client.extract(
    url="https://example.com/document.pdf",
    options={
        "extract_pages": [1, 2, 3],
        "preserve_layout": False,
        "ocr_enabled": True
    }
)

print(content.text)

✅ Validate API

Basic Validation

# Validate a factual claim
result = client.validate("The Earth is round")

print(f"Result: {result.validation_result}")  # VERIFIED
print(f"Confidence: {result.confidence_score}")
print(f"Consensus: {result.consensus_level}")

Advanced Validation

result = client.validate(
    claim="Tesla delivered over 1.8 million vehicles in 2023",
    sources=[
        "https://ir.tesla.com/press-release/tesla-q4-2023",
        "reuters.com",
        "bloomberg.com"
    ],
    context={
        "domain": "automotive",
        "claim_type": "statistical",
        "time_sensitive": True
    },
    options={
        "confidence_threshold": 0.8,
        "require_primary_sources": True,
        "max_sources": 10
    }
)

# Analyze evidence
print(f"Validation: {result.validation_result}")
print(f"Supporting sources: {len(result.evidence.supporting)}")
print(f"Contradicting sources: {len(result.evidence.contradicting)}")

# Review evidence
for evidence in result.evidence.supporting:
    print(f"+ {evidence.source}")
    print(f"  Confidence: {evidence.confidence:.2f}")
    print(f"  Authority: {evidence.authority_score:.2f}")
    print(f"  Snippet: {evidence.snippet}")

Batch Validation

claims = [
    {"id": "claim1", "claim": "Python is open source"},
    {"id": "claim2", "claim": "The moon is made of cheese"},
    {"id": "claim3", "claim": "Bitcoin was created in 2009"}
]

results = client.validate_batch(claims)

for result in results:
    print(f"{result.id}: {result.validation_result}")

📊 Cache API

Query Cache

# Check if result is cached
cached = client.cache.query("AI developments today")

if cached.cache_hit:
    print("Using cached result")
    print(f"TTL remaining: {cached.ttl_remaining}s")
    return cached.results
else:
    # Perform new search
    result = client.search("AI developments today")
    
    # Store in cache
    client.cache.store(
        query="AI developments today",
        content=result,
        ttl=3600,  # 1 hour
        trust_level="high"
    )
    
    return result

Cache Management

# Get cache statistics
stats = client.cache.stats()
print(f"Hit rate: {stats.hit_rate:.2%}")
print(f"Storage used: {stats.storage_used}")

# Clear cache by tags
client.cache.clear(tags=["outdated", "temporary"])

# Clear old entries
from datetime import datetime, timedelta
cutoff = datetime.now() - timedelta(days=7)
client.cache.clear(older_than=cutoff)

📈 Monitoring

Usage Statistics

# Get usage stats
usage = client.monitoring.usage(period="month")

print(f"Requests used: {usage.requests_used}/{usage.requests_limit}")
print(f"Success rate: {usage.performance_metrics.success_rate:.2%}")
print(f"Avg response time: {usage.performance_metrics.avg_response_time:.2f}s")
print(f"Cache hit rate: {usage.performance_metrics.cache_hit_rate:.2%}")

Health Monitoring

# Check system health
health = client.monitoring.health()

if health.status == "healthy":
    print("All systems operational")
else:
    print("System issues detected:")
    for service, status in health.services.items():
        if status.status != "healthy":
            print(f"  {service}: {status.status}")

Audit Logs

# Get recent audit logs
logs = client.monitoring.audit(limit=100)

for entry in logs.entries:
    print(f"{entry.timestamp}: {entry.endpoint} - {entry.status}")
    if entry.status != 200:
        print(f"  Error: {entry.error}")

🚨 Error Handling

Exception Types

from cortex import (
    CortexError,           # Base exception
    AuthenticationError,   # 401 errors
    RateLimitError,       # 429 errors
    ValidationError,      # 400 errors
    ServerError          # 5xx errors
)

try:
    result = client.search("test query")
except AuthenticationError as e:
    print(f"Auth error: {e.message}")
    # Handle invalid API key
except RateLimitError as e:
    print(f"Rate limited: {e.retry_after}s")
    # Implement backoff
except ValidationError as e:
    print(f"Invalid request: {e.message}")
    # Fix request parameters
except ServerError as e:
    print(f"Server error: {e.message}")
    # Retry or report issue
except CortexError as e:
    print(f"General error: {e.code} - {e.message}")

Retry Logic

import time
from cortex import CortexClient, RateLimitError, ServerError

def search_with_retry(client, query, max_retries=3):
    for attempt in range(max_retries):
        try:
            return client.search(query)
        except RateLimitError as e:
            if attempt < max_retries - 1:
                time.sleep(e.retry_after or 2 ** attempt)
                continue
            raise
        except ServerError as e:
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)
                continue
            raise
    
    raise Exception("Max retries exceeded")

# Usage
result = search_with_retry(client, "AI developments")

🔧 Advanced Features

Custom Headers

client = CortexClient(
    api_key="your_key",
    default_headers={
        "X-Custom-Header": "MyApp/1.0",
        "X-Request-Source": "background-job"
    }
)

Request Hooks

def log_request(request):
    print(f"Making request to {request.url}")

def log_response(response):
    print(f"Response: {response.status_code} in {response.elapsed}s")

client = CortexClient(
    api_key="your_key",
    request_hooks=[log_request],
    response_hooks=[log_response]
)

Async Support

import asyncio
from cortex import AsyncCortexClient

async def main():
    async with AsyncCortexClient(api_key="your_key") as client:
        # Concurrent searches
        tasks = [
            client.search("AI developments"),
            client.search("quantum computing"),
            client.search("renewable energy")
        ]
        
        results = await asyncio.gather(*tasks)
        
        for result in results:
            print(result.summary)

# Run async code
asyncio.run(main())

Context Managers

# Automatic resource cleanup
with CortexClient(api_key="your_key") as client:
    result = client.search("test query")
    # Client automatically closes connections

🧪 Testing

Mock Client

from cortex.testing import MockCortexClient

# Create mock client for testing
mock_client = MockCortexClient()

# Add mock responses
mock_client.add_search_response(
    query="test query",
    response={
        "summary": "Test summary",
        "sources": [],
        "confidence_score": 0.9
    }
)

# Use in tests
result = mock_client.search("test query")
assert result.summary == "Test summary"

Fixtures

import pytest
from cortex import CortexClient

@pytest.fixture
def cortex_client():
    return CortexClient(
        api_key="test_key",
        base_url="https://sandbox.api.usecortex.co/v1"
    )

def test_search(cortex_client):
    result = cortex_client.search("test query")
    assert result is not None

📚 Examples

RAG Integration

from cortex import CortexClient
import chromadb

# Initialize clients
cortex = CortexClient(api_key="your_key")
chroma = chromadb.Client()
collection = chroma.create_collection("knowledge_base")

def update_knowledge_base(query):
    # Search for current information
    result = cortex.search(query, max_results=10)
    
    # Extract content from sources
    documents = []
    for source in result.sources:
        content = cortex.extract(source.url)
        if content.success:
            documents.append({
                "text": content.text,
                "metadata": {
                    "url": source.url,
                    "title": source.title,
                    "confidence": source.confidence,
                    "timestamp": result.metadata.timestamp
                }
            })
    
    # Add to vector database
    for i, doc in enumerate(documents):
        collection.add(
            documents=[doc["text"]],
            metadatas=[doc["metadata"]],
            ids=[f"doc_{i}"]
        )
    
    return f"Added {len(documents)} documents to knowledge base"

# Usage
update_knowledge_base("machine learning best practices")

News Monitoring

import schedule
import time
from cortex import CortexClient

client = CortexClient(api_key="your_key")

def monitor_news(topic):
    try:
        result = client.search(
            query=f"latest news {topic}",
            recency="day",
            max_results=5
        )
        
        for source in result.sources:
            print(f"📰 {source.title}")
            print(f"   {source.url}")
            print(f"   Published: {source.published_date}")
        
        # Validate important claims
        if "breakthrough" in result.summary.lower():
            validation = client.validate(
                claim=result.summary.split('.')[0],  # First sentence
                context={"domain": "technology", "claim_type": "event"}
            )
            print(f"🔍 Validation: {validation.validation_result}")
            
    except Exception as e:
        print(f"Error monitoring {topic}: {e}")

# Schedule monitoring
schedule.every(30).minutes.do(monitor_news, "artificial intelligence")
schedule.every(30).minutes.do(monitor_news, "quantum computing")

# Run scheduler
while True:
    schedule.run_pending()
    time.sleep(1)

Content Analysis Pipeline

from cortex import CortexClient
from dataclasses import dataclass
from typing import List
import pandas as pd

@dataclass
class ContentAnalysis:
    url: str
    title: str
    word_count: int
    reading_time: str
    quality_score: float
    key_topics: List[str]
    sentiment: str
    fact_check_results: List[dict]

def analyze_content_pipeline(urls: List[str]) -> pd.DataFrame:
    client = CortexClient(api_key="your_key")
    results = []
    
    for url in urls:
        try:
            # Extract content
            content = client.extract(url, include_metadata=True)
            
            if not content.success:
                continue
            
            # Analyze key claims for fact-checking
            sentences = content.text.split('.')[:5]  # First 5 sentences
            fact_checks = []
            
            for sentence in sentences:
                if len(sentence.strip()) > 20:  # Skip short sentences
                    validation = client.validate(
                        claim=sentence.strip(),
                        context={"claim_type": "factual"}
                    )
                    fact_checks.append({
                        "claim": sentence.strip(),
                        "result": validation.validation_result,
                        "confidence": validation.confidence_score
                    })
            
            # Compile analysis
            analysis = ContentAnalysis(
                url=url,
                title=content.metadata.title,
                word_count=content.metadata.word_count,
                reading_time=content.metadata.reading_time,
                quality_score=content.quality_score,
                key_topics=content.keywords[:10],
                sentiment="neutral",  # Could add sentiment analysis
                fact_check_results=fact_checks
            )
            
            results.append(analysis)
            
        except Exception as e:
            print(f"Error analyzing {url}: {e}")
            continue
    
    # Convert to DataFrame for analysis
    df = pd.DataFrame([
        {
            "URL": r.url,
            "Title": r.title,
            "Word Count": r.word_count,
            "Quality Score": r.quality_score,
            "Verified Claims": len([fc for fc in r.fact_check_results if fc["result"] == "VERIFIED"]),
            "Total Claims": len(r.fact_check_results),
            "Key Topics": ", ".join(r.key_topics[:5])
        }
        for r in results
    ])
    
    return df

# Usage
urls = [
    "https://techcrunch.com/ai-article",
    "https://arxiv.org/paper",
    "https://nature.com/research"
]

analysis_df = analyze_content_pipeline(urls)
print(analysis_df)

📋 API Reference

CortexClient

class CortexClient:
    def __init__(
        self,
        api_key: str = None,
        base_url: str = "https://api.usecortex.co/v1",
        timeout: int = 30,
        max_retries: int = 3,
        retry_delay: float = 1.0,
        verify_ssl: bool = True,
        user_agent: str = None,
        default_headers: Dict[str, str] = None
    )
    
    def search(
        self,
        query: str,
        max_results: int = 5,
        language: str = "auto",
        country: str = "global",
        recency: str = "auto",
        format: str = "json",
        include_sources: bool = True,
        domain_filter: Dict = None,
        context: Dict = None,
        options: Dict = None
    ) -> SearchResult
    
    def extract(
        self,
        url: str,
        format: str = "text",
        include_metadata: bool = True,
        extract_images: bool = False,
        extract_links: bool = False,
        options: Dict = None
    ) -> ExtractionResult
    
    def validate(
        self,
        claim: str,
        sources: List[str] = None,
        context: Dict = None,
        options: Dict = None
    ) -> ValidationResult

---

Next: JavaScript SDK → - Official JavaScript/Node.js library