cities-mcp/cities_mcp.py

# Cities MCP: Neocities site management MCP server (stdio-only)
# Depends upon httpx and FastMCP libraries
# Created by Luxferre in 2026, released into public domain

import sys
import json
import hashlib
from pathlib import Path

try:
  import httpx
  from fastmcp import FastMCP
except ImportError:
  print("Please install required dependencies: pip install fastmcp httpx", file=sys.stderr)
  sys.exit(1)

mcp = FastMCP("neocities")

API_ROOT = "https://neocities.org/api"
CONFIG_FILE = Path.home() / ".multicities.json"

def get_auth_token(username: str) -> str:
  """Get auth token from config file for the given user"""
  if not CONFIG_FILE.exists():
    raise ValueError(f"Config file {CONFIG_FILE} not found. Please log in first.")
  with open(CONFIG_FILE, "r") as f:
    try:
      config = json.load(f)
    except json.JSONDecodeError:
      config = {}
  token = config.get(username)
  if not token:
    raise ValueError(f"No token found for user '{username}'. Please log in first.")
  return token

def get_auth_headers(username: str) -> dict:
  """Generate auth headers for HTTP requests"""
  return {"Authorization": f"Bearer {get_auth_token(username)}"}

@mcp.tool(description="Login to a Neocities account using username and password. This will fetch an API token and save it to ~/.multicities.json for future use.")
def login(username: str, password: str) -> str:
  """Login to Neocities to get an API token. Saves token to ~/.multicities.json"""
  response = httpx.get(f"{API_ROOT}/key", auth=(username, password))
  if response.status_code != 200:
    return f"Login failed: HTTP {response.status_code} - {response.text}"
  
  data = response.json()
  if data.get("result") == "error":
    return f"Login failed: {data.get('message')}"
  
  token = data.get("api_key")
  if not token:
    return "Login failed: No api_key in response."
  
  config: dict = {}
  if CONFIG_FILE.exists():
    with open(CONFIG_FILE, "r") as f:
      try:
        config = json.load(f)
      except json.JSONDecodeError:
        pass
        
  config[username] = token
  with open(CONFIG_FILE, "w") as f:
    json.dump(config, f, indent=2)
      
  return f"Login successful for {username}. Token saved."

@mcp.tool(description="Retrieve detailed information about a Neocities site, including views, hits, creation date, last update time, tags, and custom domain configuration.")
def info(username: str) -> str:
  """Get site info for the given username/sitename."""
  headers = get_auth_headers(username)
  response = httpx.get(f"{API_ROOT}/info", headers=headers)
  if response.status_code != 200:
    return f"Error {response.status_code}: {response.text}"
  
  data = response.json()
  if data.get("result") == "error":
    return f"Error: {data.get('message')}"
      
  info_dict = data.get("info", {})
  output = []
  output.append(f"Sitename: {info_dict.get('sitename')}")
  output.append(f"Views: {info_dict.get('views')}")
  output.append(f"Hits: {info_dict.get('hits')}")
  output.append(f"Created at: {info_dict.get('created_at')}")
  output.append(f"Last updated at: {info_dict.get('last_updated')}")
  output.append(f"Custom domain: {info_dict.get('domain', 'None')}")
  
  tags = info_dict.get('tags')
  if not tags:
    tags = []
  
  # Ensure standard tags logic (like cities.sh '@csv' mapping doesn't crash)
  tag_strs = [f'"{t}"' for t in tags]
  output.append(f"Tags: {','.join(tag_strs)}")
  return "\n".join(output)

@mcp.tool(description="List all files and directories on a Neocities site, including their types (file or dir), sizes in bytes, and relative paths.")
def list_files(username: str) -> str:
  """List site files for the given username/sitename."""
  headers = get_auth_headers(username)
  response = httpx.get(f"{API_ROOT}/list", headers=headers)
  if response.status_code != 200:
    return f"Error {response.status_code}: {response.text}"
      
  data = response.json()
  if data.get("result") == "error":
    return f"Error: {data.get('message')}"
      
  output = []
  for item in data.get("files", []):
    type_str = "dir" if item.get("is_directory") else "file"
    size = item.get("size", 0)
    path = item.get("path", "")
    output.append(f"{type_str}\t{size}\t{path}")
      
  return "\n".join(output)

@mcp.tool(description="Upload a single local file to a Neocities site at the specified path. If remote_path is not provided, the file's original name will be used.")
def upload(username: str, local_path: str, remote_path: str | None = None) -> str:
  """Upload a single file to Neocities."""
  headers = get_auth_headers(username)
  p = Path(local_path).resolve()
  if not p.is_file():
    return f"Error: Local file {local_path} not found."
      
  if not remote_path:
    remote_path = p.name
      
  with open(p, "rb") as f:
    files = {remote_path: f}
    response = httpx.post(f"{API_ROOT}/upload", headers=headers, files=files)
      
  if response.status_code != 200:
    return f"Error {response.status_code}: {response.text}"
       
  return f"Upload successful: {response.text}"

@mcp.tool(description="Delete a specific file or directory from a Neocities site given its remote path.")
def delete_file(username: str, remote_path: str) -> str:
  """Delete a remote file from Neocities."""
  headers = get_auth_headers(username)
  data = {"filenames[]": remote_path}
  response = httpx.post(f"{API_ROOT}/delete", headers=headers, data=data)
  if response.status_code != 200:
    return f"Error {response.status_code}: {response.text}"
  return f"Delete successful: {response.text}"

@mcp.tool(description="Synchronize a local directory with a Neocities site. It recursively hashes local files and only uploads those that have changed or are missing on the remote site, skipping hidden files and directories.")
def sync(username: str, local_dir: str) -> str:
  """Sync a local directory to Neocities. Skips hidden files and up-to-date files."""
  headers = get_auth_headers(username)
  base_dir = Path(local_dir).resolve()
  if not base_dir.is_dir():
    return f"Error: Directory {local_dir} not found."
      
  # Get remote list
  list_response = httpx.get(f"{API_ROOT}/list", headers=headers)
  if list_response.status_code != 200:
    return f"Failed to get remote list: {list_response.text}"
      
  remote_files = {}
  for item in list_response.json().get("files", []):
    if not item.get("is_directory"):
      remote_files[item["path"]] = item.get("sha1_hash")
          
  results = []
  
  # Iterate local files skipping hidden dots
  for p in base_dir.rglob("*"):
    if p.is_file():
      # Check if any parent part or filename starts with '.'
      if any(part.startswith('.') for part in p.relative_to(base_dir).parts):
        continue
              
      remote_path = p.relative_to(base_dir).as_posix()
          
      # calculate sha1
      sha1 = hashlib.sha1()
      with open(p, "rb") as f:
        while chunk := f.read(8192):
          sha1.update(chunk)
      local_hash = sha1.hexdigest()
          
      prev_hash = remote_files.get(remote_path)
      if prev_hash != local_hash:
        results.append(f"Updating {remote_path} (prev: {prev_hash}, cur: {local_hash})")
        with open(p, "rb") as f:
          files = {remote_path: f}
          upload_response = httpx.post(f"{API_ROOT}/upload", headers=headers, files=files)
          if upload_response.status_code != 200:
            results.append(f"Error uploading {remote_path}: {upload_response.text}")
          else:
            data = upload_response.json()
            if data.get("result") == "error":
              results.append(f"Error uploading {remote_path}: {data.get('message')}")
            else:
              results.append(f"Successfully uploaded {remote_path}")
                           
  results.append("All files synced!")
  return "\n".join(results)

if __name__ == "__main__":
  mcp.run()