Sequential Thinking Multi-Agent System

Instructions

Implementation Reference

• src/mcp_server_mas_sequential_thinking/main.py:147-229 (handler)

  The main MCP tool handler function 'sequentialthinking' that sanitizes inputs, creates validated ThoughtData, and delegates processing to the ThoughtProcessor with comprehensive error handling.
  @mcp.tool() async def sequentialthinking( thought: str, thoughtNumber: int, totalThoughts: int, nextThoughtNeeded: bool, isRevision: bool, branchFromThought: int | None, branchId: str | None, needsMoreThoughts: bool, ) -> str: """Advanced sequential thinking tool with multi-agent coordination. Processes thoughts through a specialized team of AI agents that coordinate to provide comprehensive analysis, planning, research, critique, and synthesis. Args: thought: Content of the thinking step (required) thoughtNumber: Sequence number starting from {ThoughtProcessingLimits.MIN_THOUGHT_SEQUENCE} (≥{ThoughtProcessingLimits.MIN_THOUGHT_SEQUENCE}) totalThoughts: Estimated total thoughts required (≥1) nextThoughtNeeded: Whether another thought step follows this one isRevision: Whether this thought revises a previous thought branchFromThought: Thought number to branch from for alternative exploration branchId: Unique identifier for the branch (required if branchFromThought set) needsMoreThoughts: Whether more thoughts are needed beyond the initial estimate Returns: Synthesized response from the multi-agent team with guidance for next steps Raises: ProcessingError: When thought processing fails ValidationError: When input validation fails RuntimeError: When server state is invalid """ # Capture server state locally to avoid async race conditions current_server_state = _server_state if current_server_state is None: return "Server Error: Server not initialized" try: # Create and validate thought data using refactored function thought_data = create_validated_thought_data( thought=thought, thoughtNumber=thoughtNumber, totalThoughts=totalThoughts, nextThoughtNeeded=nextThoughtNeeded, isRevision=isRevision, branchFromThought=branchFromThought, branchId=branchId, needsMoreThoughts=needsMoreThoughts, ) # Use captured state directly to avoid race conditions global _thought_processor async with _processor_lock: if _thought_processor is None: logger.info( "Initializing ThoughtProcessor with Multi-Thinking workflow" ) _thought_processor = ThoughtProcessor(current_server_state.session) result = await _thought_processor.process_thought(thought_data) logger.info(f"Successfully processed thought #{thoughtNumber}") return result except ValidationError as e: error_msg = f"Input validation failed for thought #{thoughtNumber}: {e}" logger.exception(error_msg) return f"Validation Error: {e}" except ThoughtProcessingError as e: error_msg = f"Processing failed for thought #{thoughtNumber}: {e}" logger.exception(error_msg) if hasattr(e, "metadata") and e.metadata: logger.exception(f"Error metadata: {e.metadata}") return f"Processing Error: {e}" except Exception as e: error_msg = f"Unexpected error processing thought #{thoughtNumber}: {e}" logger.exception(error_msg) return f"Unexpected Error: {e}"
• src/mcp_server_mas_sequential_thinking/core/models.py:53-135 (schema)

  Pydantic model 'ThoughtData' providing input schema validation and type determination for sequential thinking thoughts.
  class ThoughtData(BaseModel): """Streamlined thought data model with consolidated validation.""" model_config = {"validate_assignment": True, "frozen": True} # Core fields thought: str = Field( ..., min_length=FieldLengthLimits.MIN_STRING_LENGTH, description="Content of the thought", ) # MCP API compatibility - camelCase field names required thoughtNumber: ThoughtNumber = Field( # noqa: N815 ..., ge=ValidationLimits.MIN_THOUGHT_NUMBER, description="Sequence number starting from 1", ) totalThoughts: int = Field( # noqa: N815 ..., ge=1, description="Estimated total thoughts", ) nextThoughtNeeded: bool = Field( # noqa: N815 ..., description="Whether another thought is needed" ) # Required workflow fields isRevision: bool = Field( # noqa: N815 ..., description="Whether this revises a previous thought" ) branchFromThought: ThoughtNumber | None = Field( # noqa: N815 ..., ge=ValidationLimits.MIN_THOUGHT_NUMBER, description="Thought number to branch from", ) branchId: BranchId | None = Field( # noqa: N815 ..., description="Unique branch identifier" ) needsMoreThoughts: bool = Field( # noqa: N815 ..., description="Whether more thoughts are needed beyond estimate" ) @property def thought_type(self) -> ThoughtType: """Determine the type of thought based on field values.""" if self.isRevision: return ThoughtType.REVISION if self.branchFromThought is not None: return ThoughtType.BRANCH return ThoughtType.STANDARD @model_validator(mode="before") @classmethod def validate_thought_data(cls, data: dict[str, Any]) -> dict[str, Any]: """Consolidated validation with simplified logic.""" if isinstance(data, dict): _validate_thought_relationships(data) return data def format_for_log(self) -> str: """Format thought for logging with optimized type-specific formatting.""" # Use match statement for modern Python pattern matching match self.thought_type: case ThoughtType.REVISION: prefix = ( f"Revision {self.thoughtNumber}/{self.totalThoughts} " f"(revising #{self.branchFromThought})" ) case ThoughtType.BRANCH: prefix = ( f"Branch {self.thoughtNumber}/{self.totalThoughts} " f"(from #{self.branchFromThought}, ID: {self.branchId})" ) case _: # ThoughtType.STANDARD prefix = f"Thought {self.thoughtNumber}/{self.totalThoughts}" # Use multiline string formatting for better readability return ( f"{prefix}\n" f" Content: {self.thought}\n" f" Next: {self.nextThoughtNeeded}, More: {self.needsMoreThoughts}" )
• src/mcp_server_mas_sequential_thinking/main.py:147-147 (registration)

  FastMCP @mcp.tool() decorator registering the 'sequentialthinking' tool.
  @mcp.tool()
• src/mcp_server_mas_sequential_thinking/services/thought_processor_refactored.py:73-136 (helper)

  ThoughtProcessor.process_thought method orchestrating the core sequential thinking workflow: logging, session management, context building, multi-thinking execution, response formatting.
  async def process_thought(self, thought_data: ThoughtData) -> str: """Process a thought through the appropriate workflow with comprehensive error handling. This is the main public API method that maintains backward compatibility while using the new service-based architecture internally. Args: thought_data: The thought data to process Returns: Processed thought response Raises: ThoughtProcessingError: If processing fails """ try: return await self._process_thought_internal(thought_data) except Exception as e: error_msg = f"Failed to process {thought_data.thought_type.value} thought #{thought_data.thoughtNumber}: {e}" logger.error(error_msg, exc_info=True) metadata: ProcessingMetadata = { "error_count": ProcessingDefaults.ERROR_COUNT_INITIAL, "retry_count": ProcessingDefaults.RETRY_COUNT_INITIAL, "processing_time": ProcessingDefaults.PROCESSING_TIME_INITIAL, } raise ThoughtProcessingError(error_msg, metadata) from e async def _process_thought_internal(self, thought_data: ThoughtData) -> str: """Internal thought processing logic using specialized services. Args: thought_data: The thought data to process Returns: Processed thought response """ start_time = time.time() # Log thought data and add to session (now async for thread safety) self._log_thought_data(thought_data) await self._session.add_thought(thought_data) # Build context using specialized service (now async for thread safety) input_prompt = await self._context_builder.build_context_prompt(thought_data) await self._context_builder.log_context_building(thought_data, input_prompt) # Execute Multi-Thinking workflow using specialized service ( content, workflow_result, total_time, ) = await self._workflow_executor.execute_workflow( thought_data, input_prompt, start_time ) # Format response using specialized service final_response = self._response_formatter.format_response(content, thought_data) # Log workflow completion self._workflow_executor.log_workflow_completion( thought_data, workflow_result, total_time, final_response ) return final_response
• src/mcp_server_mas_sequential_thinking/processors/multi_thinking_processor.py:57-122 (helper)

  MultiThinkingSequentialProcessor.process_with_multi_thinking implementing intelligent routing and parallel multi-agent thinking execution (factual, emotional, critical, optimistic, creative, synthesis).
  async def process_with_multi_thinking( self, thought_data: "ThoughtData", context_prompt: str = "" ) -> MultiThinkingProcessingResult: """Process thoughts using multi-thinking methodology with parallel execution.""" start_time = time.time() logger.info("Multi-thinking processing started") if logger.isEnabledFor(logging.INFO): logger.info("Input preview: %s", thought_data.thought[:100]) logger.info("Context length: %d chars", len(context_prompt)) try: # Step 1: Intelligent routing decision routing_decision = await self.router.route_thought(thought_data) logger.info("Selected strategy: %s", routing_decision.strategy.name) if logger.isEnabledFor(logging.INFO): sequence = [ direction.value for direction in routing_decision.strategy.thinking_sequence ] logger.info("Thinking sequence: %s", sequence) # Step 2: Execute processing based on complexity if routing_decision.strategy.complexity == ProcessingDepth.SINGLE: result = await self._process_single_direction( thought_data, context_prompt, routing_decision ) elif routing_decision.strategy.complexity == ProcessingDepth.DOUBLE: result = await self._process_double_direction_sequence( thought_data, context_prompt, routing_decision ) elif routing_decision.strategy.complexity == ProcessingDepth.TRIPLE: result = await self._process_triple_direction_sequence( thought_data, context_prompt, routing_decision ) else: # FULL result = await self._process_full_direction_sequence( thought_data, context_prompt, routing_decision ) processing_time = time.time() - start_time # Create final result final_result = MultiThinkingProcessingResult( content=result["final_content"], strategy_used=routing_decision.strategy.name, thinking_sequence=[ direction.value for direction in routing_decision.strategy.thinking_sequence ], processing_time=processing_time, complexity_score=routing_decision.complexity_metrics.complexity_score, cost_reduction=routing_decision.estimated_cost_reduction, individual_results=result.get("individual_results", {}), step_name="multi_thinking_processing", ) logger.info( "Multi-thinking processing completed - Time: %.3fs, Cost reduction: %.1f%%, Output: %d chars", processing_time, routing_decision.estimated_cost_reduction, len(final_result.content), ) return final_result