Add Sequential Thinking compatibility server and JWKS support

deploy/seqthink/mcp_server.py

@@ -0,0 +1,160 @@
+#!/usr/bin/env python3
+"""Sequential Thinking MCP compatibility server (HTTP wrapper)."""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+from typing import Any, Dict, List, Optional
+
+from fastapi import FastAPI, HTTPException
+import uvicorn
+from pydantic import BaseModel, Field, validator
+
+logging.basicConfig(level=logging.INFO, format="%(message)s")
+logger = logging.getLogger("seqthink")
+
+
+class ToolRequest(BaseModel):
+    tool: str
+    payload: Dict[str, Any]
+
+    @validator("tool")
+    def validate_tool(cls, value: str) -> str:
+        allowed = {
+            "sequentialthinking",
+            "mcp__sequential-thinking__sequentialthinking",
+        }
+        if value not in allowed:
+            raise ValueError(f"Unknown tool '{value}'")
+        return value
+
+
+class ToolResponse(BaseModel):
+    result: Optional[Dict[str, Any]] = None
+    error: Optional[str] = None
+
+
+class ThoughtData(BaseModel):
+    thought: str
+    thoughtNumber: int = Field(..., ge=1)
+    totalThoughts: int = Field(..., ge=1)
+    nextThoughtNeeded: bool
+    isRevision: Optional[bool] = False
+    revisesThought: Optional[int] = Field(default=None, ge=1)
+    branchFromThought: Optional[int] = Field(default=None, ge=1)
+    branchId: Optional[str] = None
+    needsMoreThoughts: Optional[bool] = None
+
+    @validator("totalThoughts")
+    def normalize_total(cls, value: int, values: Dict[str, Any]) -> int:
+        thought_number = values.get("thoughtNumber")
+        if thought_number is not None and value < thought_number:
+            return thought_number
+        return value
+
+
+class SequentialThinkingEngine:
+    """Replicates the upstream sequential thinking MCP behaviour."""
+
+    def __init__(self) -> None:
+        self._thought_history: List[ThoughtData] = []
+        self._branches: Dict[str, List[ThoughtData]] = {}
+        env = os.environ.get("DISABLE_THOUGHT_LOGGING", "")
+        self._disable_logging = env.lower() == "true"
+
+    def _record_branch(self, data: ThoughtData) -> None:
+        if data.branchFromThought and data.branchId:
+            self._branches.setdefault(data.branchId, []).append(data)
+
+    def _log_thought(self, data: ThoughtData) -> None:
+        if self._disable_logging:
+            return
+
+        header = []
+        if data.isRevision:
+            header.append("🔄 Revision")
+            if data.revisesThought:
+                header.append(f"(revising thought {data.revisesThought})")
+        elif data.branchFromThought:
+            header.append("🌿 Branch")
+            header.append(f"(from thought {data.branchFromThought})")
+            if data.branchId:
+                header.append(f"[ID: {data.branchId}]")
+        else:
+            header.append("💭 Thought")
+
+        header.append(f"{data.thoughtNumber}/{data.totalThoughts}")
+        header_line = " ".join(part for part in header if part)
+
+        border_width = max(len(header_line), len(data.thought)) + 4
+        border = "─" * border_width
+        message = (
+            f"\n┌{border}┐\n"
+            f"│ {header_line.ljust(border_width - 2)} │\n"
+            f"├{border}┤\n"
+            f"│ {data.thought.ljust(border_width - 2)} │\n"
+            f"└{border}┘"
+        )
+        logger.error(message)
+
+    def process(self, payload: Dict[str, Any]) -> Dict[str, Any]:
+        try:
+            thought = ThoughtData(**payload)
+        except Exception as exc:  # pylint: disable=broad-except
+            logger.exception("Invalid thought payload")
+            return {
+                "content": [
+                    {
+                        "type": "text",
+                        "text": json.dumps({"error": str(exc)}, indent=2),
+                    }
+                ],
+                "isError": True,
+            }
+
+        self._thought_history.append(thought)
+        self._record_branch(thought)
+        self._log_thought(thought)
+
+        response_payload = {
+            "thoughtNumber": thought.thoughtNumber,
+            "totalThoughts": thought.totalThoughts,
+            "nextThoughtNeeded": thought.nextThoughtNeeded,
+            "branches": list(self._branches.keys()),
+            "thoughtHistoryLength": len(self._thought_history),
+        }
+
+        return {
+            "content": [
+                {
+                    "type": "text",
+                    "text": json.dumps(response_payload, indent=2),
+                }
+            ]
+        }
+
+
+engine = SequentialThinkingEngine()
+app = FastAPI(title="Sequential Thinking MCP Compatibility Server")
+
+
+@app.get("/health")
+def health() -> Dict[str, str]:
+    return {"status": "ok"}
+
+
+@app.post("/mcp/tool")
+def call_tool(request: ToolRequest) -> ToolResponse:
+    try:
+        result = engine.process(request.payload)
+        if result.get("isError"):
+            return ToolResponse(error=result["content"][0]["text"])
+        return ToolResponse(result=result)
+    except Exception as exc:  # pylint: disable=broad-except
+        raise HTTPException(status_code=400, detail=str(exc)) from exc
+
+
+if __name__ == "__main__":
+    uvicorn.run(app, host="127.0.0.1", port=8000, log_level="info")