plexus.cli.procedure.procedure_sop_agent module

Procedure-specific StandardOperatingProcedureAgent implementation.

This module provides the ProcedureSOPAgent, which uses the general-purpose StandardOperatingProcedureAgent to execute procedure hypothesis generation procedures. It implements the experiment-specific behavior while delegating the core SOP orchestration to the reusable base agent.

Architecture: - ProcedureProcedureDefinition: Defines experiment-specific prompts, tools, and logic - ProcedureFlowManager: Manages procedure conversation flow (exploration → synthesis → hypothesis) - ProcedureChatRecorder: Records procedure conversations - ProcedureSOPAgent: Convenience wrapper for procedure procedures

class plexus.cli.procedure.procedure_sop_agent.ProcedureChatRecorderAdapter(client, procedure_id: str, node_id: str | None = None)

Bases: ChatRecorder

Adapter that wraps ProcedureChatRecorder to implement the ChatRecorder interface.

This allows procedure chat recording to work with the general-purpose StandardOperatingProcedureAgent while maintaining experiment-specific recording logic.

__init__(client, procedure_id: str, node_id: str | None = None)

async end_session(status: str, name: str = None) → bool: End the procedure recording session.

async record_message(role: str, content: str, message_type: str, tool_name: str | None = None, tool_parameters: Dict[str, Any] | None = None, tool_response: Dict[str, Any] | None = None, parent_message_id: str | None = None) → str | None: Record a message in the procedure session.

async record_system_message(content: str) → str | None: Record a system message in the procedure session.

async start_session(context: Dict[str, Any]) → str | None: Start an procedure recording session.

class plexus.cli.procedure.procedure_sop_agent.ProcedureProcedureDefinition

Bases: object

Procedure-specific procedure definition for the StandardOperatingProcedureAgent.

This class encapsulates all experiment-specific knowledge: - Procedure hypothesis generation prompts (configurable via YAML) - Procedure tool scoping by phase - Procedure completion criteria - Procedure-specific guidance logic

__init__()

get_allowed_tools() → List[str]: Get all available tools for the worker agent.

get_completion_summary(state_data: Dict[str, Any]) → str: Get procedure completion summary.

get_manager_user_prompt(context: Dict[str, Any], state_data: Dict[str, Any] | None = None) → str: Get the initial manager user prompt from YAML configuration.

get_sop_guidance_prompt(context: Dict[str, Any], state_data: Dict[str, Any]) → str: Get the SOP Agent guidance prompt from YAML configuration.

get_system_prompt(context: Dict[str, Any]) → str: Get the procedure hypothesis generation system prompt from YAML configuration.

get_user_prompt(context: Dict[str, Any]) → str: Get the initial procedure user prompt from YAML configuration.

should_continue(state_data: Dict[str, Any]) → bool

Determine if the procedure procedure should continue.

With the new stop tool approach, the worker agent decides when to stop. This method only enforces safety limits and respects explicit stop requests.

class plexus.cli.procedure.procedure_sop_agent.ProcedureSOPAgent(procedure_id: str, mcp_server, client=None, openai_api_key: str | None = None, experiment_context: Dict[str, Any] | None = None, model_config: Dict[str, Any] | None = None)

Bases: object

Procedure-specific wrapper for StandardOperatingProcedureAgent.

This class provides a convenient interface for running procedure procedures while hiding the general-purpose SOP Agent complexity from experiment-specific code.

It maintains the same interface as the original ProcedureAIRunner for backward compatibility while using the new architecture under the hood.

Initialize the ProcedureSOPAgent with experiment-specific configuration.

__init__(procedure_id: str, mcp_server, client=None, openai_api_key: str | None = None, experiment_context: Dict[str, Any] | None = None, model_config: Dict[str, Any] | None = None): Initialize the ProcedureSOPAgent with experiment-specific configuration.

async execute_sop_guided_procedure() → Dict[str, Any]

Execute the SOP-guided procedure procedure.

This method delegates to the general-purpose SOP Agent while maintaining the experiment-specific interface for backward compatibility.

get_exploration_prompt() → str: Get the exploration prompt (backward compatibility).

get_system_prompt() → str: Get the system prompt (backward compatibility).

get_user_prompt() → str: Get the user prompt (backward compatibility).

async setup(experiment_yaml: str) → bool: Set up the procedure procedure with the given YAML configuration.

async plexus.cli.procedure.procedure_sop_agent.run_sop_guided_procedure(procedure_id: str, experiment_yaml: str, mcp_server, openai_api_key: str | None = None, experiment_context: Dict[str, Any] | None = None, client=None, model_config: Dict[str, Any] | None = None) → Dict[str, Any]

Run SOP-guided procedure using the ProcedureSOPAgent.

This function maintains backward compatibility with the original experiment execution interface while using the new general-purpose SOP Agent architecture.

Args:: procedure_id: The procedure ID experiment_yaml: The procedure YAML configuration mcp_server: The MCP server instance providing tools openai_api_key: Optional OpenAI API key for SOP Agent experiment_context: Optional procedure context with known information client: Optional GraphQL client for recording conversations model_config: Optional model configuration dict (e.g., {“model”: “gpt-5”, “reasoning_effort”: “medium”})
Returns:: Dictionary with execution results including nodes created, tools used, etc.