PyroMind SDK

Lightweight Python SDK for the PyroMind AI Platform API — manage training workflows, Jupyter instances, inference jobs, EchoMind and more.

Installation

pip install pyromind-sdk

Requires Python >= 3.8.

Quick Start

from pyromind_sdk import PyroMindAPIClient
from pyromind_sdk.client.models import TrainingTaskCreateRequest

client = PyroMindAPIClient(api_key="your-api-key")

# Create and run a studio task
task = client.studio.create(
    TrainingTaskCreateRequest(
        name="my-workflow",
        workflow={"nodes": [...]}
    )
)
print(f"Created task: {task.task_id}")

Configuration

Client parameters

Param	Required	Type	Default	Description
`api_key`	Yes*	`str`	`PYROMIND_API_KEY` env	Bearer token for API auth
`cluster`	No	`str`	`PYROMIND_CLUSTER` env or `"us-west-2"`	Target cluster (`X-Cluster` header)
`timeout`	No	`int`	`30`	Request timeout in seconds
`max_retries`	No	`int`	`3`	Max retries for failed requests

* api_key can be provided as a parameter or via PYROMIND_API_KEY environment variable.

Environment variables

Variable	Required	Default	Description
`PYROMIND_API_KEY`	Yes	—	API bearer token
`PYROMIND_CLUSTER`	No	`us-west-2`	Target cluster identifier
`PYROMIND_STORAGE_ENDPOINT`	No	`https://storage.pyromind.ai`	Storage endpoint URL
`PYROMIND_STORAGE_SECRET_KEY`	No	—	Storage secret key
`PYROMIND_STORAGE_BUCKET`	No	—	Default storage bucket name

Project Structure

pyromind_sdk/
├── client/                          # API clients
│   ├── base.py                      # Base HTTP client
│   ├── client.py                    # PyroMindAPIClient (unified entry)
│   ├── async_client.py              # PyroMindAsyncAPIClient (async entry)
│   ├── studio.py / async_studio.py  # Studio / Training tasks
│   ├── jupyterLab.py / async_jupyterlab.py  # Jupyter instances
│   ├── inference.py / async_inference.py    # Inference jobs
│   ├── echomind.py / async_echomind.py      # EchoMind instances
│   ├── storage.py                   # File storage
│   ├── profile.py                   # User profile & SSH keys
│   ├── models.py                    # Pydantic models
│   └── workflow/                    # Workflow validation & conversion
├── nodes/                           # Custom node SDK
│   ├── function_call_wrapper.py     # Python function → node
│   ├── python_function_executor.py  # Python node executor
│   ├── python_to_yaml.py            # Convert Python to YAML
│   └── yaml_loader.py               # YAML node loader
├── common/                          # Shared utilities
│   ├── constants.py
│   └── node_sdk.py
├── cli.py                           # CLI entry points
├── python_function_to_yaml_cli.py   # Python → YAML CLI tool
├── examples/                        # Usage examples
│   └── openapi/                     # API usage examples
└── tests/                           # Test suite

Services

Studio (`client.studio`)

Training workflow management — create, monitor, and manage workflow tasks.

Method	Input	Output	Description
`list()`	—	`List[TrainingTaskResponse]`	List all studio tasks
`create(request)`	`TrainingTaskCreateRequest`	`TrainingTaskCreateResponse`	Create a new training task
`get_job(task_id)` / `get_task(task_id)`	`str`	`TrainingTaskResponse`	Get task details
`delete(task_id, force=False)`	`str`, `bool`	`None`	Delete a task
`stop(task_id)`	`str`	`TrainingTaskResponse`	Stop a running task
`get_node_output(task_id, node_id)`	`str`, `str`	`Optional[Dict]`	Get node-level output
`get_node_info(names=None)`	`Optional[str]`	`Dict[str, Any]`	Get node definition info
`reload_nodes(node_name=None)`	`Optional[str]`	`Dict[str, Any]`	Reload node YAML definitions
`create_node(...)`	`yaml_path/yaml_content` + opts	`Dict[str, Any]`	Register a custom node
`delete_node_by_name(node_name)`	`str`	`Dict[str, Any]`	Delete a custom node
`move_node(node_name, source_file_path)`	`str`, `str`	`Dict[str, Any]`	Move node source
`run_with_params(request)`	`WorkflowRunRequest`	`TrainingTaskCreateResponse`	Run stored workflow with params
`export_node_outputs(task_id, nodes_info, ...)`	`str`, `List`, `Optional[List]`	`List[Dict]`	Export all node outputs
`wait_for_task_completion(task_id, ...)`	`str` + opts	`str` (status)	Poll until terminal status
`create_and_wait(request, ...)`	`TrainingTaskCreateRequest` + opts	`Dict[str, Any]`	Create + poll + optionally export outputs

TrainingTaskCreateRequest parameters:

Param	Required	Type	Description
`name`	Yes	`str`	Task name
`workflow`	Yes	`Dict[str, Any]`	Workflow JSON structure with node definitions

WorkflowRunRequest parameters:

Param	Required	Type	Description
`workflow_name`	Yes	`str`	Name of the stored workflow
`primitive_node_map`	No	`Dict[str, Any]`	Injected primitive node values (default: `{}`)

Example:

from pyromind_sdk.client.models import TrainingTaskCreateRequest, WorkflowRunRequest

# Create a training task
task = client.studio.create(
    TrainingTaskCreateRequest(
        name="my-workflow",
        workflow={"nodes": [...]}
    )
)
print(f"Task ID: {task.task_id}")

# List tasks
tasks = client.studio.list()

# Run workflow with params
result = client.studio.run_with_params(
    WorkflowRunRequest(workflow_name="my-workflow", primitive_node_map={"key": "value"})
)

# Wait for completion
status = client.studio.wait_for_task_completion(task.task_id, timeout=600)
print(f"Final status: {status}")

Jupyter (`client.jupyter`)

Jupyter instance management.

Method	Input	Output	Description
`list()`	—	`List[JupyterResponse]`	List all Jupyter instances
`create(request)`	`JupyterRequest`	`JupyterResponse`	Create new instance
`get_instance(jupyter_id)`	`str`	`JupyterResponse`	Get instance details
`update(jupyter_id, request)`	`str`, `JupyterRequest`	`JupyterResponse`	Update instance config
`delete(jupyter_id)`	`str`	`None`	Delete an instance
`pause(jupyter_id)` / `resume(jupyter_id)`	`str`	`JupyterResponse`	Pause/resume

JupyterRequest parameters:

Param	Required	Type	Description
`name`	No	`str`	Instance display name
`resources`	No	`ResourceConfig`	CPU/memory/gpu config

Example:

from pyromind_sdk.client.models import JupyterRequest, ResourceConfig

# Create Jupyter instance
jupyter = client.jupyter.create(
    JupyterRequest(
        name="my-notebook",
        resources=ResourceConfig(cpu="4", memory="16Gi", gpu="1")
    )
)
print(f"Jupyter ID: {jupyter.id}, URL: {jupyter.url}")

Inference (`client.inference`)

Inference job management.

Method	Input	Output	Description
`list()`	—	`List[InferenceJobResponse]`	List all inference jobs
`create(request)`	`InferenceJobRequest`	`str` (job_id)	Create inference job
`get_job(job_id)`	`str`	`InferenceJobResponse`	Get job details
`update(job_id, request)`	`str`, `InferenceJobRequest`	`InferenceJobResponse`	Update job config
`delete(job_id)`	`str`	`None`	Delete a job
`pause(job_id)` / `resume(job_id)`	`str`	`InferenceJobResponse`	Pause/resume job
`get_framework()`	—	`List[str]`	List available frameworks
`get_inf_image(framework)`	`str`	`List[str]`	List inference images

InferenceJobRequest parameters:

Param	Required	Type	Description
`model_path`	Yes	`str`	Path to the model
`inference_framework`	No	`str`	Framework name (get via `get_framework()`)
`resources`	No	`ResourceConfig`	CPU/memory/gpu config
`name`	No	`str`	Job display name
`inf_image`	No	`str`	Inference image (get via `get_inf_image()`)
`model_name`	No	`str`	Model name override
`model_length`	No	`int`	Model context length
`startup_args`	No	`List[dict]` or `List[str]`	Custom inference server startup args. Prefer `[{"--arg": value}]`; include the leading `-` or `--` yourself. Duplicate default options are overridden by user args

Example:

from pyromind_sdk.client.models import InferenceJobRequest, ResourceConfig

# List available frameworks and images
frameworks = client.inference.get_framework()
images = client.inference.get_inf_image(frameworks[0])

# Create inference job
job_id = client.inference.create(
    InferenceJobRequest(
        model_path="/path/to/model",
        inference_framework=frameworks[0],
        resources=ResourceConfig(cpu="8", memory="32Gi", gpu="1", gpu_card="H100"),
        startup_args=[{"--trust-remote-code": None}],
        name="my-inference"
    )
)
print(f"Job ID: {job_id}")

# Get job details
job = client.inference.get_job(job_id)
print(f"Status: {job.status}")

EchoMind (`client.echomind`)

EchoMind instance lifecycle management.

Method	Input	Output	Description
`list()`	—	`List[EchoMindJobResponse]`	List all EchoMind instances
`create(request)`	`EchoMindJobRequest`	`str` (job_id)	Create EchoMind instance
`get_job(job_id)`	`str`	`EchoMindJobResponse`	Get instance details
`update(job_id, request)`	`str`, `EchoMindJobRequest`	`EchoMindJobResponse`	Update instance config
`delete(job_id)`	`str`	`None`	Delete an instance
`pause(job_id)` / `resume(job_id)`	`str`	`EchoMindJobResponse`	Pause/resume

EchoMindJobRequest parameters:

Param	Required	Type	Description
`name`	No	`str`	Instance display name
`resources`	No	`ResourceConfig`	CPU/memory/gpu config

Example:

from pyromind_sdk.client.models import EchoMindJobRequest, ResourceConfig

# Create EchoMind instance
job_id = client.echomind.create(
    EchoMindJobRequest(
        name="my-echomind",
        resources=ResourceConfig(cpu="4", memory="16Gi")
    )
)
print(f"EchoMind ID: {job_id}")

# List instances
instances = client.echomind.list()

# Cleanup
client.echomind.delete(job_id)

Storage (`client.storage`)

MinIO/S3-compatible file storage. Requires minio package (pip install minio).

Method	Input	Output	Description
`list_files(folder_path, ...)`	`str` + opts	`List[Dict]`	List files in a directory
`file_exists(file_path)`	`str`	`bool`	Check file existence
`upload_file(file_path, object_name, ...)`	`str/Path/BinaryIO` + opts	`Dict[str, Any]`	Upload file (multipart support)
`upload_folder(folder_path, ...)`	`str/Path` + opts	`List[Dict]`	Upload entire folder
`download_file(object_name, ...)`	`str` + opts	`Union[bytes, Path]`	Download file
`download_folder(folder_path, local_path)`	`str`, `str/Path` + opts	`List[Dict]`	Download folder
`delete_file(object_name)`	`str`	`None`	Delete a file
`delete_folder(folder_path)`	`str` + opts	`Dict`	Delete a folder

Storage init parameters:

Param	Required	Type	Description
`endpoint`	No	`str`	Storage endpoint (env: `PYROMIND_STORAGE_ENDPOINT`, default: `https://storage.pyromind.ai`)
`access_key`	No	`str`	Access key (env: `PYROMIND_API_KEY`)
`secret_key`	No	`str`	Secret key (env: `PYROMIND_STORAGE_SECRET_KEY`)
`bucket_name`	No	`str`	Default bucket (env: `PYROMIND_STORAGE_BUCKET`)
`secure`	No	`bool`	Use HTTPS (auto-detected from endpoint URL)
`region`	No	`str`	Storage region (default: `us-east-1`)

Example:

from pyromind_sdk.client.storage import StorageClient

storage = StorageClient()

# List files
files = storage.list_files(folder_path="documents/")
for f in files:
    print(f"{f['object_name']} ({f['size']} bytes)")

# Upload file
storage.upload_file("local/file.txt", "remote/file.txt")

# Download file
storage.download_file("remote/file.txt", "downloaded/file.txt")

# Check existence
if storage.file_exists("remote/file.txt"):
    print("File exists")

Profile (`client.profile`)

User profile and SSH keys.

Method	Input	Output	Description
`get_user_info(credit_info=False)`	`bool`	`ProfileUserInfoResponse`	Get user info
`get_access_key()`	—	`str`	Get access key
`get_storage_info()`	—	`ProfileStorageInfoResponse`	Get storage credentials
`add_key(request)`	`UserPubKeyRequest`	`bool`	Add SSH public key
`list_keys()`	—	`List[UserPubKey]`	List SSH public keys

Example:

# Get user info
user = client.profile.get_user_info()
print(f"User: {user.username}")

# Get storage info
storage_info = client.profile.get_storage_info()
print(f"Used: {storage_info.human_used_size} / Total: {storage_info.human_total_size}")

# SSH key management
from pyromind_sdk.client.models import UserPubKeyRequest

client.profile.add_key(UserPubKeyRequest(key="ssh-ed25519 AAAA..."))
keys = client.profile.list_keys()

Async Support

All services have async counterparts via PyroMindAsyncAPIClient:

from pyromind_sdk import PyroMindAsyncAPIClient

async with PyroMindAsyncAPIClient(api_key="your-api-key") as client:
    tasks = await client.studio.list()
    task = await client.studio.create(request)

Async clients (same method set as sync):

client.studio → AsyncStudioClient
client.instances → AsyncJupyterLabClient
client.inference → AsyncInferenceClient
client.echomind → AsyncEchoMindClient

Error Handling

All API calls raise PyroMindAPIError (sync) or PyroMindAsyncAPIError (async) on failure:

from pyromind_sdk.client.base import PyroMindAPIError

try:
    task = client.studio.get_task("invalid-id")
except PyroMindAPIError as e:
    print(f"Error {e.status_code}: {e.message}")
    if e.response:
        print(f"Response: {e.response}")

Attribute	Type	Description
`message`	`str`	Error description
`status_code`	`Optional[int]`	HTTP status code
`response`	`Optional[Dict]`	API error response body

Key Response Models

Each service returns structured Pydantic model objects. Key fields:

`TrainingTaskResponse` (Studio)

Field	Type	Description
`task_id`	`str`	Task unique ID
`name`	`str`	Task name
`status`	`str`	Current status (`running`, `completed`, `failed`, etc.)
`workflow`	`Dict`	Workflow configuration
`nodes`	`List[TrainingTaskNodeInfo]`	Node execution details
`error_message`	`Optional[str]`	Error info if failed
`created_at`	`datetime`	Creation timestamp

`JupyterResponse` (Jupyter)

Field	Type	Description
`id`	`str`	Instance ID
`name`	`str`	Instance name
`status`	`str`	Current status
`url`	`Optional[str]`	Jupyter URL
`password`	`Optional[str]`	Access password

`InferenceJobResponse` (Inference)

Field	Type	Description
`id`	`str`	Job ID
`name`	`str`	Job name
`model_path`	`str`	Model path
`status`	`str`	Current status
`endpoint_url`	`Optional[str]`	Inference endpoint
`resources`	`Optional[ResourceConfig]`	Allocated resources

`EchoMindJobResponse` (EchoMind)

Field	Type	Description
`id`	`str`	Instance ID
`name`	`str`	Instance name
`status`	`str`	Current status

Workflow Validation & Conversion

The client/workflow/ module provides workflow validation and format conversion:

from pyromind_sdk.client import validate_workflow, ValidationError

# Validate a workflow structure
try:
    validate_workflow(workflow_dict)
    print("Workflow is valid")
except ValidationError as e:
    print(f"Invalid workflow: {e}")

Tool	Description
`validate_workflow(workflow)`	Validate workflow JSON structure
`ValidationError`	Raised on invalid workflow
`converter.py`	Convert between workflow formats

CLI Tools

Command	Description
`python -m pyromind_sdk.cli`	SDK CLI (various utilities)
`python -m pyromind_sdk.python_function_to_yaml_cli`	Convert Python function → YAML node definition

Custom Node SDK

Beyond YAML definitions, the SDK provides programmatic node creation tools:

Wrap a Python function as a custom node:

from pyromind_sdk.nodes.function_call_wrapper import create_node_from_function

# Decorate any function to become a node definition
@create_node_from_function(
    name="my_custom_node",
    description="Processes input data",
    category="data-processing"
)
def process_data(input_text: str, threshold: float = 0.5) -> dict:
    # Your logic here
    return {"result": "processed", "value": len(input_text)}

Execute Python functions as nodes at runtime:

from pyromind_sdk.nodes.python_function_executor import execute_python_node

result = execute_python_node(
    source_code="print('hello')",
    node_type="python"
)

Convert Python functions to YAML config:

from pyromind_sdk.nodes.python_to_yaml import python_function_to_yaml_config

def my_func(input: str) -> str:
    return input.upper()

yaml_config = python_function_to_yaml_config(my_func)
# yaml_config can be saved to a .yaml file and registered via studio.create_node()

Validate and load YAML node definitions:

from pyromind_sdk.nodes.yaml_loader import load_yaml_node
from pyromind_sdk.nodes.node_validator import validate_node_config

node_config = load_yaml_node("path/to/node.yaml")
validate_node_config(node_config)

Testing

pytest

Examples

Example	Description
`api_client_basic.py`	Basic client setup
`studio_example.py`	Studio task CRUD + node output
`studio_monitor.py`	Monitor task status in a loop
`workflow_cli.py`	CLI tool for workflow management
`complete_workflow_example.py`	End-to-end workflow demo
`jupyter_instance_example.py`	Jupyter instance CRUD
`inference_example.py`	Inference job management
`echomind_example.py`	EchoMind lifecycle
`storage_example.py`	File upload/download
`release_all_instance.py`	Bulk release resources
`async_training_example.py`	Async studio training
`async_inference_example.py`	Async inference
`async_echomind_example.py`	Async EchoMind
`async_jupyter_instance_example.py`	Async Jupyter

Development

Install from source

git clone https://github.com/pyromind/pyromind-sdk.git
cd pyromind-sdk
pip install -e .

Name		Name	Last commit message	Last commit date
Latest commit History 287 Commits
pyromind_sdk		pyromind_sdk
script		script
skill		skill
.gitignore		.gitignore
LICENSE		LICENSE
MANIFEST.in		MANIFEST.in
MIGRATION_SUMMARY.md		MIGRATION_SUMMARY.md
README.md		README.md
README.zh.md		README.zh.md
auto_layout_feature_summary.md		auto_layout_feature_summary.md
build_and_publish.sh		build_and_publish.sh
cli_auto_layout_quickref.md		cli_auto_layout_quickref.md
cli_auto_layout_usage.md		cli_auto_layout_usage.md
complete_work_summary.md		complete_work_summary.md
create_release.sh		create_release.sh
enhanced_validator_summary.md		enhanced_validator_summary.md
pyproject.toml		pyproject.toml
pyproject_fix_summary.md		pyproject_fix_summary.md
setup.py		setup.py
update_version.py		update_version.py

Navigation Menu

Uh oh!

Repository files navigation

PyroMind SDK

Installation

Quick Start

Configuration

Client parameters

Environment variables

Project Structure

Services

Studio (`client.studio`)

Jupyter (`client.jupyter`)

Inference (`client.inference`)

EchoMind (`client.echomind`)

Storage (`client.storage`)

Profile (`client.profile`)

Async Support

Error Handling

Key Response Models

`TrainingTaskResponse` (Studio)

`JupyterResponse` (Jupyter)

`InferenceJobResponse` (Inference)

`EchoMindJobResponse` (EchoMind)

Workflow Validation & Conversion

CLI Tools

Custom Node SDK

Testing

Examples

Development

Install from source

About

Releases

Packages

Contributors

Languages

Uh oh!

Folders and files

Latest commit

History

Repository files navigation

PyroMind SDK

Installation

Quick Start

Configuration

Client parameters

Environment variables

Project Structure

Services

Studio (client.studio)

Jupyter (client.jupyter)

Inference (client.inference)

EchoMind (client.echomind)

Storage (client.storage)

Profile (client.profile)

Async Support

Error Handling

Key Response Models

TrainingTaskResponse (Studio)

JupyterResponse (Jupyter)

InferenceJobResponse (Inference)

EchoMindJobResponse (EchoMind)

Workflow Validation & Conversion

CLI Tools

Custom Node SDK

Testing

Examples

Development

Install from source

About

Resources

Stars

Watchers

Forks

Releases

Packages

Contributors

Languages

Studio (`client.studio`)

Jupyter (`client.jupyter`)

Inference (`client.inference`)

EchoMind (`client.echomind`)

Storage (`client.storage`)

Profile (`client.profile`)

`TrainingTaskResponse` (Studio)

`JupyterResponse` (Jupyter)

`InferenceJobResponse` (Inference)

`EchoMindJobResponse` (EchoMind)