Skip to content

Job Classes

All job types inherit from BaseCall and provide specialized configurations for different inference tasks.

LLMComplete

Single-turn text generation using a prompt.

LLMComplete dataclass

Bases: BaseCall

LLM Completion job - for text completion, generation, and single-turn tasks.

This class is designed for generation endpoints that accept a single prompt without system context. It's optimized for: - Text completion - Code generation - Creative writing - Single-turn responses - Q&A without conversation context

Note: This does NOT support system messages. Use LLMChat for conversational interactions with system/user/assistant message flows.

Example

job = LLMComplete(model="llama3.3") job.set_prompt("Write a haiku about Python") job.temperature = 0.9

Source code in microdc/jobs/llm_complete.py 
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
@dataclass
class LLMComplete(BaseCall):
    """
    LLM Completion job - for text completion, generation, and single-turn tasks.

    This class is designed for generation endpoints that accept a single prompt
    without system context. It's optimized for:
    - Text completion
    - Code generation
    - Creative writing
    - Single-turn responses
    - Q&A without conversation context

    Note: This does NOT support system messages. Use LLMChat for conversational
    interactions with system/user/assistant message flows.

    Example:
        >>> job = LLMComplete(model="llama3.3")
        >>> job.set_prompt("Write a haiku about Python")
        >>> job.temperature = 0.9
    """

    model: str = ""
    prompt: str = ""
    file_tokens: List[str] = field(default_factory=list)
    input_modalities: List[str] = field(default_factory=lambda: ["text"])
    output_modalities: List[str] = field(default_factory=lambda: ["text"])
    temperature: float = 0.7
    max_tokens: Optional[int] = None
    top_p: float = 1.0
    top_k: Optional[int] = None
    frequency_penalty: float = 0.0
    presence_penalty: float = 0.0
    stop: Optional[List[str]] = None
    stream: bool = False

    def __post_init__(self) -> None:
        self.type = "llm"

    def set_prompt(self, prompt: str) -> None:
        """Set the generation prompt."""
        self.prompt = prompt

    def add_file(self, file_token: str) -> None:
        """
        Add a file token to the completion job.

        Args:
            file_token: File ID from client.upload_file() response
        """
        self.file_tokens.append(file_token)

    def add_files(self, file_tokens: List[str]) -> None:
        """
        Add multiple file tokens to the completion job.

        Args:
            file_tokens: List of file IDs from client.upload_file() responses
        """
        self.file_tokens.extend(file_tokens)

    def to_api_payload(self) -> Dict[str, Any]:
        """Convert to API request format matching JobRequest schema."""
        # Build inner payload with prompt and parameters
        inner_payload: Dict[str, Any] = {
            "prompt": self.prompt,
            "temperature": self.temperature,
            "top_p": self.top_p,
            "frequency_penalty": self.frequency_penalty,
            "presence_penalty": self.presence_penalty,
            "stream": self.stream,
        }

        if self.max_tokens is not None:
            inner_payload["max_tokens"] = self.max_tokens
        if self.top_k is not None:
            inner_payload["top_k"] = self.top_k
        if self.stop is not None:
            inner_payload["stop"] = self.stop

        # Build outer payload matching JobRequest schema
        payload: Dict[str, Any] = {
            "type": "llm",
            "model": self.model,
            "llm_interaction_type": "generation",
            "input_modalities": self.input_modalities,
            "output_modalities": self.output_modalities,
            "payload": inner_payload,
            "priority": normalize_priority(self.priority),
            "estimated_cost": 0.0,
        }

        if self.file_tokens:
            payload["file_ids"] = self.file_tokens

        return payload

    def validate(self) -> None:
        """Validate LLM completion job configuration."""
        if not self.model:
            raise ValidationError("Model must be specified")
        if not self.prompt:
            raise ValidationError("Prompt is required for LLMComplete")
        if not 0.0 <= self.temperature <= 2.0:
            raise ValidationError("Temperature must be between 0.0 and 2.0")
        if not 0.0 <= self.top_p <= 1.0:
            raise ValidationError("top_p must be between 0.0 and 1.0")
        if self.max_tokens is not None and self.max_tokens <= 0:
            raise ValidationError("max_tokens must be positive")
        if self.top_k is not None and self.top_k <= 0:
            raise ValidationError("top_k must be positive")

set_prompt(prompt)

Set the generation prompt.

Source code in microdc/jobs/llm_complete.py 
49
50
51
def set_prompt(self, prompt: str) -> None:
    """Set the generation prompt."""
    self.prompt = prompt

add_file(file_token)

Add a file token to the completion job.

Parameters:

Name Type Description Default
file_token str

File ID from client.upload_file() response

 required
Source code in microdc/jobs/llm_complete.py 
53
54
55
56
57
58
59
60
def add_file(self, file_token: str) -> None:
    """
    Add a file token to the completion job.

    Args:
        file_token: File ID from client.upload_file() response
    """
    self.file_tokens.append(file_token)

add_files(file_tokens)

Add multiple file tokens to the completion job.

Parameters:

Name Type Description Default
file_tokens List[str]

List of file IDs from client.upload_file() responses

 required
Source code in microdc/jobs/llm_complete.py 
62
63
64
65
66
67
68
69
def add_files(self, file_tokens: List[str]) -> None:
    """
    Add multiple file tokens to the completion job.

    Args:
        file_tokens: List of file IDs from client.upload_file() responses
    """
    self.file_tokens.extend(file_tokens)

to_api_payload()

Convert to API request format matching JobRequest schema.

Source code in microdc/jobs/llm_complete.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
def to_api_payload(self) -> Dict[str, Any]:
    """Convert to API request format matching JobRequest schema."""
    # Build inner payload with prompt and parameters
    inner_payload: Dict[str, Any] = {
        "prompt": self.prompt,
        "temperature": self.temperature,
        "top_p": self.top_p,
        "frequency_penalty": self.frequency_penalty,
        "presence_penalty": self.presence_penalty,
        "stream": self.stream,
    }

    if self.max_tokens is not None:
        inner_payload["max_tokens"] = self.max_tokens
    if self.top_k is not None:
        inner_payload["top_k"] = self.top_k
    if self.stop is not None:
        inner_payload["stop"] = self.stop

    # Build outer payload matching JobRequest schema
    payload: Dict[str, Any] = {
        "type": "llm",
        "model": self.model,
        "llm_interaction_type": "generation",
        "input_modalities": self.input_modalities,
        "output_modalities": self.output_modalities,
        "payload": inner_payload,
        "priority": normalize_priority(self.priority),
        "estimated_cost": 0.0,
    }

    if self.file_tokens:
        payload["file_ids"] = self.file_tokens

    return payload

validate()

Validate LLM completion job configuration.

Source code in microdc/jobs/llm_complete.py 
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def validate(self) -> None:
    """Validate LLM completion job configuration."""
    if not self.model:
        raise ValidationError("Model must be specified")
    if not self.prompt:
        raise ValidationError("Prompt is required for LLMComplete")
    if not 0.0 <= self.temperature <= 2.0:
        raise ValidationError("Temperature must be between 0.0 and 2.0")
    if not 0.0 <= self.top_p <= 1.0:
        raise ValidationError("top_p must be between 0.0 and 1.0")
    if self.max_tokens is not None and self.max_tokens <= 0:
        raise ValidationError("max_tokens must be positive")
    if self.top_k is not None and self.top_k <= 0:
        raise ValidationError("top_k must be positive")

LLMChat

Multi-turn conversational AI with system/user/assistant messages.

LLMChat dataclass

Bases: BaseCall

LLM Chat job - for conversational and multi-turn interactions.

This class supports full message-based interactions with system, user, and assistant roles. It's designed for: - Conversational AI - Multi-turn dialogue - Question answering with context - Assistant-style interactions - Context-aware responses

Example

chat = LLMChat(model="gpt-4") chat.set_system("You are a helpful assistant") chat.add_user_message("What is Python?") chat.add_assistant_message("Python is a programming language") chat.add_user_message("Tell me more")

Source code in microdc/jobs/llm_chat.py 
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
@dataclass
class LLMChat(BaseCall):
    """
    LLM Chat job - for conversational and multi-turn interactions.

    This class supports full message-based interactions with system, user,
    and assistant roles. It's designed for:
    - Conversational AI
    - Multi-turn dialogue
    - Question answering with context
    - Assistant-style interactions
    - Context-aware responses

    Example:
        >>> chat = LLMChat(model="gpt-4")
        >>> chat.set_system("You are a helpful assistant")
        >>> chat.add_user_message("What is Python?")
        >>> chat.add_assistant_message("Python is a programming language")
        >>> chat.add_user_message("Tell me more")
    """

    model: str = ""
    system: str = ""
    messages: List[Dict[str, str]] = field(default_factory=list)
    input_modalities: List[str] = field(default_factory=lambda: ["text"])
    output_modalities: List[str] = field(default_factory=lambda: ["text"])
    temperature: float = 0.7
    max_tokens: Optional[int] = None
    top_p: float = 1.0
    top_k: Optional[int] = None
    frequency_penalty: float = 0.0
    presence_penalty: float = 0.0
    stop: Optional[List[str]] = None
    stream: bool = False

    def __post_init__(self) -> None:
        self.type = "llm"

    def set_system(self, content: str) -> None:
        """Set the system message for the conversation context."""
        self.system = content

    def add_user_message(self, content: str) -> None:
        """Add a user message to the conversation."""
        self.messages.append({"role": "user", "content": content})

    def add_assistant_message(self, content: str) -> None:
        """Add an assistant message to the conversation."""
        self.messages.append({"role": "assistant", "content": content})

    def to_api_payload(self) -> Dict[str, Any]:
        """Convert to API request format matching JobRequest schema."""
        # Build inner payload with system, messages and parameters
        inner_payload: Dict[str, Any] = {
            "system": self.system,
            "messages": self.messages,
            "temperature": self.temperature,
            "top_p": self.top_p,
            "frequency_penalty": self.frequency_penalty,
            "presence_penalty": self.presence_penalty,
            "stream": self.stream,
        }

        if self.max_tokens is not None:
            inner_payload["max_tokens"] = self.max_tokens
        if self.top_k is not None:
            inner_payload["top_k"] = self.top_k
        if self.stop is not None:
            inner_payload["stop"] = self.stop

        # Build outer payload matching JobRequest schema
        payload: Dict[str, Any] = {
            "type": "llm",
            "model": self.model,
            "llm_interaction_type": "chat",
            "input_modalities": self.input_modalities,
            "output_modalities": self.output_modalities,
            "payload": inner_payload,
            "priority": normalize_priority(self.priority),
            "estimated_cost": 0.0,
        }

        return payload

    def validate(self) -> None:
        """Validate LLM chat job configuration."""
        if not self.model:
            raise ValidationError("Model must be specified")
        if not self.messages:
            raise ValidationError("At least one message is required for LLMChat")
        if not 0.0 <= self.temperature <= 2.0:
            raise ValidationError("Temperature must be between 0.0 and 2.0")
        if not 0.0 <= self.top_p <= 1.0:
            raise ValidationError("top_p must be between 0.0 and 1.0")
        if self.max_tokens is not None and self.max_tokens <= 0:
            raise ValidationError("max_tokens must be positive")
        if self.top_k is not None and self.top_k <= 0:
            raise ValidationError("top_k must be positive")

set_system(content)

Set the system message for the conversation context.

Source code in microdc/jobs/llm_chat.py 
48
49
50
def set_system(self, content: str) -> None:
    """Set the system message for the conversation context."""
    self.system = content

add_user_message(content)

Add a user message to the conversation.

Source code in microdc/jobs/llm_chat.py 
52
53
54
def add_user_message(self, content: str) -> None:
    """Add a user message to the conversation."""
    self.messages.append({"role": "user", "content": content})

add_assistant_message(content)

Add an assistant message to the conversation.

Source code in microdc/jobs/llm_chat.py 
56
57
58
def add_assistant_message(self, content: str) -> None:
    """Add an assistant message to the conversation."""
    self.messages.append({"role": "assistant", "content": content})

to_api_payload()

Convert to API request format matching JobRequest schema.

Source code in microdc/jobs/llm_chat.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def to_api_payload(self) -> Dict[str, Any]:
    """Convert to API request format matching JobRequest schema."""
    # Build inner payload with system, messages and parameters
    inner_payload: Dict[str, Any] = {
        "system": self.system,
        "messages": self.messages,
        "temperature": self.temperature,
        "top_p": self.top_p,
        "frequency_penalty": self.frequency_penalty,
        "presence_penalty": self.presence_penalty,
        "stream": self.stream,
    }

    if self.max_tokens is not None:
        inner_payload["max_tokens"] = self.max_tokens
    if self.top_k is not None:
        inner_payload["top_k"] = self.top_k
    if self.stop is not None:
        inner_payload["stop"] = self.stop

    # Build outer payload matching JobRequest schema
    payload: Dict[str, Any] = {
        "type": "llm",
        "model": self.model,
        "llm_interaction_type": "chat",
        "input_modalities": self.input_modalities,
        "output_modalities": self.output_modalities,
        "payload": inner_payload,
        "priority": normalize_priority(self.priority),
        "estimated_cost": 0.0,
    }

    return payload

validate()

Validate LLM chat job configuration.

Source code in microdc/jobs/llm_chat.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def validate(self) -> None:
    """Validate LLM chat job configuration."""
    if not self.model:
        raise ValidationError("Model must be specified")
    if not self.messages:
        raise ValidationError("At least one message is required for LLMChat")
    if not 0.0 <= self.temperature <= 2.0:
        raise ValidationError("Temperature must be between 0.0 and 2.0")
    if not 0.0 <= self.top_p <= 1.0:
        raise ValidationError("top_p must be between 0.0 and 1.0")
    if self.max_tokens is not None and self.max_tokens <= 0:
        raise ValidationError("max_tokens must be positive")
    if self.top_k is not None and self.top_k <= 0:
        raise ValidationError("top_k must be positive")

LLMEmbed

Text embedding generation for semantic search and RAG.

LLMEmbed dataclass

Bases: BaseCall

Embedding generation job configuration.

Use this for generating text embeddings for semantic search, similarity, clustering, and other vector-based operations.

Example

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

Source code in microdc/jobs/embed_call.py 
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
@dataclass
class LLMEmbed(BaseCall):
    """
    Embedding generation job configuration.

    Use this for generating text embeddings for semantic search, similarity,
    clustering, and other vector-based operations.

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

    model: str = ""
    input_texts: List[str] = field(default_factory=list)
    dimensions: Optional[int] = None
    normalize: bool = True
    encoding_format: str = "float"  # float or base64

    def __post_init__(self) -> None:
        self.type = "embed"

    def add_text(self, text: str) -> None:
        """Add a text to the embedding batch."""
        self.input_texts.append(text)

    def add_texts(self, texts: List[str]) -> None:
        """Add multiple texts to the embedding batch."""
        self.input_texts.extend(texts)

    def to_api_payload(self) -> Dict[str, Any]:
        """Convert to API request format matching JobRequest schema."""
        # Build inner payload with texts and parameters
        inner_payload: Dict[str, Any] = {
            "texts": self.input_texts,  # API expects 'texts' field
            "normalize": self.normalize,
            "encoding_format": self.encoding_format,
        }

        if self.dimensions is not None:
            inner_payload["dimensions"] = self.dimensions

        # Build outer payload matching JobRequest schema
        payload: Dict[str, Any] = {
            "type": "embed",
            "model": self.model,
            "payload": inner_payload,
            "priority": normalize_priority(self.priority),
            "estimated_cost": 0.0,  # Placeholder for future feature
        }

        return payload

    def validate(self) -> None:
        """Validate embedding job configuration."""
        if not self.model:
            raise ValidationError("Model must be specified")
        if not self.input_texts:
            raise ValidationError("At least one input text is required")
        if any(not text.strip() for text in self.input_texts):
            raise ValidationError("All input texts must be non-empty")
        if self.encoding_format not in ("float", "base64"):
            raise ValidationError("encoding_format must be 'float' or 'base64'")
        if self.dimensions is not None and self.dimensions <= 0:
            raise ValidationError("dimensions must be positive")

add_text(text)

Add a text to the embedding batch.

Source code in microdc/jobs/embed_call.py 
33
34
35
def add_text(self, text: str) -> None:
    """Add a text to the embedding batch."""
    self.input_texts.append(text)

add_texts(texts)

Add multiple texts to the embedding batch.

Source code in microdc/jobs/embed_call.py 
37
38
39
def add_texts(self, texts: List[str]) -> None:
    """Add multiple texts to the embedding batch."""
    self.input_texts.extend(texts)

to_api_payload()

Convert to API request format matching JobRequest schema.

Source code in microdc/jobs/embed_call.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
def to_api_payload(self) -> Dict[str, Any]:
    """Convert to API request format matching JobRequest schema."""
    # Build inner payload with texts and parameters
    inner_payload: Dict[str, Any] = {
        "texts": self.input_texts,  # API expects 'texts' field
        "normalize": self.normalize,
        "encoding_format": self.encoding_format,
    }

    if self.dimensions is not None:
        inner_payload["dimensions"] = self.dimensions

    # Build outer payload matching JobRequest schema
    payload: Dict[str, Any] = {
        "type": "embed",
        "model": self.model,
        "payload": inner_payload,
        "priority": normalize_priority(self.priority),
        "estimated_cost": 0.0,  # Placeholder for future feature
    }

    return payload

validate()

Validate embedding job configuration.

Source code in microdc/jobs/embed_call.py 
64
65
66
67
68
69
70
71
72
73
74
75
def validate(self) -> None:
    """Validate embedding job configuration."""
    if not self.model:
        raise ValidationError("Model must be specified")
    if not self.input_texts:
        raise ValidationError("At least one input text is required")
    if any(not text.strip() for text in self.input_texts):
        raise ValidationError("All input texts must be non-empty")
    if self.encoding_format not in ("float", "base64"):
        raise ValidationError("encoding_format must be 'float' or 'base64'")
    if self.dimensions is not None and self.dimensions <= 0:
        raise ValidationError("dimensions must be positive")

DocumentCall

Document processing for file-based analysis workflows.

DocumentCall dataclass

Bases: BaseCall

Document processing job configuration for file-only processing models.

Document processing models like docling analyze files without requiring text prompts. Files MUST be uploaded before creating the document processing job.

Workflow
  1. Upload file(s) using client.upload_file()
  2. Get file token(s) from upload response
  3. Create DocumentCall with model and file tokens
  4. Submit job using client.send_job()
Example

Step 1: Upload file

upload_result = client.upload_file("document.pdf") file_token = upload_result['id']

Step 2: Create document processing job

job = DocumentCall(model="docling") job.add_file(file_token)

Step 3: Submit job

job_id = client.send_job(job)

Source code in microdc/jobs/document_call.py 
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
@dataclass
class DocumentCall(BaseCall):
    """
    Document processing job configuration for file-only processing models.

    Document processing models like docling analyze files without requiring text prompts.
    Files MUST be uploaded before creating the document processing job.

    Workflow:
        1. Upload file(s) using client.upload_file()
        2. Get file token(s) from upload response
        3. Create DocumentCall with model and file tokens
        4. Submit job using client.send_job()

    Example:
        >>> # Step 1: Upload file
        >>> upload_result = client.upload_file("document.pdf")
        >>> file_token = upload_result['id']
        >>>
        >>> # Step 2: Create document processing job
        >>> job = DocumentCall(model="docling")
        >>> job.add_file(file_token)
        >>>
        >>> # Step 3: Submit job
        >>> job_id = client.send_job(job)
    """

    model: str = ""
    file_tokens: List[str] = field(default_factory=list)
    max_tokens: Optional[int] = None
    temperature: float = 0.7

    def __post_init__(self) -> None:
        self.type = "document"

    def add_file(self, file_token: str) -> None:
        """
        Add a file token to the document processing job.

        Args:
            file_token: Token from file upload response

        Note:
            Files must be uploaded using client.upload_file() before
            adding their tokens to the job.
        """
        self.file_tokens.append(file_token)

    def add_files(self, file_tokens: List[str]) -> None:
        """
        Add multiple file tokens to the document processing job.

        Args:
            file_tokens: List of tokens from file upload responses
        """
        self.file_tokens.extend(file_tokens)

    def to_api_payload(self) -> Dict[str, Any]:
        """
        Convert to API request format.

        Note: Document processing jobs do NOT include 'input' field in payload.
        They only process uploaded files referenced by file_ids at the top level.

        The API expects file_ids (not file_tokens) at the top level of the request.
        """
        # Build inner payload with document processing parameters
        inner_payload: Dict[str, Any] = {
            "temperature": self.temperature,
        }

        if self.max_tokens is not None:
            inner_payload["max_tokens"] = self.max_tokens

        # Build outer payload matching JobRequest schema
        # file_ids is at top level, not in payload
        payload: Dict[str, Any] = {
            "type": "document",
            "model": self.model,
            "payload": inner_payload,
            "priority": normalize_priority(self.priority),
            "file_ids": self.file_tokens,  # API expects 'file_ids' at top level
            "estimated_cost": 0.0,  # Placeholder for future feature
        }

        if self.timeout is not None:
            payload["timeout"] = self.timeout
        if self.callback_url is not None:
            payload["callback_url"] = self.callback_url
        if self.metadata:
            payload["metadata"] = self.metadata

        return payload

    def validate(self) -> None:
        """Validate document processing job configuration."""
        if not self.model:
            raise ValidationError("Model must be specified")
        if not self.file_tokens:
            raise ValidationError(
                "At least one file token is required. Upload files first using client.upload_file()"
            )
        if self.max_tokens is not None and self.max_tokens < 1:
            raise ValidationError("max_tokens must be positive")
        if not 0.0 <= self.temperature <= 2.0:
            raise ValidationError("Temperature must be between 0.0 and 2.0")

add_file(file_token)

Add a file token to the document processing job.

Parameters:

Name Type Description Default
file_token str

Token from file upload response

 required
Note

Files must be uploaded using client.upload_file() before adding their tokens to the job.

Source code in microdc/jobs/document_call.py 
45
46
47
48
49
50
51
52
53
54
55
56
def add_file(self, file_token: str) -> None:
    """
    Add a file token to the document processing job.

    Args:
        file_token: Token from file upload response

    Note:
        Files must be uploaded using client.upload_file() before
        adding their tokens to the job.
    """
    self.file_tokens.append(file_token)

add_files(file_tokens)

Add multiple file tokens to the document processing job.

Parameters:

Name Type Description Default
file_tokens List[str]

List of tokens from file upload responses

 required
Source code in microdc/jobs/document_call.py 
58
59
60
61
62
63
64
65
def add_files(self, file_tokens: List[str]) -> None:
    """
    Add multiple file tokens to the document processing job.

    Args:
        file_tokens: List of tokens from file upload responses
    """
    self.file_tokens.extend(file_tokens)

to_api_payload()

Convert to API request format.

Note: Document processing jobs do NOT include 'input' field in payload. They only process uploaded files referenced by file_ids at the top level.

The API expects file_ids (not file_tokens) at the top level of the request.

Source code in microdc/jobs/document_call.py 
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def to_api_payload(self) -> Dict[str, Any]:
    """
    Convert to API request format.

    Note: Document processing jobs do NOT include 'input' field in payload.
    They only process uploaded files referenced by file_ids at the top level.

    The API expects file_ids (not file_tokens) at the top level of the request.
    """
    # Build inner payload with document processing parameters
    inner_payload: Dict[str, Any] = {
        "temperature": self.temperature,
    }

    if self.max_tokens is not None:
        inner_payload["max_tokens"] = self.max_tokens

    # Build outer payload matching JobRequest schema
    # file_ids is at top level, not in payload
    payload: Dict[str, Any] = {
        "type": "document",
        "model": self.model,
        "payload": inner_payload,
        "priority": normalize_priority(self.priority),
        "file_ids": self.file_tokens,  # API expects 'file_ids' at top level
        "estimated_cost": 0.0,  # Placeholder for future feature
    }

    if self.timeout is not None:
        payload["timeout"] = self.timeout
    if self.callback_url is not None:
        payload["callback_url"] = self.callback_url
    if self.metadata:
        payload["metadata"] = self.metadata

    return payload

validate()

Validate document processing job configuration.

Source code in microdc/jobs/document_call.py 
104
105
106
107
108
109
110
111
112
113
114
115
def validate(self) -> None:
    """Validate document processing job configuration."""
    if not self.model:
        raise ValidationError("Model must be specified")
    if not self.file_tokens:
        raise ValidationError(
            "At least one file token is required. Upload files first using client.upload_file()"
        )
    if self.max_tokens is not None and self.max_tokens < 1:
        raise ValidationError("max_tokens must be positive")
    if not 0.0 <= self.temperature <= 2.0:
        raise ValidationError("Temperature must be between 0.0 and 2.0")

ContainerJob

Run an arbitrary Docker container on the platform, with optional GPU, files, environment variables, and resource limits.

ContainerJob dataclass

Bases: BaseCall

Docker container execution job configuration.

Runs a Docker container on the MicroDC distributed platform. Supports file attachments, GPU access, environment variables, and resource limits.

Workflow
  1. Create ContainerJob with Docker image
  2. Optionally set command, environment, resource limits
  3. Optionally attach files (upload first via client.upload_file())
  4. Submit job using client.send_job()
Example

Run a Python script in a container

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

Attach a file (uploaded separately)

upload_result = client.upload_file("script.py") job.add_file(upload_result['id'])

job_id = client.send_job(job)

Source code in microdc/jobs/container_job.py 
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
@dataclass
class ContainerJob(BaseCall):
    """
    Docker container execution job configuration.

    Runs a Docker container on the MicroDC distributed platform. Supports
    file attachments, GPU access, environment variables, and resource limits.

    Workflow:
        1. Create ContainerJob with Docker image
        2. Optionally set command, environment, resource limits
        3. Optionally attach files (upload first via client.upload_file())
        4. Submit job using client.send_job()

    Example:
        >>> # Run a Python script in a container
        >>> job = ContainerJob(image="python:3.11")
        >>> job.set_command("python /input/script.py")
        >>> job.add_env_var("API_KEY", "sk-...")
        >>>
        >>> # Attach a file (uploaded separately)
        >>> upload_result = client.upload_file("script.py")
        >>> job.add_file(upload_result['id'])
        >>>
        >>> job_id = client.send_job(job)
    """

    image: str = ""
    command: Optional[Union[str, List[str]]] = None
    entrypoint: Optional[str] = None
    environment: Dict[str, str] = field(default_factory=dict)
    gpu: bool = False
    network_mode: str = "none"
    mem_limit: Optional[str] = None
    cpu_count: Optional[int] = None
    file_tokens: List[str] = field(default_factory=list)
    min_capabilities: Dict[str, Any] = field(default_factory=dict)

    def __post_init__(self) -> None:
        self.type = "container"

    def set_image(self, image: str) -> None:
        """
        Set the Docker image to run.

        Args:
            image: Docker image name and tag (e.g. "python:3.11", "nvidia/cuda:12.0-runtime")
        """
        self.image = image

    def set_command(self, command: Union[str, List[str]]) -> None:
        """
        Set the command to run inside the container.

        Accepts either shell form (a string) or exec form (a list).

        Args:
            command: Shell command string (e.g. "python /input/script.py")
                or exec-form list (e.g. ["python", "/input/script.py"]).
        """
        self.command = command

    def set_entrypoint(self, entrypoint: str) -> None:
        """
        Set the container entrypoint, overriding the image's default.

        Pass an empty string to bypass the image's entrypoint entirely
        (equivalent to ``docker run --entrypoint ""``). Useful for images
        whose default ENTRYPOINT runs init / desktop services that get in
        the way of a headless command (e.g. linuxserver.io images).

        Args:
            entrypoint: Entrypoint binary path, or "" to disable.
        """
        self.entrypoint = entrypoint

    def set_environment(self, env: Dict[str, str]) -> None:
        """
        Replace the environment variables dict.

        Args:
            env: Dictionary of environment variable key-value pairs
        """
        self.environment = env

    def add_env_var(self, key: str, value: str) -> None:
        """
        Add a single environment variable.

        Args:
            key: Environment variable name
            value: Environment variable value
        """
        self.environment[key] = value

    def enable_gpu(self) -> None:
        """Enable GPU access for the container."""
        self.gpu = True

    def set_memory_limit(self, limit: str) -> None:
        """
        Set the container memory limit.

        Args:
            limit: Memory limit string (e.g. "512m", "4g")
        """
        self.mem_limit = limit

    def set_cpu_count(self, count: int) -> None:
        """
        Set the number of CPUs available to the container.

        Args:
            count: Number of CPUs
        """
        self.cpu_count = count

    def add_file(self, file_token: str) -> None:
        """
        Add a file token from a previous upload.

        Files must be uploaded using client.upload_file() before
        adding their tokens to the job.

        Args:
            file_token: Token from client.upload_file() response
        """
        self.file_tokens.append(file_token)

    def add_files(self, file_tokens: List[str]) -> None:
        """
        Add multiple file tokens from previous uploads.

        Args:
            file_tokens: List of tokens from client.upload_file() responses
        """
        self.file_tokens.extend(file_tokens)

    def set_min_capabilities(self, capabilities: Dict[str, Any]) -> None:
        """
        Replace the minimum-capability scheduling hints for this job.

        These are sent at the top level of the API payload as
        ``min_capabilities`` and used by the platform to route the job to a
        worker that satisfies them (e.g. a GPU with enough VRAM, a specific
        GPU model, a memory floor, etc.).

        Args:
            capabilities: Dict of capability requirements, e.g.
                ``{"gpu_vram_gb": 32, "gpu_model": "5090", "memory_gb": 64}``.
        """
        self.min_capabilities = capabilities

    def add_min_capability(self, key: str, value: Any) -> None:
        """
        Add or update a single minimum-capability requirement.

        Args:
            key: Capability name (e.g. ``"gpu_vram_gb"``).
            value: Required value.
        """
        self.min_capabilities[key] = value

    def to_api_payload(self) -> Dict[str, Any]:
        """
        Convert to API request format.

        The model field is set to the Docker image name for searchability.
        Container config goes in the "payload" dict, matching other job types.
        """
        # Build inner payload with container config
        inner_payload: Dict[str, Any] = {
            "image": self.image,
        }

        if self.command is not None:
            inner_payload["command"] = self.command
        if self.entrypoint is not None:
            # Empty string is meaningful (== "no entrypoint"), so we
            # include it whenever the user has explicitly set the field.
            inner_payload["entrypoint"] = self.entrypoint
        if self.environment:
            inner_payload["environment"] = self.environment
        if self.gpu:
            inner_payload["gpu"] = self.gpu
        if self.network_mode != "none":
            inner_payload["network_mode"] = self.network_mode
        if self.mem_limit is not None:
            inner_payload["mem_limit"] = self.mem_limit
        if self.cpu_count is not None:
            inner_payload["cpu_count"] = self.cpu_count

        # Build outer payload matching other job types
        payload: Dict[str, Any] = {
            "type": self.type,
            "model": self.image,
            "payload": inner_payload,
            "priority": normalize_priority(self.priority),
            "estimated_cost": 0.0,
        }

        if self.file_tokens:
            payload["file_ids"] = self.file_tokens
        if self.timeout is not None:
            payload["timeout"] = self.timeout
        if self.callback_url is not None:
            payload["callback_url"] = self.callback_url
        if self.metadata:
            payload["metadata"] = self.metadata
        if self.min_capabilities:
            payload["min_capabilities"] = self.min_capabilities

        return payload

    def validate(self) -> None:
        """Validate container job configuration."""
        if not self.image:
            raise ValidationError("Image must be specified")
        if self.cpu_count is not None and self.cpu_count < 1:
            raise ValidationError("cpu_count must be positive")
        if self.mem_limit is not None:
            import re

            if not re.match(r"^\d+[mgMG]$", self.mem_limit):
                raise ValidationError(
                    "mem_limit must be a number followed by 'm' or 'g' (e.g. '512m', '4g')"
                )

set_image(image)

Set the Docker image to run.

Parameters:

Name Type Description Default
image str

Docker image name and tag (e.g. "python:3.11", "nvidia/cuda:12.0-runtime")

 required
Source code in microdc/jobs/container_job.py 
51
52
53
54
55
56
57
58
def set_image(self, image: str) -> None:
    """
    Set the Docker image to run.

    Args:
        image: Docker image name and tag (e.g. "python:3.11", "nvidia/cuda:12.0-runtime")
    """
    self.image = image

set_command(command)

Set the command to run inside the container.

Accepts either shell form (a string) or exec form (a list).

Parameters:

Name Type Description Default
command Union[str, List[str]]

Shell command string (e.g. "python /input/script.py") or exec-form list (e.g. ["python", "/input/script.py"]).

 required
Source code in microdc/jobs/container_job.py 
60
61
62
63
64
65
66
67
68
69
70
def set_command(self, command: Union[str, List[str]]) -> None:
    """
    Set the command to run inside the container.

    Accepts either shell form (a string) or exec form (a list).

    Args:
        command: Shell command string (e.g. "python /input/script.py")
            or exec-form list (e.g. ["python", "/input/script.py"]).
    """
    self.command = command

set_entrypoint(entrypoint)

Set the container entrypoint, overriding the image's default.

Pass an empty string to bypass the image's entrypoint entirely (equivalent to docker run --entrypoint ""). Useful for images whose default ENTRYPOINT runs init / desktop services that get in the way of a headless command (e.g. linuxserver.io images).

Parameters:

Name Type Description Default
entrypoint str

Entrypoint binary path, or "" to disable.

 required
Source code in microdc/jobs/container_job.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
def set_entrypoint(self, entrypoint: str) -> None:
    """
    Set the container entrypoint, overriding the image's default.

    Pass an empty string to bypass the image's entrypoint entirely
    (equivalent to ``docker run --entrypoint ""``). Useful for images
    whose default ENTRYPOINT runs init / desktop services that get in
    the way of a headless command (e.g. linuxserver.io images).

    Args:
        entrypoint: Entrypoint binary path, or "" to disable.
    """
    self.entrypoint = entrypoint

set_environment(env)

Replace the environment variables dict.

Parameters:

Name Type Description Default
env Dict[str, str]

Dictionary of environment variable key-value pairs

 required
Source code in microdc/jobs/container_job.py 
86
87
88
89
90
91
92
93
def set_environment(self, env: Dict[str, str]) -> None:
    """
    Replace the environment variables dict.

    Args:
        env: Dictionary of environment variable key-value pairs
    """
    self.environment = env

add_env_var(key, value)

Add a single environment variable.

Parameters:

Name Type Description Default
key str

Environment variable name

 required
value str

Environment variable value

 required
Source code in microdc/jobs/container_job.py 
 95
 96
 97
 98
 99
100
101
102
103
def add_env_var(self, key: str, value: str) -> None:
    """
    Add a single environment variable.

    Args:
        key: Environment variable name
        value: Environment variable value
    """
    self.environment[key] = value

enable_gpu()

Enable GPU access for the container.

Source code in microdc/jobs/container_job.py 
105
106
107
def enable_gpu(self) -> None:
    """Enable GPU access for the container."""
    self.gpu = True

set_memory_limit(limit)

Set the container memory limit.

Parameters:

Name Type Description Default
limit str

Memory limit string (e.g. "512m", "4g")

 required
Source code in microdc/jobs/container_job.py 
109
110
111
112
113
114
115
116
def set_memory_limit(self, limit: str) -> None:
    """
    Set the container memory limit.

    Args:
        limit: Memory limit string (e.g. "512m", "4g")
    """
    self.mem_limit = limit

set_cpu_count(count)

Set the number of CPUs available to the container.

Parameters:

Name Type Description Default
count int

Number of CPUs

 required
Source code in microdc/jobs/container_job.py 
118
119
120
121
122
123
124
125
def set_cpu_count(self, count: int) -> None:
    """
    Set the number of CPUs available to the container.

    Args:
        count: Number of CPUs
    """
    self.cpu_count = count

add_file(file_token)

Add a file token from a previous upload.

Files must be uploaded using client.upload_file() before adding their tokens to the job.

Parameters:

Name Type Description Default
file_token str

Token from client.upload_file() response

 required
Source code in microdc/jobs/container_job.py 
127
128
129
130
131
132
133
134
135
136
137
def add_file(self, file_token: str) -> None:
    """
    Add a file token from a previous upload.

    Files must be uploaded using client.upload_file() before
    adding their tokens to the job.

    Args:
        file_token: Token from client.upload_file() response
    """
    self.file_tokens.append(file_token)

add_files(file_tokens)

Add multiple file tokens from previous uploads.

Parameters:

Name Type Description Default
file_tokens List[str]

List of tokens from client.upload_file() responses

 required
Source code in microdc/jobs/container_job.py 
139
140
141
142
143
144
145
146
def add_files(self, file_tokens: List[str]) -> None:
    """
    Add multiple file tokens from previous uploads.

    Args:
        file_tokens: List of tokens from client.upload_file() responses
    """
    self.file_tokens.extend(file_tokens)

set_min_capabilities(capabilities)

Replace the minimum-capability scheduling hints for this job.

These are sent at the top level of the API payload as min_capabilities and used by the platform to route the job to a worker that satisfies them (e.g. a GPU with enough VRAM, a specific GPU model, a memory floor, etc.).

Parameters:

Name Type Description Default
capabilities Dict[str, Any]

Dict of capability requirements, e.g. {"gpu_vram_gb": 32, "gpu_model": "5090", "memory_gb": 64}.

 required
Source code in microdc/jobs/container_job.py 
148
149
150
151
152
153
154
155
156
157
158
159
160
161
def set_min_capabilities(self, capabilities: Dict[str, Any]) -> None:
    """
    Replace the minimum-capability scheduling hints for this job.

    These are sent at the top level of the API payload as
    ``min_capabilities`` and used by the platform to route the job to a
    worker that satisfies them (e.g. a GPU with enough VRAM, a specific
    GPU model, a memory floor, etc.).

    Args:
        capabilities: Dict of capability requirements, e.g.
            ``{"gpu_vram_gb": 32, "gpu_model": "5090", "memory_gb": 64}``.
    """
    self.min_capabilities = capabilities

add_min_capability(key, value)

Add or update a single minimum-capability requirement.

Parameters:

Name Type Description Default
key str

Capability name (e.g. "gpu_vram_gb").

 required
value Any

Required value.

 required
Source code in microdc/jobs/container_job.py 
163
164
165
166
167
168
169
170
171
def add_min_capability(self, key: str, value: Any) -> None:
    """
    Add or update a single minimum-capability requirement.

    Args:
        key: Capability name (e.g. ``"gpu_vram_gb"``).
        value: Required value.
    """
    self.min_capabilities[key] = value

to_api_payload()

Convert to API request format.

The model field is set to the Docker image name for searchability. Container config goes in the "payload" dict, matching other job types.

Source code in microdc/jobs/container_job.py 
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
def to_api_payload(self) -> Dict[str, Any]:
    """
    Convert to API request format.

    The model field is set to the Docker image name for searchability.
    Container config goes in the "payload" dict, matching other job types.
    """
    # Build inner payload with container config
    inner_payload: Dict[str, Any] = {
        "image": self.image,
    }

    if self.command is not None:
        inner_payload["command"] = self.command
    if self.entrypoint is not None:
        # Empty string is meaningful (== "no entrypoint"), so we
        # include it whenever the user has explicitly set the field.
        inner_payload["entrypoint"] = self.entrypoint
    if self.environment:
        inner_payload["environment"] = self.environment
    if self.gpu:
        inner_payload["gpu"] = self.gpu
    if self.network_mode != "none":
        inner_payload["network_mode"] = self.network_mode
    if self.mem_limit is not None:
        inner_payload["mem_limit"] = self.mem_limit
    if self.cpu_count is not None:
        inner_payload["cpu_count"] = self.cpu_count

    # Build outer payload matching other job types
    payload: Dict[str, Any] = {
        "type": self.type,
        "model": self.image,
        "payload": inner_payload,
        "priority": normalize_priority(self.priority),
        "estimated_cost": 0.0,
    }

    if self.file_tokens:
        payload["file_ids"] = self.file_tokens
    if self.timeout is not None:
        payload["timeout"] = self.timeout
    if self.callback_url is not None:
        payload["callback_url"] = self.callback_url
    if self.metadata:
        payload["metadata"] = self.metadata
    if self.min_capabilities:
        payload["min_capabilities"] = self.min_capabilities

    return payload

validate()

Validate container job configuration.

Source code in microdc/jobs/container_job.py 
224
225
226
227
228
229
230
231
232
233
234
235
236
def validate(self) -> None:
    """Validate container job configuration."""
    if not self.image:
        raise ValidationError("Image must be specified")
    if self.cpu_count is not None and self.cpu_count < 1:
        raise ValidationError("cpu_count must be positive")
    if self.mem_limit is not None:
        import re

        if not re.match(r"^\d+[mgMG]$", self.mem_limit):
            raise ValidationError(
                "mem_limit must be a number followed by 'm' or 'g' (e.g. '512m', '4g')"
            )

WorkerJob

Run a local project folder as a streaming container. Subclass of ContainerJob that zips a folder containing setup_run.sh, uploads it, and runs it with internet access enabled.

WorkerJob dataclass

Bases: ContainerJob

Run a project folder as a container job on the MicroDC platform.

WorkerJob wraps ContainerJob to provide a simple workflow: point it at a folder containing a setup_run.sh entrypoint and the library handles zipping, uploading, and configuring the container job.

The folder must contain a setup_run.sh at its root. The worker will execute this script inside the container with your files mapped in.

Example folder::

my_project/
├── setup_run.sh       # Required entrypoint
├── main.py
└── requirements.txt

Usage::

>>> from microdc import Client, WorkerJob
>>> client = Client(api_key="mDC_...")
>>> 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 — just pick the right image::

>>> 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")

Attributes:

Name Type Description
folder_path str

Path to the project folder to run.
image str

Docker image (required, e.g. "python:3.11", "node:20", "ubuntu:22.04").
network_mode str

Network mode (default: "bridge" for internet access).
Source code in microdc/jobs/worker_job.py 
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
@dataclass
class WorkerJob(ContainerJob):
    """
    Run a project folder as a container job on the MicroDC platform.

    WorkerJob wraps ContainerJob to provide a simple workflow: point it at a
    folder containing a ``setup_run.sh`` entrypoint and the library handles
    zipping, uploading, and configuring the container job.

    The folder must contain a ``setup_run.sh`` at its root. The worker will
    execute this script inside the container with your files mapped in.

    Example folder::

        my_project/
        ├── setup_run.sh       # Required entrypoint
        ├── main.py
        └── requirements.txt

    Usage::

        >>> from microdc import Client, WorkerJob
        >>> client = Client(api_key="mDC_...")
        >>> 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 — just pick the right image::

        >>> 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")

    Attributes:
        folder_path: Path to the project folder to run.
        image: Docker image (required, e.g. "python:3.11", "node:20", "ubuntu:22.04").
        network_mode: Network mode (default: "bridge" for internet access).
    """

    folder_path: str = ""
    image: str = ""
    network_mode: str = "bridge"

    _zip_path: Optional[str] = field(default=None, init=False, repr=False)

    def __post_init__(self) -> None:
        # WorkerJob's zipped-folder + setup_run.sh flow uses a distinct
        # platform-side type discriminator from a plain ContainerJob.
        self.type = "container_stream"

    def to_api_payload(self) -> Dict[str, Any]:
        # WorkerJob always emits explicit gpu and network_mode so the
        # platform never has to infer them from absence (the parent's
        # builder skips both when they hold their default values).
        payload = super().to_api_payload()
        payload["payload"]["gpu"] = self.gpu
        payload["payload"]["network_mode"] = self.network_mode
        return payload

    def validate(self) -> None:
        """Validate worker job configuration."""
        if not self.folder_path:
            raise ValidationError("folder_path must be specified")

        folder = os.path.abspath(self.folder_path)

        if not os.path.isdir(folder):
            raise ValidationError(f"folder_path is not a directory: {self.folder_path}")

        setup_script = os.path.join(folder, "setup_run.sh")
        if not os.path.isfile(setup_script):
            raise ValidationError(
                f"setup_run.sh not found in {self.folder_path}. "
                "The folder must contain a setup_run.sh script as the entrypoint."
            )

        if not self.image:
            raise ValidationError("Image must be specified")

    def prepare(self) -> str:
        """
        Zip the folder contents into a temporary file.

        Returns:
            Path to the created zip file.
        """
        folder = os.path.abspath(self.folder_path)

        fd, zip_path = tempfile.mkstemp(suffix=".zip", prefix="microdc_worker_")
        os.close(fd)

        with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
            for root, _dirs, files in os.walk(folder):
                for filename in files:
                    file_path = os.path.join(root, filename)
                    arcname = os.path.relpath(file_path, folder)
                    zf.write(file_path, arcname)

        self._zip_path = zip_path
        return zip_path

    def cleanup(self) -> None:
        """Remove the temporary zip file if it exists."""
        if self._zip_path and os.path.exists(self._zip_path):
            os.unlink(self._zip_path)
            self._zip_path = None

validate()

Validate worker job configuration.

Source code in microdc/jobs/worker_job.py 
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
def validate(self) -> None:
    """Validate worker job configuration."""
    if not self.folder_path:
        raise ValidationError("folder_path must be specified")

    folder = os.path.abspath(self.folder_path)

    if not os.path.isdir(folder):
        raise ValidationError(f"folder_path is not a directory: {self.folder_path}")

    setup_script = os.path.join(folder, "setup_run.sh")
    if not os.path.isfile(setup_script):
        raise ValidationError(
            f"setup_run.sh not found in {self.folder_path}. "
            "The folder must contain a setup_run.sh script as the entrypoint."
        )

    if not self.image:
        raise ValidationError("Image must be specified")

prepare()

Zip the folder contents into a temporary file.

Returns:

Type Description
str

Path to the created zip file.
Source code in microdc/jobs/worker_job.py 
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
def prepare(self) -> str:
    """
    Zip the folder contents into a temporary file.

    Returns:
        Path to the created zip file.
    """
    folder = os.path.abspath(self.folder_path)

    fd, zip_path = tempfile.mkstemp(suffix=".zip", prefix="microdc_worker_")
    os.close(fd)

    with zipfile.ZipFile(zip_path, "w", zipfile.ZIP_DEFLATED) as zf:
        for root, _dirs, files in os.walk(folder):
            for filename in files:
                file_path = os.path.join(root, filename)
                arcname = os.path.relpath(file_path, folder)
                zf.write(file_path, arcname)

    self._zip_path = zip_path
    return zip_path

cleanup()

Remove the temporary zip file if it exists.

Source code in microdc/jobs/worker_job.py 
114
115
116
117
118
def cleanup(self) -> None:
    """Remove the temporary zip file if it exists."""
    if self._zip_path and os.path.exists(self._zip_path):
        os.unlink(self._zip_path)
        self._zip_path = None

BaseCall

Abstract base class for all job types.

BaseCall dataclass

Bases: ABC

Base class for all MicroDC job types.

Attributes:

Name Type Description
type str

Job type identifier (set by subclasses)
metadata Dict[str, Any]

User-defined metadata for tracking
priority Union[int, str]

Job priority as int in range 0..100 (higher = scheduled first). Use PRIORITY_LOW (5), PRIORITY_DEFAULT (10), or PRIORITY_HIGH (20). Legacy string aliases ("low", "standard", "high") are accepted and normalized to ints at submission time.
timeout Optional[int]

Maximum execution time in seconds
callback_url Optional[str]

Optional webhook URL for notifications
Source code in microdc/jobs/base.py 
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
@dataclass
class BaseCall(ABC):
    """
    Base class for all MicroDC job types.

    Attributes:
        type: Job type identifier (set by subclasses)
        metadata: User-defined metadata for tracking
        priority: Job priority as int in range 0..100 (higher = scheduled
            first). Use PRIORITY_LOW (5), PRIORITY_DEFAULT (10), or
            PRIORITY_HIGH (20). Legacy string aliases ("low", "standard",
            "high") are accepted and normalized to ints at submission time.
        timeout: Maximum execution time in seconds
        callback_url: Optional webhook URL for notifications
    """

    type: str = field(init=False)
    metadata: Dict[str, Any] = field(default_factory=dict)
    priority: Union[int, str] = PRIORITY_DEFAULT
    timeout: Optional[int] = None
    callback_url: Optional[str] = None

    # Internal tracking (set by client)
    _job_id: Optional[str] = field(default=None, init=False, repr=False)
    _submitted_at: Optional[datetime] = field(default=None, init=False, repr=False)

    @abstractmethod
    def to_api_payload(self) -> Dict[str, Any]:
        """
        Convert job to API request payload.

        Returns:
            Dict containing API-compatible job specification
        """
        pass

    @abstractmethod
    def validate(self) -> None:
        """
        Validate job configuration before submission.

        Raises:
            ValidationError: If configuration is invalid
        """
        pass

to_api_payload() abstractmethod

Convert job to API request payload.

Returns:

Type Description
Dict[str, Any]

Dict containing API-compatible job specification
Source code in microdc/jobs/base.py 
83
84
85
86
87
88
89
90
91
@abstractmethod
def to_api_payload(self) -> Dict[str, Any]:
    """
    Convert job to API request payload.

    Returns:
        Dict containing API-compatible job specification
    """
    pass

validate() abstractmethod

Validate job configuration before submission.

Raises:

Type Description
ValidationError

If configuration is invalid
Source code in microdc/jobs/base.py 
 93
 94
 95
 96
 97
 98
 99
100
101
@abstractmethod
def validate(self) -> None:
    """
    Validate job configuration before submission.

    Raises:
        ValidationError: If configuration is invalid
    """
    pass

JobDetails

Data class containing job status and results.

JobDetails dataclass

Complete job information including status and results. Maps to JobResponse schema from the API.

Attributes:

Name Type Description
job_id str

Unique job identifier (UUID)
type str

Job type (LLM, EMBED, etc.)
status str

Current status (queued, processing, completed, failed, cancelled)
model str

Model used for inference
created_at datetime

Job creation timestamp
started_at Optional[datetime]

Job start timestamp (if started)
completed_at Optional[datetime]

Job completion timestamp (if completed)
estimated_cost Optional[float]

Estimated cost in credits
actual_cost Optional[float]

Actual cost in credits (if completed)
result Optional[Any]

Job results (if completed)
error_message Optional[str]

Error message (if failed)
metadata Optional[Dict[str, Any]]

User-defined metadata
priority int

Job priority as int (0..100, higher = scheduled first)
retry_count int

Number of times job has been retried
user_id Optional[str]

User ID who submitted the job
Source code in microdc/jobs/job_details.py 
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
@dataclass
class JobDetails:
    """
    Complete job information including status and results.
    Maps to JobResponse schema from the API.

    Attributes:
        job_id: Unique job identifier (UUID)
        type: Job type (LLM, EMBED, etc.)
        status: Current status (queued, processing, completed, failed, cancelled)
        model: Model used for inference
        created_at: Job creation timestamp
        started_at: Job start timestamp (if started)
        completed_at: Job completion timestamp (if completed)
        estimated_cost: Estimated cost in credits
        actual_cost: Actual cost in credits (if completed)
        result: Job results (if completed)
        error_message: Error message (if failed)
        metadata: User-defined metadata
        priority: Job priority as int (0..100, higher = scheduled first)
        retry_count: Number of times job has been retried
        user_id: User ID who submitted the job
    """

    job_id: str
    type: str
    status: str
    model: str
    created_at: datetime
    started_at: Optional[datetime] = None
    completed_at: Optional[datetime] = None
    estimated_cost: Optional[float] = None
    actual_cost: Optional[float] = None
    result: Optional[Any] = None
    error_message: Optional[str] = None
    metadata: Optional[Dict[str, Any]] = None
    priority: int = 10
    retry_count: int = 0
    user_id: Optional[str] = None
    encrypted: bool = False

    def is_completed(self) -> bool:
        """Check if job is completed (success or failure)."""
        return self.status.lower() in ("completed", "failed", "cancelled")

    def is_successful(self) -> bool:
        """Check if job completed successfully."""
        return self.status.lower() == "completed"

    def is_failed(self) -> bool:
        """Check if job failed."""
        return self.status.lower() == "failed"

    def duration_seconds(self) -> Optional[float]:
        """Calculate job duration in seconds."""
        if self.started_at and self.completed_at:
            return (self.completed_at - self.started_at).total_seconds()
        return None

is_completed()

Check if job is completed (success or failure).

Source code in microdc/jobs/job_details.py 
49
50
51
def is_completed(self) -> bool:
    """Check if job is completed (success or failure)."""
    return self.status.lower() in ("completed", "failed", "cancelled")

is_successful()

Check if job completed successfully.

Source code in microdc/jobs/job_details.py 
53
54
55
def is_successful(self) -> bool:
    """Check if job completed successfully."""
    return self.status.lower() == "completed"

is_failed()

Check if job failed.

Source code in microdc/jobs/job_details.py 
57
58
59
def is_failed(self) -> bool:
    """Check if job failed."""
    return self.status.lower() == "failed"

duration_seconds()

Calculate job duration in seconds.

Source code in microdc/jobs/job_details.py 
61
62
63
64
65
def duration_seconds(self) -> Optional[float]:
    """Calculate job duration in seconds."""
    if self.started_at and self.completed_at:
        return (self.completed_at - self.started_at).total_seconds()
    return None