MicroDC Python Client Library

Full documentation: microdc.gitlab.io/python-client

Overview

The MicroDC Python Client Library provides a simple, intuitive interface for developers to submit inference jobs to the MicroDC.ai distributed inference platform. This document serves as the specification for implementing the client library in a separate repository.

Repository Information

Repository: microdc/python-client (separate from server) Package Name: microdc-client Import Name: microdc (follows PEP 8 - all lowercase) Python Version: 3.8+ Status: ✅ Core implementation complete (v1.0.13) | 🧪 278 tests, 86% coverage | 🔧 Pre-commit hooks + GitLab CI

Quick Start

from microdc import Client, LLMComplete

# Initialize client
client = Client(api_key="mDC_499FC19C-686A-47C5-AA93-E619C55EBE98")

# Create and configure job (single-turn generation)
job = LLMComplete(model="llama3.3")
job.set_prompt("Why is the sky blue?")

# Submit job
job_id = client.send_job(job)

# Wait for completion
client.wait_for_all()

# Get results
result = client.get_job_details(job_id)
print(result.result)

Installation

# Install from GitLab (v1 branch)
pip install git+https://gitlab.com/microdc/python-client.git

# Or install from source
git clone https://gitlab.com/microdc/python-client.git
cd python-client
pip install -e .

Core Components

1. Client Class

The main interface for all API interactions.

class Client:
    def __init__(
        self,
        api_key: str,
        base_url: str = "https://api.microdc.ai",
        timeout: int = 30,
        verify_ssl: bool = True,
        auto_start_polling: bool = True,
        encryption_key_path: Optional[str] = None
    ):
        """Initialize MicroDC client."""
        pass

Methods:

• send_job(job: BaseCall, encrypt: bool = False) -> str - Submit a job and get job ID (set encrypt=True for end-to-end encryption)
• get_job_details(job_id: str) -> JobDetails - Get complete job information
• get_job_status(job_id: str) -> str - Get current job status
• cancel_job(job_id: str) -> bool - Cancel a pending/processing job
• acknowledge_job(job_id: str) -> Dict[str, Any] - Acknowledge job completion (returns message, job_id, acknowledged_at)
• list_jobs(status=None, acknowledged=None, limit=100, offset=0) -> List[JobDetails] - List user's jobs (filter by status and/or acknowledgment state)
• set_callback(callback: Callable) -> None - Set completion callback
• wait_for_all(timeout: Optional[float]) -> None - Block until all jobs complete
• wait_for_job(job_id: str, timeout: Optional[float]) -> JobDetails - Block until specific job completes
• upload_file(file_path: str, description: Optional[str]) -> Dict[str, Any] - Upload a file
• delete_file(file_id: str, permanent: bool) -> Dict[str, Any] - Delete an uploaded file
• create_download_token(file_id: str, expires_in_minutes: int, max_uses: int) -> Dict[str, Any] - Create download token
• upload_and_tokenize(file_path: str, ...) -> Dict[str, Any] - Upload file and create token in one step
• close() -> None - Stop the polling thread and close the HTTP client (also called on context-manager exit)

2. Job Classes

LLM Jobs - Language Model Inference

LLMComplete - For single-turn text generation (prompt-only, no system context):

from microdc import LLMComplete

job = LLMComplete(
    model="llama3.3",
    temperature=0.7,
    max_tokens=500
)
job.set_prompt("Write a haiku about Python")

Configuration (LLMComplete):

• model: str - Model name (required)
• prompt: str - Generation prompt (set via set_prompt())
• file_tokens: List[str] - File IDs from uploaded files (for vision/multimodal models)
• input_modalities: List[str] - Input types: ["text"], ["text", "image"], ["audio"], etc. (default: ["text"])
• output_modalities: List[str] - Output types: ["text"], ["image"], ["text", "image"], etc. (default: ["text"])
• temperature: float - Sampling temperature (0.0-2.0, default: 0.7)
• max_tokens: Optional[int] - Maximum tokens to generate
• top_p: float - Nucleus sampling (default: 1.0)
• top_k: Optional[int] - Top-k sampling
• frequency_penalty: float - Frequency penalty (default: 0.0)
• presence_penalty: float - Presence penalty (default: 0.0)
• stop: Optional[List[str]] - Stop sequences
• stream: bool - Enable streaming (default: False)

Helper Methods (LLMComplete):

• set_prompt(prompt: str) - Set the generation prompt
• add_file(file_token: str) - Add a file ID from client.upload_file()
• add_files(file_tokens: List[str]) - Add multiple file IDs

LLMChat - For conversational and multi-turn interactions (system/user/assistant messages):

from microdc import LLMChat

chat = LLMChat(
    model="gpt-4",
    temperature=0.7,
    max_tokens=500
)
chat.set_system("You are a helpful assistant.")
chat.add_user_message("Hello!")

Configuration (LLMChat):

• model: str - Model name (required)
• system: str - System message for conversation context (set via set_system())
• messages: List[Dict[str, str]] - Conversation messages (user/assistant turns)
• input_modalities: List[str] - Input types: ["text"], ["text", "image"], ["audio"], etc. (default: ["text"])
• output_modalities: List[str] - Output types: ["text"], ["image"], ["text", "image"], etc. (default: ["text"])
• temperature: float - Sampling temperature (0.0-2.0, default: 0.7)
• max_tokens: Optional[int] - Maximum tokens to generate
• top_p: float - Nucleus sampling (default: 1.0)
• top_k: Optional[int] - Top-k sampling
• frequency_penalty: float - Frequency penalty (default: 0.0)
• presence_penalty: float - Presence penalty (default: 0.0)
• stop: Optional[List[str]] - Stop sequences
• stream: bool - Enable streaming (default: False)

Helper Methods (LLMChat):

• set_system(content: str) - Set system message
• add_user_message(content: str) - Add user message
• add_assistant_message(content: str) - Add assistant message

LLMEmbed - Embedding Generation

from microdc import LLMEmbed

job = LLMEmbed(model="text-embedding-ada-002")
job.add_texts(["Hello world", "Goodbye world"])

Configuration:

• model: str - Model name (required)
• input_texts: List[str] - Texts to embed
• dimensions: Optional[int] - Embedding dimensions
• normalize: bool - Normalize embeddings (default: True)
• encoding_format: str - Output format: "float" or "base64"

Helper Methods:

• add_text(text: str) - Add single text
• add_texts(texts: List[str]) - Add multiple texts

DocumentCall - Document Processing

IMPORTANT: Document processing requires files to be uploaded FIRST before creating the job.

from microdc import DocumentCall

# Step 1: Upload file
upload_result = client.upload_file("document.pdf")
file_token = upload_result['id']

# Step 2: Create job with file token
job = DocumentCall(model="docling")
job.add_file(file_token)

# Step 3: Submit job
job_id = client.send_job(job)

Configuration:

• model: str - Model name (required, e.g., "docling")
• file_tokens: List[str] - File tokens from uploaded files
• max_tokens: Optional[int] - Maximum tokens to generate
• temperature: float - Sampling temperature (default: 0.7)

Helper Methods:

• add_file(file_token: str) - Add single file token
• add_files(file_tokens: List[str]) - Add multiple file tokens

Workflow:

1. Upload file: upload_result = client.upload_file("document.pdf")
2. Get token: file_token = upload_result['id']
3. Create job: job = DocumentCall(model="docling")
4. Add file: job.add_file(file_token)
5. Submit: job_id = client.send_job(job)

ContainerJob - Docker Container Execution

Run Docker containers on the MicroDC distributed platform with optional GPU access, file attachments, and resource limits.

from microdc import ContainerJob

job = ContainerJob(image="python:3.11")
job.set_command("python /input/script.py")
job.add_env_var("API_KEY", "sk-...")
job.set_memory_limit("4g")
job.set_cpu_count(2)

job_id = client.send_job(job)

Configuration:

• image: str - Docker image name and tag (required, e.g., "python:3.11")
• command: Optional[str | List[str]] - Command to run inside the container
• entrypoint: Optional[str] - Override the image entrypoint ("" means "no entrypoint")
• environment: Dict[str, str] - Environment variables
• gpu: bool - Enable GPU access (default: False)
• network_mode: str - Network mode (default: "none")
• mem_limit: Optional[str] - Memory limit (e.g., "512m", "4g")
• cpu_count: Optional[int] - Number of CPUs
• file_tokens: List[str] - File IDs from uploaded files (sent as top-level file_ids)
• min_capabilities: Dict[str, Any] - Minimum worker capabilities used as scheduling hints

Helper Methods:

• set_image(image: str) - Set Docker image
• set_command(command: str | List[str]) - Set command to execute
• set_entrypoint(entrypoint: str) - Override the image entrypoint
• set_environment(env: Dict[str, str]) - Replace environment variables
• add_env_var(key: str, value: str) - Add single environment variable
• enable_gpu() - Enable GPU access
• set_memory_limit(limit: str) - Set memory limit
• set_cpu_count(count: int) - Set CPU count
• add_file(file_token: str) - Add file token from upload
• add_files(file_tokens: List[str]) - Add multiple file tokens
• set_min_capabilities(capabilities: Dict[str, Any]) - Replace scheduling capability hints
• add_min_capability(key: str, value: Any) - Add a single capability hint

Result Format:

Container job results are returned as JSON-encoded text in result.result["text"]. Parse with json.loads() to access:

• exit_code: int - Container exit code (0 = success)
• stdout: str - Standard output from the container
• stderr: str - Standard error from the container
• output_files: list - Output file paths (if any)
• runtime_seconds: float - Total container runtime

BaseCall - Base Class

All job types inherit from BaseCall:

class BaseCall:
    type: str  # Job type identifier
    metadata: Dict[str, Any]  # User-defined metadata
    priority: int  # 0..100 (higher = scheduled first); default 10.
                   # Use PRIORITY_LOW (5), PRIORITY_DEFAULT (10), or
                   # PRIORITY_HIGH (20). Legacy strings ("low", "standard",
                   # "high") are still accepted and converted to ints
                   # automatically before the request is sent.
    timeout: Optional[int]  # Max execution time (seconds)
    callback_url: Optional[str]  # Webhook URL

3. JobDetails Class

Contains job status and results.

@dataclass
class JobDetails:
    job_id: str
    type: str  # "llm", "embed", "document", "container", "container_stream"
    status: str  # "queued", "processing", "completed", "failed", "cancelled"
    model: str
    created_at: datetime
    started_at: Optional[datetime]
    completed_at: Optional[datetime]
    estimated_cost: Optional[float]
    actual_cost: Optional[float]
    result: Optional[Any]
    error_message: Optional[str]
    metadata: Optional[Dict[str, Any]]
    priority: int  # 0..100 (default 10)
    retry_count: int  # Server-side retry attempts (default 0)
    user_id: Optional[str]
    encrypted: bool  # True if the result arrived encrypted and was decrypted

Methods:

• is_completed() -> bool - Check if job is done
• is_successful() -> bool - Check if job succeeded
• is_failed() -> bool - Check if job failed
• duration_seconds() -> Optional[float] - Calculate duration

4. Exception Classes

# Base exception
class MicroDCError(Exception): pass

# Specific exceptions
class AuthenticationError(MicroDCError): pass
class ValidationError(MicroDCError): pass
class APIError(MicroDCError): pass
class TimeoutError(MicroDCError): pass
class JobNotFoundError(MicroDCError): pass
class RateLimitError(MicroDCError): pass
class InsufficientCreditsError(MicroDCError): pass
class EncryptionError(MicroDCError): pass

Usage Patterns

Pattern 1: Simple Synchronous

from microdc import Client, LLMComplete

client = Client(api_key="mDC_...")

job = LLMComplete(model="llama3.3")
job.set_prompt("Hello!")

job_id = client.send_job(job)
client.wait_for_all()

result = client.get_job_details(job_id)
print(result.result)

# Acknowledge the job after processing
client.acknowledge_job(job_id)

Pattern 2: Callback-Based Async with Acknowledgment

from microdc import Client, LLMComplete

def handle_completion(client: Client, job_id: str):
    details = client.get_job_details(job_id)

    if details.is_successful():
        print(f"Success: {details.result}")

        # Acknowledge the job after processing
        ack = client.acknowledge_job(job_id)
        print(f"Acknowledged at: {ack['acknowledged_at']}")
    else:
        print(f"Failed: {details.error_message}")

client = Client(api_key="mDC_...")
client.set_callback(handle_completion)

job = LLMComplete(model="llama3.3")
job.set_prompt("Hello!")

job_id = client.send_job(job)
client.wait_for_all()

Pattern 3: Custom Type Tracking

from microdc import Client, LLMComplete

def callback(client: Client, job_id: str):
    details = client.get_job_details(job_id)

    # Route based on custom type
    if details.metadata.get("type") == "summarization":
        handle_summarization(details)
    elif details.metadata.get("type") == "translation":
        handle_translation(details)

    # Acknowledge after processing
    if details.is_successful():
        client.acknowledge_job(job_id)

client = Client(api_key="mDC_...")
client.set_callback(callback)

# Job 1: Summarization
job1 = LLMComplete(model="llama3.3")
job1.metadata = {"type": "summarization", "doc_id": "123"}
job1.set_prompt("Summarize: ...")
client.send_job(job1)

# Job 2: Translation
job2 = LLMComplete(model="llama3.3")
job2.metadata = {"type": "translation", "target_lang": "es"}
job2.set_prompt("Translate: Hello")
client.send_job(job2)

client.wait_for_all()

Pattern 4: Batch Processing

from microdc import Client, LLMComplete

client = Client(api_key="mDC_...")

questions = [
    "What is Python?",
    "Explain machine learning",
    "What is a neural network?"
]

job_ids = []
for question in questions:
    job = LLMComplete(model="llama3.3")
    job.set_prompt(question)
    job.metadata = {"question": question}

    job_id = client.send_job(job)
    job_ids.append(job_id)

client.wait_for_all(timeout=300)

for job_id in job_ids:
    details = client.get_job_details(job_id)
    print(f"Q: {details.metadata['question']}")
    print(f"A: {details.result['choices'][0]['message']['content']}\n")

    # Acknowledge each job
    if details.is_successful():
        client.acknowledge_job(job_id)

Pattern 5: Context Manager

from microdc import Client, LLMComplete

with Client(api_key="mDC_...") as client:
    job = LLMComplete(model="llama3.3")
    job.set_prompt("Hello!")

    job_id = client.send_job(job)
    client.wait_for_all()

    result = client.get_job_details(job_id)
    print(result.result)
# Client automatically closes

Pattern 6: Error Handling

from microdc import (
    Client, LLMChat,
    AuthenticationError,
    ValidationError,
    InsufficientCreditsError,
    APIError,
    TimeoutError
)

try:
    client = Client(api_key="mDC_...")

    job = LLMChat(model="llama3.3")
    job.add_user_message("Hello!")

    job_id = client.send_job(job)
    result = client.wait_for_job(job_id, timeout=60)

    print(result.result)

except AuthenticationError:
    print("Invalid API key")
except ValidationError as e:
    print(f"Invalid job config: {e}")
except InsufficientCreditsError as e:
    print(f"Need {e.required} credits, have {e.available}")
except TimeoutError:
    print("Job took too long")
except APIError as e:
    print(f"API error: {e}")
finally:
    client.close()

Pattern 7: Embeddings

from microdc import Client, LLMEmbed

client = Client(api_key="mDC_...")

job = LLMEmbed(model="text-embedding-ada-002")
job.add_texts([
    "The quick brown fox",
    "A journey of a thousand miles",
    "To be or not to be"
])

job_id = client.send_job(job)
client.wait_for_all()

details = client.get_job_details(job_id)
embeddings = details.result['embeddings']

for i, embedding in enumerate(embeddings):
    print(f"Text {i}: {len(embedding)} dimensions")

Pattern 8: Document Processing

from microdc import Client, DocumentCall

client = Client(api_key="mDC_...")

# Step 1: Upload the document file
upload_result = client.upload_file("contract.pdf")
file_token = upload_result['id']
print(f"File uploaded: {file_token}")

# Step 2: Create document processing job
job = DocumentCall(model="docling")
job.add_file(file_token)
job.metadata = {"document_type": "contract"}

# Step 3: Submit and wait
job_id = client.send_job(job)
client.wait_for_all()

# Step 4: Get results
details = client.get_job_details(job_id)
if details.is_successful():
    print("Document processed successfully!")
    print(details.result)

    # Acknowledge the job
    client.acknowledge_job(job_id)
else:
    print(f"Processing failed: {details.error_message}")

Pattern 9: Batch Document Processing

from microdc import Client, DocumentCall

client = Client(api_key="mDC_...")

# Upload multiple documents
documents = ["doc1.pdf", "doc2.pdf", "doc3.pdf"]
file_tokens = []

for doc_path in documents:
    upload_result = client.upload_file(doc_path)
    file_tokens.append(upload_result['id'])
    print(f"Uploaded {doc_path}")

# Create and submit jobs
job_ids = []
for i, token in enumerate(file_tokens):
    job = DocumentCall(model="docling")
    job.add_file(token)
    job.metadata = {"filename": documents[i]}

    job_id = client.send_job(job)
    job_ids.append(job_id)

# Wait for all jobs to complete
client.wait_for_all(timeout=600)

# Collect results
for job_id in job_ids:
    details = client.get_job_details(job_id)
    filename = details.metadata['filename']

    if details.is_successful():
        print(f"✓ {filename} processed successfully")
        client.acknowledge_job(job_id)
    else:
        print(f"✗ {filename} failed: {details.error_message}")

Pattern 10: Image Description (LLMComplete with Files)

from microdc import Client, LLMComplete

client = Client(api_key="mDC_...")

# Step 1: Upload the image
upload_result = client.upload_file("photo.jpg")
file_id = upload_result['id']

# Step 2: Create LLMComplete job with file attached
job = LLMComplete(model="qwen2.5vl:7b", temperature=0.3)
job.set_prompt("Describe this image in detail.")
job.add_file(file_id)

# Step 3: Submit and wait
job_id = client.send_job(job)
details = client.wait_for_job(job_id, timeout=120)

if details.is_successful():
    print(details.result)
    client.acknowledge_job(job_id)

Pattern 11: End-to-End Encryption

from microdc import Client, LLMComplete

# Encryption keys are auto-generated on first use (~/.microdc/keys/)
client = Client(api_key="mDC_...")

job = LLMComplete(model="llama3.3")
job.set_prompt("Sensitive query here")

# encrypt=True encrypts the payload with AES-256-GCM before submission
job_id = client.send_job(job, encrypt=True)
client.wait_for_all()

# Results are automatically decrypted
details = client.get_job_details(job_id)
print(details.result)       # Decrypted result
print(details.encrypted)    # True — indicates result was encrypted

# Custom key storage location
client = Client(api_key="mDC_...", encryption_key_path="/path/to/keys/")

Note: Encryption requires the cryptography package: pip install microdc-client[encryption]

Pattern 12: Container Job Execution

import json
from microdc import Client, ContainerJob

client = Client(api_key="mDC_...")

# Upload a script file
upload_result = client.upload_file("script.py")

# Create container job
job = ContainerJob(image="python:3.11")
job.set_command("python /input/script.py")
job.add_file(upload_result['id'])
job.timeout = 3600

job_id = client.send_job(job)
client.wait_for_all()
result = client.get_job_details(job_id)

if result.is_successful():
    # Container results are JSON-encoded in the text field
    container_result = json.loads(result.result["text"])
    print(f"Exit code: {container_result['exit_code']}")
    print(f"Runtime: {container_result['runtime_seconds']}s")
    print(container_result["stdout"])
    if container_result.get("stderr"):
        print(container_result["stderr"])
    client.acknowledge_job(job_id)

Pattern 13: GPU Container Job

import json
from microdc import PRIORITY_HIGH, Client, ContainerJob

client = Client(api_key="mDC_...")

job = ContainerJob(image="nvidia/cuda:12.0-runtime")
job.set_command("python /input/train.py")
job.enable_gpu()
job.set_memory_limit("16g")
job.set_cpu_count(8)
job.add_env_var("WANDB_API_KEY", "key123")
job.priority = PRIORITY_HIGH  # int (20). "high" is also accepted.
job.timeout = 7200

job_id = client.send_job(job)
client.wait_for_all()
result = client.get_job_details(job_id)

if result.is_successful():
    container_result = json.loads(result.result["text"])
    print(f"Exit code: {container_result['exit_code']}")
    print(container_result["stdout"])

Pattern 14: WorkerJob - Run a Project Folder

The simplest way to run code on MicroDC. Point WorkerJob at a folder containing a setup_run.sh entrypoint and the library handles zipping, uploading, and configuring the container automatically.

from microdc import Client, WorkerJob

client = Client(api_key="mDC_...")

# Point at your project folder and pick the right image for your language
job = WorkerJob(folder_path="./my_project", image="python:3.11")
job_id = client.send_job(job)
result = client.wait_for_job(job_id)

Works with any language:

WorkerJob(folder_path="./my_node_app", image="node:20")
WorkerJob(folder_path="./my_go_app", image="golang:1.22")
WorkerJob(folder_path="./my_rust_app", image="rust:1.77")

Folder structure:

my_project/
├── setup_run.sh       # Required - entrypoint executed by the worker
├── main.py            # Your code (any language)
└── requirements.txt   # Your dependencies

Configuration:

• folder_path: str - Path to the project folder (required)
• image: str - Docker image (required, e.g. "python:3.11", "node:20", "golang:1.22")
• network_mode: str - Network mode (default: "bridge" for internet access)
• Inherits all ContainerJob options: environment, gpu, mem_limit, cpu_count, etc.

How it works:

1. client.send_job() detects a WorkerJob
2. Zips the folder contents
3. Uploads the zip to MicroDC
4. Submits a container job that extracts and runs setup_run.sh
5. Cleans up the temporary zip file

Configuration

Environment Variables

MICRODC_API_KEY=mDC_499FC19C-686A-47C5-AA93-E619C55EBE98
MICRODC_BASE_URL=https://api.microdc.ai
MICRODC_TIMEOUT=30
MICRODC_POLL_INTERVAL=2.0
MICRODC_VERIFY_SSL=true
MICRODC_ENCRYPTION_KEY_PATH=~/.microdc/keys

Configuration File

Create .microdc.json in your project root:

{
    "api_key": "mDC_499FC19C-686A-47C5-AA93-E619C55EBE98",
    "base_url": "https://api.microdc.ai",
    "timeout": 30,
    "poll_interval": 2.0,
    "default_model": "llama3.3",
    "default_priority": 10
}

Programmatic Configuration

Config resolves settings from a file, the environment, or explicit values. The Client does not read Config automatically — load it, then pass the fields you need to the constructor:

from microdc import Client, Config

# Config.load() checks ./.microdc.json first, then MICRODC_* env vars
config = Config.load()
config.validate()

client = Client(
    api_key=config.api_key,
    base_url=config.base_url,
    timeout=config.timeout,
    verify_ssl=config.verify_ssl,
    auto_start_polling=config.auto_start_polling,
    encryption_key_path=config.encryption_key_path,
)

You can also build a Config directly:

config = Config(api_key="mDC_...", base_url="https://api.microdc.ai", timeout=60)

Note: Config.poll_interval, max_retries, retry_backoff, and default_model are carried by Config but are not yet consumed by Client. The polling interval is currently fixed at 2.0s and retry behaviour is fixed at the module defaults in microdc/core/http.py. A Client.from_config() constructor is on the roadmap.

API Endpoint Mapping

Client Method	HTTP Method	API Endpoint	Description
send_job(LLMComplete)	POST	/api/v1/jobs/submit	Submit LLM generation job
send_job(LLMChat)	POST	/api/v1/jobs/submit	Submit LLM chat job
send_job(LLMEmbed)	POST	/api/v1/jobs/submit	Submit embedding job
send_job(DocumentCall)	POST	/api/v1/jobs/submit	Submit document processing job
send_job(ContainerJob)	POST	/api/v1/jobs/submit	Submit container execution job
send_job(WorkerJob)	POST	/api/files/upload then /api/v1/jobs/submit	Zip + upload the folder, then submit
upload_file(file_path)	POST	/api/files/upload	Upload file for processing
delete_file(file_id)	DELETE	/api/files/{file_id}	Delete an uploaded file
create_download_token(file_id)	POST	/api/files/{file_id}/create-download-token	Create a one-time download token
get_job_details(job_id)	GET	/api/v1/jobs/{job_id}	Get full job details
get_job_status(job_id)	GET	/api/v1/jobs/{job_id}/status	Get job status only
cancel_job(job_id)	DELETE	/api/v1/jobs/{job_id}	Cancel job
acknowledge_job(job_id)	POST	/api/v1/jobs/{job_id}/acknowledge	Acknowledge job completion
list_jobs()	GET	/api/v1/jobs/	List user's jobs (trailing slash is required)

Request/Response Formats

LLM Job Submission

Request:

LLMChat (llm_interaction_type: "chat") sends system + messages:

{
    "type": "llm",
    "model": "llama3.3",
    "llm_interaction_type": "chat",
    "input_modalities": ["text"],
    "output_modalities": ["text"],
    "payload": {
        "system": "You are a helpful assistant.",
        "messages": [
            {"role": "user", "content": "Why is the sky blue?"}
        ],
        "temperature": 0.7,
        "top_p": 1.0,
        "frequency_penalty": 0.0,
        "presence_penalty": 0.0,
        "stream": false,
        "max_tokens": 500
    },
    "priority": 10,
    "estimated_cost": 0.0,
    "metadata": {
        "custom_type": "test_call"
    }
}

LLMComplete (llm_interaction_type: "generation") sends a single prompt instead of system/messages, and adds a top-level file_ids array when files have been attached with add_file().

Response:

{
    "job_id": "job_abc123",
    "status": "queued",
    "estimated_completion": "2025-01-15T10:30:00Z",
    "estimated_cost": 150
}

Job Details Response

{
    "job_id": "job_abc123",
    "user_id": "user_xyz",
    "type": "llm",
    "model": "llama3.3",
    "status": "completed",
    "priority": 10,
    "estimated_cost": 150,
    "actual_cost": 145,
    "created_at": "2025-01-15T10:25:00Z",
    "started_at": "2025-01-15T10:26:00Z",
    "completed_at": "2025-01-15T10:28:30Z",
    "result": {
        "choices": [
            {
                "message": {
                    "role": "assistant",
                    "content": "The sky appears blue because of Rayleigh scattering..."
                },
                "finish_reason": "stop"
            }
        ],
        "usage": {
            "prompt_tokens": 20,
            "completion_tokens": 150,
            "total_tokens": 170
        }
    },
    "metadata": {
        "custom_type": "test_call"
    }
}

Embedding Job Submission

Request:

{
    "type": "embed",
    "model": "text-embedding-ada-002",
    "payload": {
        "texts": ["Hello world", "Goodbye world"],
        "normalize": true,
        "encoding_format": "float"
    },
    "priority": 10,
    "estimated_cost": 0.0
}

Response:

{
    "job_id": "job_def456",
    "status": "queued",
    "estimated_completion": "2025-01-15T10:30:00Z",
    "estimated_cost": 50
}

Error Response

{
    "error": {
        "code": "insufficient_credits",
        "message": "Insufficient credits to execute job",
        "details": {
            "required": 150,
            "available": 50
        }
    }
}

Implementation Details

Authentication

The client uses Bearer token authentication:

Authorization: Bearer mDC_499FC19C-686A-47C5-AA93-E619C55EBE98

Polling Mechanism

The client automatically polls for job completion:

1. Background thread starts on client initialization
2. Polls every 2 seconds
3. When job completes, invokes callback if set
4. Updates internal job tracking

# Polling is on by default; pass auto_start_polling=False to manage it yourself
client = Client(api_key="mDC_...", auto_start_polling=False)

Note: The 2.0s interval is currently fixed inside PollingManager. It is not yet exposed on the Client constructor.

Retry Logic

Automatic retries for transient errors:

• Retry attempts: 3
• Retry delay: 1 second (exponential backoff)
• Retryable status codes: 408, 429, 500, 502, 503, 504

Connection Pooling

Uses a shared httpx.Client (HTTP/1.1) for connection pooling and performance. It is created with follow_redirects=True. HTTP/2 is intentionally disabled — see Dependencies below.

Thread Safety

The client is thread-safe for:

• Job submission
• Job status queries
• Callback invocation

Testing

Unit Tests

# tests/unit/test_llm_complete.py
import pytest
from microdc import LLMComplete, LLMChat, ValidationError

def test_llm_complete_creation():
    job = LLMComplete(model="llama3.3")
    job.set_prompt("Hello")
    assert job.temperature == 0.7

def test_llm_complete_validation():
    job = LLMComplete(model="")
    with pytest.raises(ValidationError):
        job.validate()

def test_add_messages():
    job = LLMChat(model="llama3.3")
    job.add_user_message("Hello")
    assert len(job.messages) == 1
    assert job.messages[0]["role"] == "user"

Integration Tests

# tests/integration/test_client.py
import pytest
from microdc import Client, LLMComplete

@pytest.mark.integration
def test_job_submission(test_api_key):
    client = Client(api_key=test_api_key)

    job = LLMComplete(model="llama3.3")
    job.set_prompt("Test")

    job_id = client.send_job(job)
    assert job_id is not None

    details = client.wait_for_job(job_id, timeout=60)
    assert details.is_successful()

Package Structure

microdc-client/
├── microdc/
│   ├── __init__.py            # Public API exports
│   ├── _version.py            # Single source of truth for the version
│   ├── client.py              # Client class
│   ├── jobs/                  # Job types
│   │   ├── base.py            # BaseCall abstract class + priority constants
│   │   ├── llm_complete.py    # LLMComplete (single-turn generation)
│   │   ├── llm_chat.py        # LLMChat (multi-turn chat)
│   │   ├── embed_call.py      # LLMEmbed (embeddings)
│   │   ├── document_call.py   # DocumentCall (file processing)
│   │   ├── container_job.py   # ContainerJob (Docker container execution)
│   │   ├── worker_job.py      # WorkerJob (folder-based streaming container)
│   │   └── job_details.py     # JobDetails dataclass
│   ├── core/                  # Core functionality
│   │   ├── http.py            # HTTPTransport
│   │   ├── polling.py         # PollingManager
│   │   ├── encryption.py      # EncryptionManager (RSA/AES-256-GCM)
│   │   └── config.py          # Configuration
│   └── exceptions/            # Error handling
│       └── errors.py          # All exception classes
├── tests/
│   ├── unit/                  # Unit tests
│   ├── integration/           # Integration tests
│   └── conftest.py            # Shared fixtures
├── examples/                  # Example scripts (see examples/README.md)
├── notebooks/                 # Jupyter notebooks
├── docs/                      # Documentation (mkdocs site)
├── setup.py
├── pyproject.toml
├── requirements.txt
├── README.md
├── LICENSE
└── CHANGELOG.md

Dependencies

Required

• httpx>=0.27.0 - Modern HTTP client with an async-ready API (configured with follow_redirects=True so the API's 307s — e.g. trailing-slash normalization on /api/v1/jobs — are followed transparently). Note: HTTP/2 is disabled as of v1.0.13 for thread-safety; the client uses the thread-safe HTTP/1.1 connection pool.
• typing-extensions>=4.5.0 - Type hints (Python < 3.10)

Optional

• cryptography>=41.0.0 - End-to-end encryption (pip install microdc-client[encryption])

Development

Install with pip install -e ".[dev]":

• pytest>=7.0.0 - Testing framework
• pytest-cov>=4.0.0 - Coverage reporting
• pytest-mock>=3.10.0 - Mocking fixtures used by the test suite
• pytest-asyncio>=0.21.0 - Async test support
• cryptography>=41.0.0 - Required by the encryption test suite
• black>=23.0.0 - Code formatting
• ruff>=0.1.0 - Linting
• mypy>=1.0.0 - Type checking

Continuous Integration

.gitlab-ci.yml runs three jobs:

Job	Stage	What it does
test	test	pytest with coverage; publishes JUnit + Cobertura reports to the MR
lint	test	ruff check, black --check, mypy microdc/
pages	deploy	mkdocs build --strict → GitLab Pages (main only)

Versioning

Follow Semantic Versioning (SemVer):

• 1.0.0 - Initial stable release
• 1.1.0 - Add streaming support
• 1.1.1 - Fix timeout bug
• 2.0.0 - Breaking API changes

Example: Complete Application

from microdc import Client, LLMChat

API_KEY = "mDC_499FC19C-686A-47C5-AA93-E619C55EBE98"

def callback(client: Client, job_id: str):
    """Handle job completion."""
    details = client.get_job_details(job_id)

    if details.metadata.get("type") == "test_call":
        print(f"Status: {details.status}")
    else:
        print(f"Result: {details.result}")

if __name__ == "__main__":
    # Initialize client
    client = Client(api_key=API_KEY)
    client.set_callback(callback)

    # Create LLM chat job
    job = LLMChat(model="llama3.3")
    job.metadata = {"type": "test_call"}
    job.set_system("You are a helpful assistant. You enjoy writing short, concise, and humorous answers.")
    job.add_user_message("Why is the sky blue?")

    # Submit job
    job_id = client.send_job(job)
    print(f"Job ID Created: {job_id}")

    # Wait for completion
    client.wait_for_all()

    # Interactive debugging
    import IPython
    IPython.embed()

Jupyter Notebooks

Interactive Jupyter notebook examples are available in the notebooks/ directory:

01_basic_usage.ipynb

Introduction to the MicroDC client library covering:

• Simple LLM calls
• Callback-based async processing
• Embedding generation
• Multi-turn conversations
• Job management
• Context manager patterns

02_simple_rag.ipynb

Complete RAG (Retrieval-Augmented Generation) implementation:

• PDF text extraction and chunking
• Embedding generation using MicroDC (cloud-based, no local GPU needed)
• Vector store creation and similarity search
• Question-answering with retrieved context
• Interactive Q&A sessions
• Perfect for running on low-power computers

03_batch_processing.ipynb

Efficient batch processing patterns:

• Batch text classification
• Large-scale embedding generation
• Multi-document summarization
• Progress tracking with metadata
• Error handling and retry logic
• Performance metrics and optimization tips

These notebooks demonstrate how to leverage MicroDC's distributed computing platform to run sophisticated AI applications on any computer, regardless of hardware capabilities.

Roadmap

v1.0.x (Shipped) ✅

• Client class with authentication
• LLMComplete / LLMChat for LLM inference
• LLMEmbed for embeddings
• DocumentCall for document processing
• ContainerJob / WorkerJob for Docker container execution (v1.0.11)
• End-to-end payload/result encryption (v1.0.10)
• Integer job priority with legacy-string compatibility (v1.0.12)
• Callback-based async handling and automatic polling
• Error handling and retries
• Context manager support
• Configuration management (Config class)
• File upload, deletion, and download-token helpers
• Example scripts, notebooks, and test infrastructure

v1.1.0 (Streaming Support)

• Streaming responses for LLMComplete / LLMChat
• Real-time token streaming
• Stream callback interface

v1.2.0 (Advanced Features)

• Async/await API with asyncio
• Progress callbacks
• Cost estimation API

v2.0.0 (Enterprise Features)

• Custom model deployment API
• A/B testing support
• Advanced caching
• Webhook verification

Changelog

See CHANGELOG.md for detailed version history and release notes.

Latest Release: v1.0.13 (2026-06-08)

Fixed:

• ✅ Intermittent KeyError: <stream_id> crashes under concurrent load — HTTP/2 is now disabled in favour of the thread-safe HTTP/1.1 connection pool

Shipped in the v1.0.x line:

• ✅ Complete client implementation with all core functionality
• ✅ LLM, Embed, Document, Container, and Worker job types
• ✅ Callback-based async patterns
• ✅ File upload, deletion, and download-token support
• ✅ End-to-end payload/result encryption (v1.0.10)
• ✅ Integer job priority with legacy-string compatibility (v1.0.12)
• ✅ Comprehensive error handling

Contributing

We welcome contributions! Please read our Contributing Guide to learn about:

• Setting up your development environment
• Our coding standards and style guide
• How to submit pull requests
• Testing requirements
• Release process

License

This project is licensed under the MIT License - see the LICENSE file for details.

Support

• Documentation: https://microdc.gitlab.io/python-client/
• Issues: https://gitlab.com/microdc/python-client/-/issues
• Merge Requests: https://gitlab.com/microdc/python-client/-/merge_requests
• Email: support@microdc.ai

---

Library Version: 1.0.13 Last Updated: 2026-07-31 Status: Production Ready