plexus.cli.procedure.sop_agent_base module

Base StandardOperatingProcedureAgent - General-purpose multi-agent orchestration.

This module provides the core StandardOperatingProcedureAgent that can be used across different domains. It’s decoupled from experiment-specific logic and can orchestrate any procedure-driven multi-agent workflow.

Key Design Principles: - Domain-agnostic: Works for experiments, reports, analysis, etc. - Configurable: Procedures defined through pluggable components - Reusable: Same agent can handle different types of structured workflows - Extensible: Easy to add new domains without changing core logic

class plexus.cli.procedure.sop_agent_base.ChatRecorder

Bases: ABC

Abstract chat recorder for procedure-specific conversation recording.

Different procedures may record conversations differently (e.g., procedure sessions, report generation logs, etc.)

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

abstractmethod async record_message(role: str, content: str, message_type: str) → str | None: Record a message in the current session.

abstractmethod async record_system_message(content: str) → str | None: Record a system message.

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

class plexus.cli.procedure.sop_agent_base.FlowManager

Bases: ABC

Abstract flow manager for procedure-specific conversation flow.

Different procedures can implement their own flow logic (e.g., procedure phases, report generation steps, etc.)

abstractmethod get_completion_summary() → str: Get summary when flow is complete.

abstractmethod get_next_guidance() → str | None: Get template guidance for the next step.

abstractmethod should_continue() → bool: Determine if the flow should continue.

abstractmethod update_state(tools_used: List[str], response_content: str, **kwargs) → Dict[str, Any]: Update the flow state based on coding assistant actions.

class plexus.cli.procedure.sop_agent_base.ProcedureDefinition(*args, **kwargs)

Bases: Protocol

Protocol defining how a specific procedure (like experiments) configures the SOP Agent.

This allows different domains to provide their own procedure definitions while using the same core SOP Agent orchestration logic.

__init__(*args, **kwargs)

get_allowed_tools(phase: str) → List[str]: Get the tools allowed in the given phase for this procedure.

get_completion_summary(state_data: Dict[str, Any]) → str: Get a summary when the procedure is complete.

get_sop_guidance_prompt(context: Dict[str, Any], state_data: Dict[str, Any]) → str: Get the SOP Agent guidance prompt for this procedure.

get_system_prompt(context: Dict[str, Any]) → str: Get the system prompt for the coding assistant in this procedure.

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

should_continue(state_data: Dict[str, Any]) → bool: Determine if the procedure should continue based on current state.

class plexus.cli.procedure.sop_agent_base.StandardOperatingProcedureAgent(procedure_id: str, procedure_definition: ProcedureDefinition, mcp_server, flow_manager: FlowManager | None = None, chat_recorder: ChatRecorder | None = None, openai_api_key: str | None = None, context: Dict[str, Any] | None = None, model_config: Dict[str, Any] | None = None)

Bases: object

General-purpose Standard Operating Procedure Agent for multi-agent orchestration.

This agent can orchestrate any procedure-driven multi-agent workflow by: 1. Following configurable standard operating procedures 2. Managing conversation flow through defined phases 3. Providing contextual guidance to coding assistant agents 4. Enforcing tool scoping and safety constraints 5. Recording and reporting procedure execution results

The agent is domain-agnostic and gets its procedure-specific behavior through injected ProcedureDefinition, FlowManager, and ChatRecorder components.

Initialize the StandardOperatingProcedureAgent.

Args:: procedure_id: Unique identifier for this procedure execution procedure_definition: Defines procedure-specific behavior (prompts, tools, etc.) mcp_server: MCP server providing tools to the coding assistant flow_manager: Optional flow manager for conversation state chat_recorder: Optional recorder for conversation logging openai_api_key: Optional OpenAI API key for SOP guidance LLM context: Optional procedure context with known information model_config: Optional model configuration dict for LLM parameters

__init__(procedure_id: str, procedure_definition: ProcedureDefinition, mcp_server, flow_manager: FlowManager | None = None, chat_recorder: ChatRecorder | None = None, openai_api_key: str | None = None, context: Dict[str, Any] | None = None, model_config: Dict[str, Any] | None = None)

Initialize the StandardOperatingProcedureAgent.

Args:: procedure_id: Unique identifier for this procedure execution procedure_definition: Defines procedure-specific behavior (prompts, tools, etc.) mcp_server: MCP server providing tools to the coding assistant flow_manager: Optional flow manager for conversation state chat_recorder: Optional recorder for conversation logging openai_api_key: Optional OpenAI API key for SOP guidance LLM context: Optional procedure context with known information model_config: Optional model configuration dict for LLM parameters

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

Execute the SOP-guided procedure with coding assistant.

This is the main execution method that orchestrates the multi-agent ReAct loop: 1. SOP Agent provides procedural guidance 2. Coding Assistant executes tools based on guidance 3. SOP Agent evaluates progress and provides next guidance 4. Loop continues until procedure completion criteria met

Returns:: Dictionary with execution results

get_system_prompt() → str: Get the system prompt for the coding assistant.

get_user_prompt() → str: Get the initial user prompt for the coding assistant.

async setup(procedure_yaml: str) → bool

Set up the SOP Agent with procedure configuration.

Args:: procedure_yaml: YAML configuration for this procedure
Returns:: True if setup successful, False otherwise

async plexus.cli.procedure.sop_agent_base.execute_sop_procedure(procedure_id: str, procedure_definition: ProcedureDefinition, procedure_yaml: str, mcp_server, flow_manager: FlowManager | None = None, chat_recorder: ChatRecorder | None = None, openai_api_key: str | None = None, context: Dict[str, Any] | None = None, model_config: Dict[str, Any] | None = None) → Dict[str, Any]

Execute a Standard Operating Procedure with the SOP Agent.

This is the main entry point for executing any procedure using the general-purpose StandardOperatingProcedureAgent.

Args:: procedure_id: Unique identifier for this procedure execution procedure_definition: Defines procedure-specific behavior procedure_yaml: YAML configuration for the procedure mcp_server: MCP server providing tools flow_manager: Optional flow manager for conversation state chat_recorder: Optional recorder for conversation logging openai_api_key: Optional OpenAI API key context: Optional procedure context model_config: Optional model configuration dict for LLM parameters
Returns:: Dictionary with execution results