Python SDK user guide

The Crusoe AI Platform Python SDK provides a clean, typed interface to manage and interact with all platform services.

Installation

Install the Python SDK from the local package path or repository:

# Install from local SDK directory
pip install ./sdk/python

# Or install from source repository
pip install git+https://github.com/crusoe/cai.git#subdirectory=sdk/python

Requirements

Python 3.8 or higher
pydantic >= 2.0
urllib3 >= 1.25.3

Authentication & Configuration

All client operations accept a Configuration instance where you specify the API base host and bearer access token:

import os
import crusoe_ai_secrets

# Configure client credentials and base URL
configuration = crusoe_ai_secrets.Configuration(
    host=os.getenv("CAI_API", "https://api.codyhill.dev"),
    access_token=os.getenv("CAI_API_KEY", "cai_pk_live_1234567890abcdef")
)

Code examples by service

1. Agents

Interact with deployed agents, stream responses, manage session state, and commit sessions to long-term memory.

import os
import json
import urllib3

# Setup base request configuration
api_base = os.getenv("CAI_API", "https://api.codyhill.dev")
api_key = os.getenv("CAI_API_KEY", "cai_pk_live_1234567890abcdef")
project_id = os.getenv("CAI_PROJECT", "0191f2c4-7777-7c3d-8e4f-5a6b7c8d9e0f")
headers = {
    "Authorization": f"Bearer {api_key}",
    "Content-Type": "application/json"
}

http = urllib3.PoolManager()

# 1. Invoke an agent synchronously
def invoke_agent(agent_name: str, message: str, session_id: str = "sess_001"):
    url = f"{api_base}/v1/agents/{agent_name}/invoke?project={project_id}"
    payload = json.dumps({
        "prompt": message,
        "session_id": session_id,
        "user_id": "user_42"
    })
    
    response = http.request("POST", url, headers=headers, body=payload)
    data = json.loads(response.data.decode("utf-8"))
    print("Agent output:", data.get("response"))
    return data

# 2. Invoke an agent with streaming output (NDJSON)
def invoke_agent_stream(agent_name: str, message: str, session_id: str = "sess_001"):
    url = f"{api_base}/v1/agents/{agent_name}/invoke/stream?project={project_id}"
    payload = json.dumps({
        "prompt": message,
        "session_id": session_id
    })
    
    response = http.request("POST", url, headers=headers, body=payload, preload_content=False)
    for line in response.stream():
        if line:
            chunk = json.loads(line.decode("utf-8"))
            print(chunk.get("text", ""), end="", flush=True)
    print()

# 3. Commit a session transcript to agent memory bank
def memorize_session(agent_name: str, session_id: str):
    url = f"{api_base}/v1/agents/{agent_name}/sessions/{session_id}/memorize?project={project_id}"
    response = http.request("POST", url, headers=headers)
    print("Memorize status:", response.status)

# Example usage
invoke_agent("research-assistant", "Summarize Q3 cloud market trends.")
memorize_session("research-assistant", "sess_001")

2. VectorDB

Create vector indexes, upsert embeddings, perform similarity queries, and scroll through stored points.

import os
import json
import urllib3

vectordb_base = os.getenv("CAI_VECTORDB_API", "https://api.codyhill.dev")
api_key = os.getenv("CAI_API_KEY", "cai_pk_live_1234567890abcdef")
project_id = os.getenv("CAI_PROJECT", "0191f2c4-7777-7c3d-8e4f-5a6b7c8d9e0f")

headers = {
    "Authorization": f"Bearer {api_key}",
    "Content-Type": "application/json"
}
http = urllib3.PoolManager()

# 1. Create a VectorDB index
def create_index(index_name: str, dimension: int = 1536):
    url = f"{vectordb_base}/v1/projects/{project_id}/indexes"
    payload = json.dumps({
        "name": index_name,
        "dimension": dimension,
        "metric": "cosine"
    })
    resp = http.request("POST", url, headers=headers, body=payload)
    print("Create index response:", resp.status)

# 2. Upsert vector points
def upsert_points(index_name: str, points: list):
    url = f"{vectordb_base}/v1/projects/{project_id}/indexes/{index_name}:upsert"
    payload = json.dumps({"points": points})
    resp = http.request("POST", url, headers=headers, body=payload)
    print("Upsert response:", json.loads(resp.data.decode("utf-8")))

# 3. Perform similarity query
def query_vector(index_name: str, vector: list, top_k: int = 5):
    url = f"{vectordb_base}/v1/projects/{project_id}/indexes/{index_name}:query"
    payload = json.dumps({
        "vector": vector,
        "top_k": top_k,
        "include_payload": True
    })
    resp = http.request("POST", url, headers=headers, body=payload)
    results = json.loads(resp.data.decode("utf-8"))
    print("Matches:", len(results.get("matches", [])))
    return results

# Example usage
sample_points = [
    {
        "id": "doc_001",
        "vector": [0.012] * 1536,
        "payload": {"category": "cloud", "title": "Serverless Workload autoscaling"}
    }
]
upsert_points("kb-vectors", sample_points)
query_vector("kb-vectors", [0.012] * 1536)

3. Functions

Deploy serverless functions, execute code payloads, inspect status, and stream function logs.

import os
import json
import urllib3

api_base = os.getenv("CAI_API", "https://api.codyhill.dev")
api_key = os.getenv("CAI_API_KEY", "cai_pk_live_1234567890abcdef")
project_id = os.getenv("CAI_PROJECT", "0191f2c4-7777-7c3d-8e4f-5a6b7c8d9e0f")

headers = {
    "Authorization": f"Bearer {api_key}",
    "Content-Type": "application/json"
}
http = urllib3.PoolManager()

# 1. Execute a serverless function
def execute_function(function_name: str, input_data: dict):
    url = f"{api_base}/v1/agents/{function_name}/invoke?project={project_id}"
    payload = json.dumps(input_data)
    resp = http.request("POST", url, headers=headers, body=payload)
    result = json.loads(resp.data.decode("utf-8"))
    print("Function result:", result)
    return result

# 2. Get function execution logs
def get_function_logs(function_name: str):
    url = f"{api_base}/v1/agents/{function_name}/logs?project={project_id}"
    resp = http.request("GET", url, headers={"Authorization": f"Bearer {api_key}"})
    print("Logs:")
    print(resp.data.decode("utf-8"))

# Example execution
execute_function("pdf-parser", {"document_url": "https://example.com/doc.pdf"})
get_function_logs("pdf-parser")

4. Secrets

Manage encrypted secrets, bind values to workloads, apply mappings, and reveal stored values with audit logs.

import os
import crusoe_ai_secrets
from crusoe_ai_secrets.api import secrets_api, bindings_api
from crusoe_ai_secrets.models import (
    CreateSecretRequest,
    PutBindingRequest
)

# 1. Initialize Secrets API Client
configuration = crusoe_ai_secrets.Configuration(
    host=os.getenv("CAI_API", "https://api.codyhill.dev"),
    access_token=os.getenv("CAI_API_KEY", "cai_pk_live_1234567890abcdef")
)

project_id = "0191f2c4-7777-7c3d-8e4f-5a6b7c8d9e0f"

with crusoe_ai_secrets.ApiClient(configuration) as api_client:
    sec_api = secrets_api.SecretsApi(api_client)
    bind_api = bindings_api.BindingsApi(api_client)

    # 2. Store or update a secret
    create_req = CreateSecretRequest(
        name="openai-api-key",
        value="sk-proj-1234567890abcdef"
    )
    sec_response = sec_api.create_project_secret(project_id, create_req)
    print(f"Stored secret '{sec_response.name}' version {sec_response.version}")

    # 3. Bind stored secret to an agent environment variable
    bind_req = PutBindingRequest(
        secret_name="openai-api-key"
    )
    bind_api.put_agent_secret_binding(project_id, "research-assistant", "OPENAI_API_KEY", bind_req)
    print("Bound secret to OPENAI_API_KEY on research-assistant")

    # 4. Apply binding rules to deploy secret to workload
    apply_resp = bind_api.apply_agent_secret_bindings(project_id, "research-assistant")
    print("Applied secret bindings:", apply_resp)

    # 5. Reveal secret value (Project Admin only, audited)
    revealed = sec_api.reveal_secret(project_id, "openai-api-key")
    print("Revealed value:", revealed.value)

5. MemoryStore

Manage Redis-compatible key-value instances, inspect live metrics, and rotate access credentials.

import os
import json
import urllib3

mem_base = os.getenv("CAI_MEMORYSTORE_API", "https://api.codyhill.dev")
api_key = os.getenv("CAI_API_KEY", "cai_pk_live_1234567890abcdef")
project_id = os.getenv("CAI_PROJECT", "0191f2c4-7777-7c3d-8e4f-5a6b7c8d9e0f")

headers = {
    "Authorization": f"Bearer {api_key}",
    "Content-Type": "application/json"
}
http = urllib3.PoolManager()

# 1. Create a MemoryStore instance
def create_memorystore(name: str):
    url = f"{mem_base}/v1/projects/{project_id}/memorystores"
    payload = json.dumps({"name": name, "max_memory_mb": 512})
    resp = http.request("POST", url, headers=headers, body=payload)
    data = json.loads(resp.data.decode("utf-8"))
    print("Created MemoryStore endpoint:", data.get("endpoint"))
    return data

# 2. Get live instance statistics
def get_stats(name: str):
    url = f"{mem_base}/v1/projects/{project_id}/memorystores/{name}/stats"
    resp = http.request("GET", url, headers={"Authorization": f"Bearer {api_key}"})
    stats = json.loads(resp.data.decode("utf-8"))
    print("Connected clients:", stats.get("connected_clients"))
    print("Used memory:", stats.get("used_memory_human"))

# 3. Rotate access credentials (Admin only)
def rotate_credentials(name: str):
    url = f"{mem_base}/v1/projects/{project_id}/memorystores/{name}/rotate-credential"
    resp = http.request("POST", url, headers=headers)
    new_creds = json.loads(resp.data.decode("utf-8"))
    print("New password generated successfully.")

create_memorystore("agent-cache")
get_stats("agent-cache")

6. PubSub

Create topics, publish JSON messages, pull messages from subscriptions, and send acknowledgments.

import os
import json
import base64
import urllib3

pubsub_base = os.getenv("CAI_PUBSUB_API", "https://api.codyhill.dev")
api_key = os.getenv("CAI_API_KEY", "cai_pk_live_1234567890abcdef")
project_id = os.getenv("CAI_PROJECT", "0191f2c4-7777-7c3d-8e4f-5a6b7c8d9e0f")

headers = {
    "Authorization": f"Bearer {api_key}",
    "Content-Type": "application/json"
}
http = urllib3.PoolManager()

# 1. Publish messages to a topic
def publish_message(topic: str, message_dict: dict):
    url = f"{pubsub_base}/v1/projects/{project_id}/topics/{topic}:publish"
    encoded_data = base64.b64encode(json.dumps(message_dict).encode("utf-8")).decode("utf-8")
    
    payload = json.dumps({
        "messages": [
            {
                "data": encoded_data,
                "attributes": {"event_type": "user_signup"}
            }
        ]
    })
    resp = http.request("POST", url, headers=headers, body=payload)
    print("Published message IDs:", json.loads(resp.data.decode("utf-8")).get("message_ids"))

# 2. Pull messages from a subscription
def pull_messages(topic: str, subscription: str, max_messages: int = 10):
    url = f"{pubsub_base}/v1/projects/{project_id}/topics/{topic}/subscriptions/{subscription}:pull"
    payload = json.dumps({"max_messages": max_messages})
    resp = http.request("POST", url, headers=headers, body=payload)
    messages = json.loads(resp.data.decode("utf-8")).get("received_messages", [])
    
    ack_ids = []
    for msg in messages:
        raw_bytes = base64.b64decode(msg["message"]["data"])
        content = json.loads(raw_bytes.decode("utf-8"))
        print("Received message:", content)
        ack_ids.append(msg["ack_id"])
        
    return ack_ids

# 3. Acknowledge processed messages
def ack_messages(topic: str, subscription: str, ack_ids: list):
    if not ack_ids:
        return
    url = f"{pubsub_base}/v1/projects/{project_id}/topics/{topic}/subscriptions/{subscription}:acknowledge"
    payload = json.dumps({"ack_ids": ack_ids})
    resp = http.request("POST", url, headers=headers, body=payload)
    print("Ack response status:", resp.status)

# Example flow
publish_message("user-events", {"user_id": "usr_99", "email": "dev@example.com"})
received_ack_ids = pull_messages("user-events", "user-events-worker", max_messages=5)
ack_messages("user-events", "user-events-worker", received_ack_ids)

Error handling

Catch ApiException from package imports to handle platform error responses cleanly:

import crusoe_ai_secrets
from crusoe_ai_secrets.rest import ApiException

try:
    # Perform API operation
    pass
except ApiException as e:
    print(f"HTTP Status: {e.status}")
    print(f"Error Body: {e.body}")
    print(f"Request ID: {e.headers.get('x-request-id')}")

Installation​

Requirements​

Authentication & Configuration​

Code examples by service​

1. Agents​

2. VectorDB​

3. Functions​

4. Secrets​

5. MemoryStore​

6. PubSub​

Error handling​