1+ from dataclasses import dataclass
2+ from typing import Dict , List , Optional , Any
3+ import json
4+ from mcp .server .fastmcp import FastMCP
5+ from rich .console import Console
6+ from rich .panel import Panel
7+ from rich .text import Text
8+
9+ console = Console (stderr = True )
10+
11+ @dataclass
12+ class ThoughtData :
13+ thought : str
14+ thought_number : int
15+ total_thoughts : int
16+ next_thought_needed : bool
17+ is_revision : Optional [bool ] = None
18+ revises_thought : Optional [int ] = None
19+ branch_from_thought : Optional [int ] = None
20+ branch_id : Optional [str ] = None
21+ needs_more_thoughts : Optional [bool ] = None
22+
23+ class SequentialThinkingServer :
24+ def __init__ (self ):
25+ self .thought_history : List [ThoughtData ] = []
26+ self .branches : Dict [str , List [ThoughtData ]] = {}
27+
28+ def _validate_thought_data (self , input_data : dict ) -> ThoughtData :
29+ """Validate and convert input dictionary to ThoughtData."""
30+ required_fields = {
31+ "thought" : str ,
32+ "thoughtNumber" : int ,
33+ "totalThoughts" : int ,
34+ "nextThoughtNeeded" : bool
35+ }
36+
37+ for field , field_type in required_fields .items ():
38+ if field not in input_data :
39+ raise ValueError (f"Missing required field: { field } )
40+ if not isinstance (input_data [field ], field_type ):
41+ raise ValueError (f"Invalid type for { field } { field_type } )
42+
43+ return ThoughtData (
44+ thought = input_data ["thought" ],
45+ thought_number = input_data ["thoughtNumber" ],
46+ total_thoughts = input_data ["totalThoughts" ],
47+ next_thought_needed = input_data ["nextThoughtNeeded" ],
48+ is_revision = input_data .get ("isRevision" ),
49+ revises_thought = input_data .get ("revisesThought" ),
50+ branch_from_thought = input_data .get ("branchFromThought" ),
51+ branch_id = input_data .get ("branchId" ),
52+ needs_more_thoughts = input_data .get ("needsMoreThoughts" )
53+ )
54+
55+ def _format_thought (self , thought_data : ThoughtData ) -> Panel :
56+ """Format a thought into a rich Panel with appropriate styling."""
57+ if thought_data .is_revision :
58+ prefix = "🔄 Revision"
59+ context = f" (revising thought { thought_data .revises_thought }
60+ style = "yellow"
61+ elif thought_data .branch_from_thought :
62+ prefix = "🌿 Branch"
63+ context = f" (from thought { thought_data .branch_from_thought } { thought_data .branch_id }
64+ style = "green"
65+ else :
66+ prefix = "💭 Thought"
67+ context = ""
68+ style = "blue"
69+
70+ header = Text (f"{ prefix } { thought_data .thought_number } { thought_data .total_thoughts } { context } , style = style )
71+ content = Text (thought_data .thought )
72+
73+ return Panel .fit (
74+ content ,
75+ title = header ,
76+ border_style = style ,
77+ padding = (1 , 2 )
78+ )
79+
80+ def process_thought (self , input_data : Any ) -> dict :
81+ """Process a thought and return formatted response."""
82+ try :
83+ thought_data = self ._validate_thought_data (input_data )
84+
85+ # Adjust total thoughts if needed
86+ if thought_data .thought_number > thought_data .total_thoughts :
87+ thought_data .total_thoughts = thought_data .thought_number
88+
89+ # Store thought in history
90+ self .thought_history .append (thought_data )
91+
92+ # Handle branching
93+ if thought_data .branch_from_thought and thought_data .branch_id :
94+ if thought_data .branch_id not in self .branches :
95+ self .branches [thought_data .branch_id ] = []
96+ self .branches [thought_data .branch_id ].append (thought_data )
97+
98+ # Display formatted thought
99+ console .print (self ._format_thought (thought_data ))
100+
101+ return {
102+ "content" : [{
103+ "type" : "text" ,
104+ "text" : json .dumps ({
105+ "thoughtNumber" : thought_data .thought_number ,
106+ "totalThoughts" : thought_data .total_thoughts ,
107+ "nextThoughtNeeded" : thought_data .next_thought_needed ,
108+ "branches" : list (self .branches .keys ()),
109+ "thoughtHistoryLength" : len (self .thought_history )
110+ }, indent = 2 )
111+ }]
112+ }
113+
114+ except Exception as e :
115+ return {
116+ "content" : [{
117+ "type" : "text" ,
118+ "text" : json .dumps ({
119+ "error" : str (e ),
120+ "status" : "failed"
121+ }, indent = 2 )
122+ }],
123+ "isError" : True
124+ }
125+
126+ def create_server () -> FastMCP :
127+ """Create and configure the MCP server."""
128+ mcp = FastMCP ("sequential-thinking" )
129+ thinking_server = SequentialThinkingServer ()
130+
131+ @mcp .tool ()
132+ async def sequential_thinking (
133+ thought : str ,
134+ thought_number : int ,
135+ total_thoughts : int ,
136+ next_thought_needed : bool ,
137+ is_revision : Optional [bool ] = None ,
138+ revises_thought : Optional [int ] = None ,
139+ branch_from_thought : Optional [int ] = None ,
140+ branch_id : Optional [str ] = None ,
141+ needs_more_thoughts : Optional [bool ] = None
142+ ) -> str :
143+ """A detailed tool for dynamic and reflective problem-solving through thoughts.
144+
145+ This tool helps analyze problems through a flexible thinking process that can adapt and evolve.
146+ Each thought can build on, question, or revise previous insights as understanding deepens.
147+
148+ Args:
149+ thought: Your current thinking step
150+ thought_number: Current thought number in sequence
151+ total_thoughts: Current estimate of thoughts needed
152+ next_thought_needed: Whether another thought step is needed
153+ is_revision: Whether this revises previous thinking
154+ revises_thought: Which thought is being reconsidered
155+ branch_from_thought: Branching point thought number
156+ branch_id: Branch identifier
157+ needs_more_thoughts: If more thoughts are needed
158+ """
159+ input_data = {
160+ "thought" : thought ,
161+ "thoughtNumber" : thought_number ,
162+ "totalThoughts" : total_thoughts ,
163+ "nextThoughtNeeded" : next_thought_needed ,
164+ "isRevision" : is_revision ,
165+ "revisesThought" : revises_thought ,
166+ "branchFromThought" : branch_from_thought ,
167+ "branchId" : branch_id ,
168+ "needsMoreThoughts" : needs_more_thoughts
169+ }
170+
171+ result = thinking_server .process_thought (input_data )
172+ return result ["content" ][0 ]["text" ]
173+
174+ return mcp
175+
176+ def main ():
177+ """Main entry point for the sequential thinking server."""
178+ server = create_server ()
179+ return server .run ()
180+
181+ if __name__ == "__main__" :
182+ server = create_server ()
183+ server .run ()
0 commit comments