RuntimeContext

RuntimeContext is a context object injected during action execution. It provides access to all dependencies required for action execution, including logging, environment variables, clients, and checkpoints.

Prerequisites

Understand the Plugin System Overview and Defining Actions before using RuntimeContext.

Overview

RuntimeContext is used in both function-based and class-based actions.

Logging: Progress, metrics, and event logging
Environment Variables: Configuration access through PluginEnvironment
Clients: Backend API and agent clients
Checkpoint: Pretrained model information

RuntimeContext Structure

RuntimeContext is a dataclass defined in synapse_sdk/plugins/context/__init__.py.

from dataclasses import dataclass
from typing import Any

from synapse_sdk.plugins.context import PluginEnvironment

@dataclass
class RuntimeContext:
    logger: BaseLogger
    env: PluginEnvironment
    job_id: str | None = None
    client: BackendClient | None = None
    agent_client: AgentClient | None = None
    checkpoint: dict[str, Any] | None = None

Property	Type	Description
`logger`	`BaseLogger`	Logger for progress, metrics, and event logging
`env`	`PluginEnvironment`	Environment variables and configuration access
`job_id`	`str \| None`	Job ID for task tracking
`client`	`BackendClient \| None`	Backend API client
`agent_client`	`AgentClient \| None`	Ray agent client
`checkpoint`	`dict[str, Any] \| None`	Pretrained model information (includes `category`, `path`)

Available Methods

Progress Tracking

Use set_progress() to track operation progress. Specify a step to manage progress individually for multi-phase operations.

def set_progress(self, current: int, total: int, step: str | None = None) -> None:
    """Set progress for the current operation.

    Args:
        current: Current progress value (0 to total).
        total: Total progress value.
        step: Optional step name for multi-phase progress.
    """

Example:

# Single progress
ctx.set_progress(50, 100)

# Step-based progress (multi-phase operations)
ctx.set_progress(10, 100, step='download')
ctx.set_progress(5, 50, step='train')
ctx.set_progress(3, 10, step='export')

Metrics

Use set_metrics() to record training metrics or performance indicators.

def set_metrics(self, value: dict[str, Any], step: str) -> None:
    """Set metrics for a step.

    Args:
        value: Dictionary of metric values.
        step: Non-empty step name.
    """

Example:

# Record training metrics
ctx.set_metrics({
    'loss': 0.1,
    'accuracy': 0.95,
    'learning_rate': 0.001
}, step='training')

# Record inference metrics
ctx.set_metrics({
    'inference_time': 0.05,
    'throughput': 200
}, step='inference')

Logging

Three logging methods are provided.

log()

Structured event logging:

def log(self, event: str, data: dict[str, Any], file: str | None = None) -> None:
    """Log an event with data.

    Args:
        event: Event name/type.
        data: Dictionary of event data.
        file: Optional file path associated with the event.
    """

ctx.log('checkpoint_saved', {'epoch': 5, 'path': '/model.pt'})
ctx.log('data_loaded', {'count': 1000}, file='/data/train.csv')

log_message()

User-facing message logging:

def log_message(self, message: str, context: str = 'info') -> None:
    """Log a user-facing message.

    Args:
        message: Message content.
        context: Message context/level ('info', 'warning', 'danger', 'success').
    """

ctx.log_message('Training started', 'info')
ctx.log_message('Low GPU memory detected', 'warning')
ctx.log_message('Training completed successfully', 'success')
ctx.log_message('Failed to load checkpoint', 'danger')

log_dev_event()

Developer debug event logging (not displayed to end users):

def log_dev_event(self, message: str, data: dict[str, Any] | None = None) -> None:
    """Log a development/debug event.

    Args:
        message: Event message.
        data: Optional additional data.
    """

ctx.log_dev_event('Model architecture initialized', {'layers': 12})
ctx.log_dev_event('Batch processing started')

Environment Variables

Access environment variables through PluginEnvironment. Type-safe methods are provided.

# Basic access
value = ctx.env.get('MY_VAR', 'default')

# Type-specific methods
api_key = ctx.env.get_str('API_KEY')
batch_size = ctx.env.get_int('BATCH_SIZE', default=32)
learning_rate = ctx.env.get_float('LEARNING_RATE', default=0.001)
debug_mode = ctx.env.get_bool('DEBUG', default=False)
gpu_ids = ctx.env.get_list('GPU_IDS', default=['0'])  # Converts comma-separated string to list

PluginEnvironment Methods:

Method	Return Type	Description
`get(key, default)`	`Any`	Returns raw value
`get_str(key, default)`	`str \| None`	Converts to string
`get_int(key, default)`	`int \| None`	Converts to integer
`get_float(key, default)`	`float \| None`	Converts to float
`get_bool(key, default)`	`bool \| None`	Converts to boolean (handles `'true'`, `'1'`, `'yes'`)
`get_list(key, default)`	`list \| None`	Converts comma-separated string to list

Backend Client

Access datasets, models, and more through the backend API client.

Important

ctx.client can be None. Always check before use.

# Check for None before use
if ctx.client:
    dataset = ctx.client.get_data_collection(dataset)
    model_info = ctx.client.get_model(model_id)

# Or use assertion
assert ctx.client is not None, "Backend client is required"
dataset = ctx.client.get_data_collection(dataset)

Checkpoint (Pretrained Model)

Access pretrained model information. The checkpoint is a dictionary containing category and path.

if ctx.checkpoint:
    category = ctx.checkpoint.get('category')  # 'base' or fine-tuned model name
    model_path = ctx.checkpoint.get('path')    # Model file path

    if category == 'base':
        print(f'Using base model from {model_path}')
    else:
        print(f'Using fine-tuned model: {category}')

Action-Specific Contexts

In step-based workflows, use action-specific contexts that inherit from BaseStepContext instead of extending RuntimeContext directly. These contexts access RuntimeContext through the runtime_ctx property.

from dataclasses import dataclass, field
from typing import Any

from synapse_sdk.plugins.steps import BaseStepContext

@dataclass
class MyStepContext(BaseStepContext):
    # BaseStepContext includes runtime_ctx: RuntimeContext
    params: dict[str, Any] = field(default_factory=dict)
    results: list[Any] = field(default_factory=list)

Available Action-Specific Contexts:

Context	Location	Description
`TrainContext`	`synapse_sdk.plugins.actions.train`	Training workflow state (dataset, model_path, etc.)
`InferenceContext`	`synapse_sdk.plugins.actions.inference`	Inference workflow state (requests, results, etc.)
`ExportContext`	`synapse_sdk.plugins.actions.export`	Export workflow state (output_path, etc.)
`UploadContext`	`synapse_sdk.plugins.actions.upload`	Upload workflow state (organized_files, etc.)
`DeploymentContext`	`synapse_sdk.plugins.actions.inference`	Deployment workflow state (serve_app_name, etc.)

Accessing RuntimeContext from Step Context:

from synapse_sdk.plugins.actions.train import TrainContext
from synapse_sdk.plugins.steps import BaseStep, StepResult

# Usage within a Step
class LoadDataStep(BaseStep[TrainContext]):
    def execute(self, context: TrainContext) -> StepResult:
        # Access RuntimeContext methods through runtime_ctx
        context.runtime_ctx.set_progress(0, 100, step='load')
        context.runtime_ctx.log_message('Loading dataset...', 'info')

        # Or use BaseStepContext convenience methods
        context.set_progress(50, 100)  # Step auto-set from current_step
        context.log('data_loaded', {'count': 1000})

        return StepResult(success=True)

Usage in Function-Based Actions

In function-based actions, use the @action decorator and receive RuntimeContext as the second argument.

from pydantic import BaseModel
from synapse_sdk.plugins import action, RuntimeContext

class TrainParams(BaseModel):
    epochs: int = 10
    batch_size: int = 32
    learning_rate: float = 0.001

@action(params=TrainParams, description='Train a model')
def train(params: TrainParams, ctx: RuntimeContext) -> dict:
    ctx.log_message('Training started', 'info')

    for epoch in range(params.epochs):
        ctx.set_progress(epoch + 1, params.epochs, step='train')
        ctx.set_metrics({'epoch': epoch, 'loss': 0.1}, step='training')

    ctx.log_message('Training completed', 'success')
    return {'status': 'completed', 'epochs': params.epochs}

Usage in Class-Based Actions

In class-based actions, inherit from BaseAction and access RuntimeContext through self.ctx. Helper methods are also provided for convenience.

from pydantic import BaseModel
from synapse_sdk.plugins import BaseAction, RuntimeContext

class TrainParams(BaseModel):
    epochs: int = 10

class TrainAction(BaseAction[TrainParams]):
    def execute(self) -> dict:
        # Access RuntimeContext through self.ctx
        self.ctx.log_message('Training started', 'info')

        for epoch in range(self.params.epochs):
            # Use helper methods (equivalent to self.ctx.set_progress)
            self.set_progress(epoch + 1, self.params.epochs, step='train')
            self.set_metrics({'epoch': epoch, 'loss': 0.1}, step='training')

        return {'status': 'completed'}

BaseAction Helper Methods:

Method	Description
`self.log(event, data, file)`	Equivalent to `self.ctx.log()`
`self.set_progress(current, total, step)`	Equivalent to `self.ctx.set_progress()`
`self.set_metrics(value, step)`	Equivalent to `self.ctx.set_metrics()`
`self.logger`	Property to access `self.ctx.logger`

Best Practices

Progress Updates

Update progress in meaningful units. Overly frequent updates can impact performance.

# Good: Update per epoch
for epoch in range(100):
    train_one_epoch()
    ctx.set_progress(epoch + 1, 100, step='train')

# Bad: Update every batch
for batch in dataloader:  # Called thousands of times
    process_batch(batch)
    ctx.set_progress(batch_idx, total_batches)  # Excessive calls

Use Appropriate Log Levels

# info: General progress
ctx.log_message('Starting data preprocessing', 'info')

# warning: Requires attention but can continue
ctx.log_message('GPU memory usage high (90%)', 'warning')

# success: Operation completed
ctx.log_message('Model training completed', 'success')

# danger: Error occurred (unrecoverable)
ctx.log_message('Failed to save checkpoint', 'danger')

Avoid Logging Sensitive Information

Never include sensitive information such as API keys or passwords in logs.

# Bad
ctx.log('config_loaded', {'api_key': api_key})

# Good
ctx.log('config_loaded', {'api_key_set': bool(api_key)})

Check for None

client, agent_client, checkpoint, and similar properties can be None. Always check before use.

# Recommended
if ctx.client:
    data = ctx.client.fetch_data()
else:
    ctx.log_message('Backend client not available', 'warning')
    data = load_local_data()

Overview​

RuntimeContext Structure​

Available Methods​

Progress Tracking​

Metrics​

Logging​

log()​

log_message()​

log_dev_event()​

Environment Variables​

Backend Client​

Checkpoint (Pretrained Model)​

Action-Specific Contexts​

Usage in Function-Based Actions​

Usage in Class-Based Actions​

Best Practices​

Progress Updates​

Use Appropriate Log Levels​

Avoid Logging Sensitive Information​

Check for None​