plexus.cli.procedure.service module

Procedure Service - Shared service layer for procedure operations.

This service provides a high-level interface for procedure management that can be used by both CLI commands and MCP tools, ensuring DRY principles and consistent behavior across different interfaces.

The service handles: - Procedure creation with proper validation - YAML configuration management - Node and version lifecycle management - Error handling and logging - Account and resource resolution

class plexus.cli.procedure.service.ProcedureCreationResult(procedure: Procedure, root_node: GraphNode, success: bool, message: str)

Bases: object

Result of creating a new procedure.

__init__(procedure: Procedure, root_node: GraphNode, success: bool, message: str) → None
message: str
procedure: Procedure
root_node: GraphNode
success: bool
class plexus.cli.procedure.service.ProcedureInfo(procedure: Procedure, root_node: GraphNode | None, node_count: int, version_count: int, scorecard_name: str | None = None, score_name: str | None = None, latest_version: Any | None = None)

Bases: object

Comprehensive information about an procedure.

__init__(procedure: Procedure, root_node: GraphNode | None, node_count: int, version_count: int, scorecard_name: str | None = None, score_name: str | None = None, latest_version: Any | None = None) → None
latest_version: Any | None = None
node_count: int
procedure: Procedure
root_node: GraphNode | None
score_name: str | None = None
scorecard_name: str | None = None
version_count: int
class plexus.cli.procedure.service.ProcedureService(client: PlexusDashboardClient)

Bases: object

Service for managing experiments with shared logic for CLI and MCP.

__init__(client: PlexusDashboardClient)
create_procedure(account_identifier: str, scorecard_identifier: str | None = None, score_identifier: str | None = None, yaml_config: str | None = None, featured: bool = False, initial_value: Dict[str, Any] | None = None, create_root_node: bool = True, template_id: str | None = None, score_version_id: str | None = None) → ProcedureCreationResult

Create a new procedure with optional root node and initial version.

Args:

account_identifier: Account ID, key, or name scorecard_identifier: Scorecard ID, key, or name score_identifier: Score ID, key, or name yaml_config: YAML configuration (uses default if None) featured: Whether to mark as featured initial_value: Initial computed value (defaults to {“initialized”: True}) create_root_node: Whether to create a root node (defaults to True for backward compatibility) template_id: Optional template/parent procedure ID (NOTE: stored as parentProcedureId in schema) score_version_id: Optional score version ID

Returns:

ProcedureCreationResult with creation details

delete_procedure(procedure_id: str) → Tuple[bool, str]

Delete an procedure and all its nodes/versions.

Args:

procedure_id: ID of the procedure to delete

Returns:

Tuple of (success, message)

get_or_create_default_template(account_id: str) → ProcedureTemplate | None

Get the default procedure template for an account.

NOTE: This no longer creates templates automatically. Users must create their own ProcedureTemplates via the dashboard or API.

Args:

account_id: The account ID

Returns:

The default ProcedureTemplate instance if one exists, None otherwise

get_procedure_info(procedure_id: str) → ProcedureInfo | None

Get comprehensive information about an procedure.

Args:

procedure_id: ID of the experiment

Returns:

ProcedureInfo with full procedure details, or None if not found

get_procedure_yaml(procedure_id: str) → str | None

Get the YAML configuration for a procedure.

Priority order: 1. Procedure.code field (directly stored YAML) 2. Procedure.templateId -> ProcedureTemplate 3. Account default template

Args:

procedure_id: ID of the procedure

Returns:

YAML configuration string, or None if not found

list_procedures(account_identifier: str, scorecard_identifier: str | None = None, limit: int = 100) → List[Procedure]

List procedures for an account, optionally filtered by scorecard.

Args:

account_identifier: Account ID, key, or name scorecard_identifier: Optional scorecard ID, key, or name to filter by limit: Maximum number of experiments to return

Returns:

List of Procedure instances ordered by most recent first

async run_experiment(procedure_id: str, **options) → Dict[str, Any]

Run an procedure with the given ID.

This function executes an procedure with MCP tool support, allowing the experiment to provide AI models with access to Plexus MCP tools during execution.

Args:

procedure_id: ID of the procedure to run **options: Optional parameters for procedure execution:

  • max_iterations: Maximum number of iterations (int)

  • timeout: Timeout in seconds (int)

  • async_mode: Whether to run asynchronously (bool)

  • dry_run: Whether to perform a dry run (bool)

  • restart_from_root_node: Delete all non-root graph nodes before starting (bool)

  • enable_mcp: Whether to enable MCP tools (bool, default True)

  • mcp_tools: List of MCP tool categories to enable (list)

Returns:
Dictionary containing:

  • procedure_id: The procedure ID

  • status: Current status (‘running’, ‘completed’, ‘error’, ‘initiated’)

  • message: Human-readable status message

  • error: Error message if applicable

  • run_id: Unique run identifier (future)

  • progress: Progress information (future)

  • mcp_info: Information about MCP tools if enabled

update_procedure_config(procedure_id: str, yaml_config: str, note: str | None = None) → Tuple[bool, str]

Update a procedure’s configuration by creating a new version.

Args:

procedure_id: ID of the experiment yaml_config: New YAML configuration note: Optional note for the version

Returns:

Tuple of (success, message)