MCP Agile Flow

Instructions

Implementation Reference

• src/mcp_agile_flow/fastmcp_tools.py:120-153 (registration)

  MCP registration (@mcp.tool()), schema (function parameters with descriptions), and wrapper handler for the 'think' tool. Delegates to think_impl from think_tool.py.

  @mcp.tool()
  def think(
      thought: str,
      category: str = "default",
      depth: int = 0,
      timestamp: Optional[int] = None,
      references: Optional[List[str]] = None,
      metadata: Optional[Dict[str, Any]] = None,
  ) -> str:
      """
      Record a thought for later reference and analysis.
  
      This tool allows you to record thoughts during development or analysis processes.
      Thoughts can be organized by category and depth to create a hierarchical structure
      of analysis.
      """
      # Extract actual values if they're Field objects
      if hasattr(thought, "default"):
          thought = thought.default
      if hasattr(category, "default"):
          category = category.default
      if hasattr(depth, "default"):
          depth = depth.default
      if hasattr(timestamp, "default"):
          timestamp = timestamp.default
      if hasattr(references, "default"):
          references = references.default
      if hasattr(metadata, "default"):
          metadata = metadata.default
  
      result = think_impl(thought, category, depth, None)
      # Convert dict to formatted JSON string
      return json.dumps(result, indent=2)

• src/mcp_agile_flow/think_tool.py:170-202 (handler)

  Core handler function that implements the logic for recording a thought, creating a thought record, storing it via _storage, and returning success response.

  def think(
      thought: str,
      category: Optional[str] = None,
      depth: int = 1,
      previous_thought_id: Optional[int] = None,
  ) -> Dict[str, Any]:
      """Record a thought."""
      thought_id = len(_storage.get_thoughts()) + 1
      timestamp = datetime.now().isoformat()
  
      thought_record = {
          "thought_id": thought_id,
          "id": thought_id,  # Alias for backward compatibility
          "index": thought_id,  # Another alias used in some tests
          "thought": thought,
          "timestamp": timestamp,
          "category": category,
          "depth": depth,
          "previous_thought_id": previous_thought_id,
      }
  
      _storage.add_thought(thought_record)
  
      message = f"Thought recorded with ID {thought_id}"
      if category:
          message += f" in category '{category}'"
      if depth > 1:
          message += f" at depth {depth} (deeper analysis)"
      if depth > 2:
          message = message.replace("deeper analysis", "much deeper analysis")
  
      return {"success": True, "thought_id": thought_id, "message": message}

• src/mcp_agile_flow/think_tool.py:16-51 (helper)

  ThoughtStorage class: helper utility for persisting thoughts to a temporary JSON file.

  class ThoughtStorage:
      def __init__(self):
          self._storage_file = None
          self._thoughts = []
          self._init_storage()
  
      def _init_storage(self):
          """Initialize temporary file for thought storage."""
          temp = tempfile.NamedTemporaryFile(prefix="mcp_thoughts_", suffix=".tmp", delete=False)
          self._storage_file = temp.name
          temp.close()
          logger.debug(f"Initialized thought storage using temporary file: {self._storage_file}")
  
      def add_thought(self, thought: Dict[str, Any]):
          """Add a thought to storage."""
          self._thoughts.append(thought)
          self._save()
  
      def get_thoughts(self) -> List[Dict[str, Any]]:
          """Get all stored thoughts."""
          return self._thoughts
  
      def clear_thoughts(self, category: Optional[str] = None):
          """Clear stored thoughts, optionally by category."""
          if category:
              self._thoughts = [t for t in self._thoughts if t.get("category") != category]
          else:
              self._thoughts = []
          self._save()
  
      def _save(self):
          """Save thoughts to storage file."""
          with open(self._storage_file, "w") as f:
              json.dump(self._thoughts, f)