cognitora 1.6.1

Official Python SDK for Cognitora - Operating System for Autonomous AI Agents

Cognitora Python SDK

The official Python SDK for Cognitora - Operating System for Autonomous AI Agents.

Features

• Code Interpreter: Execute Python, JavaScript, and Bash code in secure sandboxed environments
• Containers Platform: Run containerized workloads with flexible resource allocation
• Session Management: Persistent sessions with state management and automatic cleanup
• Execution Control: Start, monitor, and cancel long-running compute tasks
• File Operations: Upload and manipulate files in execution environments
• Networking Control: Optional internet access with security-focused defaults
• Async Support: Full async/await support for high-performance applications
• Type Safety: Comprehensive type hints and data validation

Installation

pip install cognitora

Quick Start

from cognitora import Cognitora

# Initialize the client
client = Cognitora(api_key="your_api_key_here")

# Execute Python code with networking
result = client.code_interpreter.execute(
    code="print('Hello from Cognitora!')",
    language="python",
    networking=True  # Enable internet access (default for code interpreter)
)

print(f"Status: {result.data.status}")
for output in result.data.outputs:
    print(f"{output.type}: {output.data}")

Authentication

Get your API key from the Cognitora Dashboard and set it:

# Method 1: Pass directly
client = Cognitora(api_key="cgk_1234567890abcdef")

# Method 2: Environment variable
import os
os.environ['COGNITORA_API_KEY'] = 'cgk_1234567890abcdef'
client = Cognitora()  # Will use environment variable

# Method 3: With custom configuration
client = Cognitora(
    api_key="your_api_key",
    base_url="https://api.cognitora.dev",  # Production default
    timeout=30
)

Code Interpreter

Basic Execution with Networking Control

# Execute Python code with internet access (default)
result = client.code_interpreter.execute(
    code="""
import requests
import numpy as np
import matplotlib.pyplot as plt

# Fetch data from API (requires networking)
response = requests.get('https://api.github.com/repos/microsoft/typescript')
repo_data = response.json()

print(f"Repository: {repo_data['name']}")
print(f"Stars: {repo_data['stargazers_count']}")

# Create visualization
x = np.linspace(0, 10, 100)
y = np.sin(x)

plt.figure(figsize=(10, 6))
plt.plot(x, y)
plt.title('Sine Wave')
plt.show()
""",
    language="python",
    networking=True  # Explicitly enable networking (default for code interpreter)
)

# Execute code without internet access for security
secure_result = client.code_interpreter.execute(
    code="""
import numpy as np
# No external requests - isolated execution
data = np.random.randn(1000)
print(f"Mean: {np.mean(data)}")
""",
    language="python",
    networking=False  # Disable networking for secure execution
)

Session Persistence

Sessions maintain state between executions, making them perfect for:

• Interactive data analysis workflows
• Long-running machine learning experiments
• Multi-step data processing pipelines
• Collaborative coding environments

# Create a persistent session (defaults to 1 day if no timeout specified)
session = client.code_interpreter.create_session(
    language="python",
    timeout_seconds=3600,  # 1 hour (optional)
    resources={
        "cpu_cores": 2,
        "memory_mb": 2048,
        "storage_gb": 10
    }
)

print(f"Session created: {session.session_id}")

# Execute code in session (variables persist)
result1 = client.code_interpreter.execute(
    code="x = 42; y = 'Hello World'; import pandas as pd",
    session_id=session.session_id,
    networking=True  # Enable networking for package installs
)

result2 = client.code_interpreter.execute(
    code="print(f'x = {x}, y = {y}'); print(f'Pandas version: {pd.__version__}')",
    session_id=session.session_id
)

# Variables and imports are maintained across executions
print(result2.data.outputs[0].data)  # Output: x = 42, y = Hello World

# Get session execution history
session_executions = client.code_interpreter.get_session_executions(session.session_id)
print(f"Session has {len(session_executions)} executions")

# Always clean up sessions when done
client.code_interpreter.delete_session(session.session_id)

New Execution Management Features

# List all interpreter executions across all sessions
all_executions = client.code_interpreter.list_all_executions(
    limit=20,
    status='completed'
)

print(f"Found {len(all_executions)} completed executions")

# Get specific execution details
execution_details = client.code_interpreter.get_execution('exec_123456')
print(f"Execution status: {execution_details['status']}")

# Get executions for a specific session
session_executions = client.code_interpreter.get_session_executions(
    'session_123456',
    limit=10
)

File Operations

from cognitora import FileUpload

# Prepare files
files = [
    FileUpload(
        name="data.csv",
        content="name,age,city\nJohn,30,NYC\nJane,25,LA",
        encoding="string"
    ),
    FileUpload(
        name="script.py",
        content="import pandas as pd\ndf = pd.read_csv('data.csv')\nprint(df.head())",
        encoding="string"
    )
]

# Execute with files
result = client.code_interpreter.run_with_files(
    code="exec(open('script.py').read())",
    files=files,
    language="python"
)

Containers Platform

The Containers Platform allows you to run containerized workloads with full execution control and networking security.

Basic Container Execution

# Run a secure container (isolated by default)
execution = client.containers.create_container(
    image="docker.io/library/python:3.11-slim",
    command=["python", "-c", "print('Hello from secure container!')"],
    cpu_cores=1.0,
    memory_mb=512,
    max_cost_credits=5,
    networking=False  # Default: isolated for security
)

print(f"Container ID: {execution.id}")
print(f"Status: {execution.status}")

# Run container with internet access when needed
networking_execution = client.containers.create_container(
    image="docker.io/library/python:3.11",
    command=["python", "-c", """
import requests
response = requests.get('https://api.github.com/users/octocat')
user_data = response.json()
print(f"GitHub user: {user_data['name']}")
"""],
    cpu_cores=1.0,
    memory_mb=512,
    max_cost_credits=10,
    networking=True  # Enable networking for API calls
)

Advanced Container Management

# List all container executions
container_executions = client.containers.list_all_container_executions(
    limit=50,
    status='running'
)

print(f"Active containers: {len(container_executions)}")

# Get specific container execution details
container_execution = client.containers.get_container_execution('exec_123456')
print(f"Container execution: {container_execution['status']}")

# Get executions for a specific container
container_history = client.containers.get_container_executions('container_123456')
print(f"Container has {len(container_history)} executions")

Execution Control & Cancellation

# Run a long-running task with networking control
execution = client.containers.create_container(
    image="docker.io/library/python:3.11-slim",
    command=["python", "-c", """
import time
import requests

for i in range(100):
    print(f'Processing step {i+1}/100')
    
    # Make API call every 10 steps (requires networking)
    if i % 10 == 0:
        try:
            response = requests.get('https://httpbin.org/delay/1')
            print(f'API call {i//10 + 1} completed')
        except Exception as e:
            print(f'Network error: {e}')
    
    time.sleep(2)
print('Processing complete!')
"""],
    cpu_cores=2.0,
    memory_mb=1024,
    max_cost_credits=50,
    timeout_seconds=3600,
    networking=True  # Enable networking for API calls
)

print(f"Started container: {execution.id}")

# Monitor execution status
try:
    # Wait for completion with timeout  
    completed = client.containers.wait_for_completion(
        execution.id, 
        timeout_seconds=30,  # 30 seconds timeout for demo
        poll_interval=2
    )
    print(f"Container completed: {completed.status}")
    
except Exception as e:
    print(f"Container taking too long, cancelling...")
    
    # Cancel the container
    result = client.containers.cancel_container(execution.id)
    print(f"Cancellation result: {result}")
    
    # Verify cancellation
    cancelled_container = client.containers.get_container(execution.id)
    print(f"Final status: {cancelled_container.status}")

Resource Management Best Practices

# Always estimate costs before running expensive operations
estimate = client.containers.estimate_cost(
    cpu_cores=4.0,
    memory_mb=8192,
    storage_gb=20,
    gpu_count=1,
    timeout_seconds=3600
)

print(f"Estimated cost: {estimate['estimated_credits']} credits")

if estimate['estimated_credits'] <= 100:
    # Proceed with execution
    execution = client.containers.create_container(
        image="docker.io/tensorflow/tensorflow:latest-gpu",
        command=["python", "train.py"],
        cpu_cores=4.0,
        memory_mb=8192,
        storage_gb=20,
        gpu_count=1,
        max_cost_credits=int(estimate['estimated_credits'] * 1.2),  # 20% buffer
        networking=False  # Secure by default
    )
    
    try:
        # Monitor execution
        result = client.containers.wait_for_completion(execution.id)
        logs = client.containers.get_container_logs(execution.id)
        print(f"Training completed: {result.status}")
        
    except KeyboardInterrupt:
        # Handle user interruption gracefully
        print("Interruption detected, cancelling container...")
        client.containers.cancel_container(execution.id)
        
    except Exception as e:
        # Handle errors and cleanup
        print(f"Error occurred: {e}")
        client.containers.cancel_container(execution.id)
        
else:
    print(f"Execution too expensive ({estimate['estimated_credits']} credits), skipping...")

Async Support

import asyncio
from cognitora import CognitoraAsync

async def main():
    async with CognitoraAsync(api_key="your_api_key") as client:
        # Parallel execution
        tasks = [
            client.code_interpreter.execute(
                code=f"import time; time.sleep(1); print('Task {i} completed')",
                language="python"
            )
            for i in range(5)
        ]
        
        results = await asyncio.gather(*tasks)
        
        for i, result in enumerate(results):
            print(f"Task {i}: {result.data.outputs[0].data}")

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

Error Handling

from cognitora import CognitoraError, AuthenticationError, RateLimitError

try:
    result = client.code_interpreter.execute(
        code="raise ValueError('Test error')",
        language="python"
    )
except AuthenticationError:
    print("Invalid API key")
except RateLimitError:
    print("Rate limit exceeded, please wait")
except CognitoraError as e:
    print(f"API error: {e}")
    print(f"Status code: {e.status_code}")
    print(f"Response data: {e.response_data}")

API Reference

CodeInterpreter Class

Methods

• execute(code, language='python', session_id=None, files=None, timeout_seconds=60, environment=None, networking=None) - Execute code with networking control
• create_session(language='python', timeout_seconds=None, expires_at=None, environment=None, resources=None) - Create persistent session
• list_sessions() - List active sessions
• get_session(session_id) - Get session details
• delete_session(session_id) - Delete session
• get_session_logs(session_id, limit=50, offset=0) - Get session logs
• list_all_executions(limit=50, offset=0, status=None) - NEW: List all interpreter executions
• get_execution(execution_id) - NEW: Get specific execution details
• get_session_executions(session_id, limit=50, offset=0) - NEW: List executions for specific session
• run_python(code, session_id=None) - Execute Python code
• run_javascript(code, session_id=None) - Execute JavaScript code
• run_bash(command, session_id=None) - Execute bash command
• run_with_files(code, files, language='python', session_id=None) - Execute with files

Containers Class

Methods

• create_container(image, command, cpu_cores, memory_mb, max_cost_credits, networking=None, **kwargs) - Create container with networking control
• list_containers(limit=50, offset=0, status=None) - List containers
• get_container(container_id) - Get container details
• cancel_container(container_id) - Cancel container
• get_container_logs(container_id) - Get container logs
• get_container_executions(container_id) - Get container executions
• list_all_container_executions(limit=50, offset=0, status=None) - NEW: List all container executions
• get_container_execution(execution_id) - NEW: Get specific container execution details
• estimate_cost(cpu_cores, memory_mb, storage_gb=5, gpu_count=0, timeout_seconds=300) - Estimate cost
• wait_for_completion(container_id, timeout_seconds=300, poll_interval=5) - Wait for completion
• run_and_wait(image, command, cpu_cores, memory_mb, max_cost_credits, **kwargs) - Create and wait

Security & Networking

Default Networking Behavior

Service	Default Networking	Security Rationale
Code Interpreter	True (enabled)	Needs package installs, data fetching
Containers	False (disabled)	Security-first: isolated by default

Networking Best Practices

# For data analysis that needs external data
data_analysis = client.code_interpreter.execute(
    code="""
import pandas as pd
import requests

# Fetch external data
response = requests.get('https://api.coindesk.com/v1/bpi/currentprice.json')
data = response.json()
print(f"Bitcoin price: {data['bpi']['USD']['rate']}")
""",
    networking=True  # Required for external API calls
)

# For secure computation without external access
secure_computation = client.containers.create_container(
    image="docker.io/library/python:3.11",
    command=["python", "-c", "print('Secure isolated computation')"],
    cpu_cores=1.0,
    memory_mb=512,
    max_cost_credits=5,
    networking=False  # Isolated execution (default)
)

# For containers that need external resources
data_processing = client.containers.create_container(
    image="docker.io/library/python:3.11",
    command=["pip", "install", "requests", "&&", "python", "process.py"],
    cpu_cores=2.0,
    memory_mb=1024,
    max_cost_credits=20,
    networking=True  # Enable for pip install and external APIs
)

Configuration

Environment Variables

export COGNITORA_API_KEY="your_api_key_here"
export COGNITORA_BASE_URL="https://api.cognitora.dev"  # Optional
export COGNITORA_TIMEOUT="30"  # Optional, seconds

Best Practices

1. Resource Management

# Always specify appropriate resources
session = client.code_interpreter.create_session(
    language="python",
    timeout_seconds=1800,  # 30 minutes - don't set too high
    resources={
        "cpu_cores": 1.0,    # Start small
        "memory_mb": 1024,   # Adjust based on needs
        "storage_gb": 5      # Minimum required
    }
)

2. Session Lifecycle

# Create session
session = client.code_interpreter.create_session()

try:
    # Use session for multiple operations
    for code_snippet in code_snippets:
        result = client.code_interpreter.execute(
            code=code_snippet,
            session_id=session.session_id
        )
        process_result(result)
finally:
    # Clean up
    client.code_interpreter.delete_session(session.session_id)

3. Error Recovery

import time

def execute_with_retry(client, code, max_retries=3):
    for attempt in range(max_retries):
        try:
            return client.code_interpreter.execute(code=code)
        except RateLimitError:
            if attempt < max_retries - 1:
                time.sleep(2 ** attempt)  # Exponential backoff
                continue
            raise
        except CognitoraError as e:
            if e.status_code >= 500 and attempt < max_retries - 1:
                time.sleep(1)
                continue
            raise

Recent Updates

API Refactor Alignment

• ✅ Updated endpoint paths: Code interpreter now uses /api/v1/interpreter/*
• ✅ Container-focused architecture: All compute operations use /api/v1/compute/containers/*
• ✅ Networking parameter: Security-focused networking control for all operations
• ✅ New execution endpoints: Comprehensive execution management and history
• ✅ Production-ready defaults: All clients default to https://api.cognitora.dev

New Features

• ✅ Networking Control: Optional networking parameter with security-focused defaults
• ✅ Execution Management: List, filter, and retrieve execution details across all services
• ✅ Session History: Track and manage executions within persistent sessions
• ✅ Container Execution History: Detailed tracking of container execution lifecycle

Breaking Changes from Previous Versions

• Method Names: compute.* methods renamed to containers.*
• Endpoint Paths: Code interpreter paths changed from /code-interpreter/ to /interpreter/
• Default Networking: Containers now default to networking=False for security

Data Classes and Types

from cognitora import (
    ExecuteCodeRequest,
    ExecuteCodeResponse, 
    ComputeExecutionRequest,
    FileUpload,
    Session,
    Execution
)

# All request and response types are provided as dataclasses
request = ExecuteCodeRequest(
    code="print('Hello')",
    language="python",
    networking=True  # NEW: networking control
)

container_request = ComputeExecutionRequest(
    image="python:3.11",
    command=["python", "-c", "print('test')"],
    cpu_cores=1.0,
    memory_mb=512,
    max_cost_credits=5,
    networking=False  # NEW: networking control
)

Support

• Documentation: docs.cognitora.dev
• Support or get an early access: [e-mail]

License

MIT License - see LICENSE file for details.