mcp-sequential-thinking

To support multi-user environments, you can extend the base ThoughtData model into a CollaborativeThoughtData model. This allows you to track authorship, comments, and social signals like upvotes.

Data Models

• User: Represents a participant with id, name, and email.
• Comment: Represents a user's feedback on a thought, containing id, user_id, content, and timestamp.
• CollaborativeThoughtData: Extends ThoughtData with created_by, last_modified_by, comments (List of Comment), and upvotes (Set of user IDs).
• CollaborativeSession: Manages a group of users and their shared CollaborativeThoughtData list.

Key Operations

• CollaborativeThoughtData.add_comment(user_id, content): Appends a new comment.
• CollaborativeThoughtData.toggle_upvote(user_id): Adds or removes a user's upvote.
• CollaborativeSession.add_participant(user): Adds a user to the session.

from pydantic import BaseModel, Field
from typing import Dict, List, Optional, Set
from datetime import datetime
import uuid

class CollaborativeThoughtData(ThoughtData):
    created_by: str
    last_modified_by: str
    comments: List[Comment] = Field(default_factory=list)
    upvotes: Set[str] = Field(default_factory=set)

    def add_comment(self, user_id: str, content: str) -> Comment:
        # ...

    def toggle_upvote(self, user_id: str) -> bool:
        # ...

You can enhance the server by adding an AdvancedAnalyzer that uses TfidfVectorizer and cosine_similarity from sklearn to find similar thoughts within a session. This allows the system to identify patterns or recurring themes in the thinking process.

from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.metrics.pairwise import cosine_similarity
import numpy as np

class AdvancedAnalyzer:
    """Advanced thought analysis using NLP techniques."""

    def __init__(self):
        """Initialize the analyzer."""
        self.vectorizer = TfidfVectorizer()
        self.thought_vectors = None
        self.thoughts = []

    def add_thought(self, thought: ThoughtData) -> None:
        """Add a thought to the analyzer."""
        self.thoughts.append(thought)
        # Recompute vectors
        self._compute_vectors()

    def _compute_vectors(self) -> None:
        """Compute TF-IDF vectors for all thoughts."""
        if not self.thoughts:
            return

        thought_texts = [t.thought for t in self.thoughts]
        self.thought_vectors = self.vectorizer.fit_transform(thought_texts)

    def find_similar_thoughts(self, thought: ThoughtData, top_n: int = 3) -> List[Tuple[ThoughtData, float]]:
        """Find thoughts similar to the given thought using cosine similarity."""
        if thought not in self.thoughts:
            self.add_thought(thought)

        thought_idx = self.thoughts.index(thought)
        thought_vector = self.thought_vectors[thought_idx]

        # Compute similarities
        similarities = cosine_similarity(thought_vector, self.thought_vectors).flatten()

        # Get top N similar thoughts (excluding self)
        similar_indices = np.argsort(similarities)[::-1][1:top_n+1]

        return [(self.thoughts[idx], similarities[idx]) for idx in similar_indices]

You can implement visualization capabilities for your Sequential Thinking server using matplotlib. Two useful patterns are creating a pie chart for stage distribution and a timeline for the thinking process. Both methods return a base64 encoded PNG string formatted as a data URI (data:image/png;base64,...), which is ideal for web-based or UI-driven clients.

Key methods to implement:

• create_stage_distribution_chart(thoughts: List[ThoughtData]) -> str
• create_thinking_timeline(thoughts: List[ThoughtData]) -> str

import matplotlib.pyplot as plt
import io
import base64
from typing import List, Dict, Any

class ThoughtVisualizer:
    @staticmethod
    def create_stage_distribution_chart(thoughts: List[ThoughtData]) -> str:
        # ... implementation returns base64 PNG string
        return f"data:image/png;base64,{img_str}"

    @staticmethod
    def create_thinking_timeline(thoughts: List[ThoughtData]) -> str:
        # ... implementation returns base64 PNG string
        return f"data:image/png;base64,{img_str}"

Use the @mcp.prompt() decorator to define custom prompts that guide the LLM through specific stages of the thinking process, such as problem_definition_prompt or research_prompt. These prompts can include SystemMessage and UserMessage to set context and instructions.

from mcp.server.fastmcp.prompts import base

@mcp.prompt()
def problem_definition_prompt(problem_statement: str) -> list[base.Message]:
    """Create a prompt for the Problem Definition stage."""
    return [
        base.SystemMessage(
            "You are a structured thinking assistant helping to define a problem clearly."
        ),
        base.UserMessage(f"I need to define this problem: {problem_statement}"),
        base.UserMessage(
            "Please help me create a clear problem definition by addressing:\n"
            "1. What is the core issue?\n"
            "2. Who is affected?\n"
            "3. What are the boundaries of the problem?\n"
            "4. What would a solution look like?\n"
            "5. What constraints exist?"
        )
    ]

@mcp.prompt()
def research_prompt(problem_definition: str) -> list[base.Message]:
    """Create a prompt for the Research stage."""
    return [
        base.SystemMessage(
            "You are a research assistant helping to gather information about a problem."
        ),
        base.UserMessage(f"I've defined this problem: {problem_definition}"),
        base.UserMessage(
            "Please help me research this problem by:\n"
            "1. Identifying key information needed\n"
            "2. Suggesting reliable sources\n"
            "3. Outlining research questions\n"
            "4. Proposing a research plan"
        )
    ]

To persist thought data across sessions, implement a database-backed storage solution. This involves creating a SQLAlchemy ThoughtModel that maps to the ThoughtData structure and a DatabaseStorage class to handle session management and data insertion.

from sqlalchemy import create_engine, Column, Integer, String, Float, Boolean, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, relationship

Base = declarative_base()

class ThoughtModel(Base):
    """SQLAlchemy model for thought data."""
    __tablename__ = "thoughts"

    id = Column(Integer, primary_key=True)
    thought = Column(String, nullable=False)
    thought_number = Column(Integer, nullable=False)
    total_thoughts = Column(Integer, nullable=False)
    next_thought_needed = Column(Boolean, nullable=False)
    stage = Column(String, nullable=False)
    timestamp = Column(String, nullable=False)

    tags = relationship("TagModel", back_populates="thought")
    axioms = relationship("AxiomModel", back_populates="thought")
    assumptions = relationship("AssumptionModel", back_populates="thought")

class DatabaseStorage:
    """Database-backed storage for thought data."""

    def __init__(self, db_url: str = "sqlite:///thoughts.db"):
        """Initialize database connection."""
        self.engine = create_engine(db_url)
        Base.metadata.create_all(self.engine)
        self.Session = sessionmaker(bind=self.engine)

    def add_thought(self, thought: ThoughtData) -> None:
        """Add a thought to the database."""
        with self.Session() as session:
            # Convert ThoughtData to ThoughtModel
            thought_model = ThoughtModel(
                thought=thought.thought,
                thought_number=thought.thought_number,
                total_thoughts=thought.total_thoughts,
                next_thought_needed=thought.next_thought_needed,
                stage=thought.stage.value,
                timestamp=thought.timestamp
            )

            session.add(thought_model)
            session.commit()

You can expose the Sequential Thinking server's functionality through a FastAPI web interface. This allows external clients to POST new thoughts to /thoughts/, GET all thoughts from /thoughts/, or retrieve a process summary from /summary/. Ensure CORS is configured if the UI is hosted on a different origin.

from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel

app = FastAPI(title="Sequential Thinking UI")

# Enable CORS
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

class ThoughtRequest(BaseModel):
    """Request model for adding a thought."""
    thought: str
    thought_number: int
    total_thoughts: int
    next_thought_needed: bool
    stage: str
    tags: List[str] = []
    axioms_used: List[str] = []
    assumptions_challenged: List[str] = []

@app.post("/thoughts/")
async def add_thought(request: ThoughtRequest):
    """Add a new thought."""
    try:
        # Convert stage string to enum
        thought_stage = ThoughtStage.from_string(request.stage)

        # Create thought data
        thought_data = ThoughtData(
            thought=request.thought,
            thought_number=request.thought_number,
            total_thoughts=request.total_thoughts,
            next_thought_needed=request.next_thought_needed,
            stage=thought_stage,
            tags=request.tags,
            axioms_used=request.axioms_used,
            assumptions_challenged=request.assumptions_challenged
        )

        # Store thought
        storage.add_thought(thought_data)

        # Analyze the thought
        all_thoughts = storage.get_all_thoughts()
        analysis = ThoughtAnalyzer.analyze_thought(thought_data, all_thoughts)

        return analysis
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

@app.get("/thoughts/")
async def get_thoughts():
    """Get all thoughts."""
    all_thoughts = storage.get_all_thoughts()
    return {
        "thoughts": [t.to_dict() for t in all_thoughts]
    }

@app.get("/summary/")
async def get_summary():
    """Get a summary of the thinking process."""
    all_thoughts = storage.get_all_thoughts()
    return ThoughtAnalyzer.generate_summary(all_thoughts)