fix(mcp): Move all imports to the top of modules

nextcloud_mcp_server/api/management.py

@@ -15,6 +15,7 @@ import time
 from importlib.metadata import version
 from typing import Any
 
+import httpx
 from starlette.requests import Request
 from starlette.responses import JSONResponse
 
@@ -530,8 +531,6 @@ async def get_installed_apps(request: Request) -> JSONResponse:
         )
 
     try:
-        import httpx
-
         # Get Bearer token from request
         token = extract_bearer_token(request)
         if not token:
@@ -602,8 +601,6 @@ async def list_webhooks(request: Request) -> JSONResponse:
         )
 
     try:
-        import httpx
-
         from nextcloud_mcp_server.client.webhooks import WebhooksClient
 
         # Get Bearer token from request
@@ -669,8 +666,6 @@ async def create_webhook(request: Request) -> JSONResponse:
         )
 
     try:
-        import httpx
-
         from nextcloud_mcp_server.client.webhooks import WebhooksClient
 
         # Parse request body
@@ -747,8 +742,6 @@ async def delete_webhook(request: Request) -> JSONResponse:
         )
 
     try:
-        import httpx
-
         from nextcloud_mcp_server.client.webhooks import WebhooksClient
 
         # Get webhook_id from path parameter

nextcloud_mcp_server/app.py

@@ -1,5 +1,7 @@
 from __future__ import annotations
 
+import base64
+import json
 import logging
 import os
 import time
@@ -11,13 +13,13 @@ from dataclasses import dataclass
 from typing import TYPE_CHECKING, Optional, cast
 from urllib.parse import urlparse
 
+import anyio
 from opentelemetry.instrumentation.httpx import HTTPXClientInstrumentor
 
 if TYPE_CHECKING:
     from nextcloud_mcp_server.auth.storage import RefreshTokenStorage
 
 
-import anyio
 import click
 import httpx
 from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStream
@@ -384,8 +386,6 @@ class BasicAuthMiddleware:
 
             if auth_header.startswith(b"Basic "):
                 try:
-                    import base64
-
                     # Decode base64(username:password)
                     encoded = auth_header[6:]  # Skip "Basic "
                     decoded = base64.b64decode(encoded).decode("utf-8")
@@ -1200,8 +1200,6 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
                 "OAuth credentials not configured - attempting Dynamic Client Registration..."
             )
 
-            import anyio
-
             async def setup_multi_user_basic_dcr():
                 """Setup DCR for multi-user BasicAuth background operations."""
                 # Construct registration endpoint directly from nextcloud_host
@@ -1288,7 +1286,6 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
     if mode in (AuthMode.OAUTH_SINGLE_AUDIENCE, AuthMode.OAUTH_TOKEN_EXCHANGE):
         logger.info("Configuring MCP server for OAuth mode")
         # Asynchronously get the OAuth configuration
-        import anyio
 
         (
             nextcloud_host,
@@ -1626,7 +1623,6 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
 
         # Start background vector sync tasks (ADR-007)
         # Scanner runs at server-level (once), not per-session
-        import anyio as anyio_module
 
         # Re-use settings from outer scope (already validated)
         # Note: enable_offline_access_for_sync, encryption_key, and refresh_token_storage
@@ -1666,11 +1662,11 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
                 ) from e
 
             # Initialize shared state
-            send_stream, receive_stream = anyio_module.create_memory_object_stream(
+            send_stream, receive_stream = anyio.create_memory_object_stream(
                 max_buffer_size=settings.vector_sync_queue_max_size
             )
-            shutdown_event = anyio_module.Event()
-            scanner_wake_event = anyio_module.Event()
+            shutdown_event = anyio.Event()
+            scanner_wake_event = anyio.Event()
 
             # Store in app state for access from routes (ADR-007)
             app.state.document_send_stream = send_stream
@@ -1697,7 +1693,7 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
                     break
 
             # Start background tasks using anyio TaskGroup
-            async with anyio_module.create_task_group() as tg:
+            async with anyio.create_task_group() as tg:
                 # Start scanner task
                 await tg.start(
                     scanner_task,
@@ -1828,11 +1824,11 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
                     ) from e
 
                 # Initialize shared state
-                send_stream, receive_stream = anyio_module.create_memory_object_stream(
+                send_stream, receive_stream = anyio.create_memory_object_stream(
                     max_buffer_size=settings.vector_sync_queue_max_size
                 )
-                shutdown_event = anyio_module.Event()
-                scanner_wake_event = anyio_module.Event()
+                shutdown_event = anyio.Event()
+                scanner_wake_event = anyio.Event()
 
                 # User state tracking for user manager
                 user_states: dict = {}
@@ -1869,7 +1865,7 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
                 use_basic_auth = not oauth_enabled
 
                 # Start background tasks using anyio TaskGroup
-                async with anyio_module.create_task_group() as tg:
+                async with anyio.create_task_group() as tg:
                     # Start user manager task (supervises per-user scanners)
                     await tg.start(
                         user_manager_task,
@@ -2076,7 +2072,6 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
         This is a temporary endpoint for testing webhook schemas and payloads.
         It logs the full payload and returns 200 OK immediately.
         """
-        import json
 
         try:
             payload = await request.json()
@@ -2467,8 +2462,6 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None =
                 # Starlette caches the body internally, so it's safe to read here
                 body = await request.body()
                 try:
-                    import json
-
                     data = json.loads(body)
                     # Check if this is an initialize request
                     if data.get("method") == "initialize":

nextcloud_mcp_server/auth/browser_oauth_routes.py

@@ -8,6 +8,7 @@ import hashlib
 import logging
 import os
 import secrets
+import time
 from base64 import urlsafe_b64encode
 from urllib.parse import urlencode
 
@@ -381,8 +382,6 @@ async def oauth_login_callback(request: Request) -> RedirectResponse | HTMLRespo
     refresh_expires_in = token_data.get("refresh_expires_in")
     refresh_expires_at = None
     if refresh_expires_in:
-        import time
-
         refresh_expires_at = int(time.time()) + refresh_expires_in
         logger.info(
             f"Refresh token expires in {refresh_expires_in}s (at timestamp {refresh_expires_at})"

nextcloud_mcp_server/auth/keycloak_oauth.py

@@ -8,6 +8,7 @@ Handles OAuth flows with Keycloak as the identity provider, including:
 - Integration with RefreshTokenStorage
 """
 
+import base64
 import hashlib
 import logging
 import os
@@ -155,7 +156,6 @@ class KeycloakOAuthClient:
         Returns:
             Tuple of (code_verifier, code_challenge)
         """
-        import base64
 
         # Generate code verifier (43-128 characters)
         code_verifier = secrets.token_urlsafe(32)

nextcloud_mcp_server/auth/oauth_routes.py

@@ -23,6 +23,7 @@ import hashlib
 import logging
 import os
 import secrets
+import time
 from base64 import urlsafe_b64encode
 from urllib.parse import urlencode
 
@@ -521,8 +522,6 @@ async def oauth_callback_nextcloud(request: Request):
         refresh_expires_in = token_data.get("refresh_expires_in")
         refresh_expires_at = None
         if refresh_expires_in:
-            import time
-
             refresh_expires_at = int(time.time()) + refresh_expires_in
             logger.info(f"  refresh_expires_in: {refresh_expires_in}s")
             logger.info(f"  refresh_expires_at: {refresh_expires_at}")

nextcloud_mcp_server/auth/provisioning_decorator.py

@@ -9,6 +9,7 @@ import functools
 import logging
 from typing import Callable
 
+import jwt
 from mcp.server.fastmcp import Context
 from mcp.shared.exceptions import McpError
 from mcp.types import ErrorData
@@ -78,8 +79,6 @@ def require_provisioning(func: Callable) -> Callable:
         user_id = None
         if hasattr(ctx, "authorization") and ctx.authorization:
             try:
-                import jwt
-
                 token = ctx.authorization.token
                 payload = jwt.decode(token, options={"verify_signature": False})
                 user_id = payload.get("sub")
@@ -163,8 +162,6 @@ def require_provisioning_or_suggest(func: Callable) -> Callable:
                 # Get user_id from authorization token
                 user_id = None
                 if hasattr(ctx, "authorization") and ctx.authorization:
-                    import jwt
-
                     token = ctx.authorization.token
                     payload = jwt.decode(token, options={"verify_signature": False})
                     user_id = payload.get("sub")

nextcloud_mcp_server/auth/storage.py

@@ -28,6 +28,7 @@ Sensitive data (tokens, secrets) is encrypted at rest using Fernet symmetric enc
 import json
 import logging
 import os
+import socket
 import time
 from pathlib import Path
 from typing import Any, Optional
@@ -830,7 +831,6 @@ class RefreshTokenStorage:
             resource_id: Resource identifier
             auth_method: Authentication method used
         """
-        import socket
 
         hostname = socket.gethostname()
         timestamp = int(time.time())

nextcloud_mcp_server/auth/userinfo_routes.py

@@ -9,6 +9,7 @@ For OAuth mode: Requires browser-based OAuth login to establish session.
 
 import logging
 import os
+import traceback
 from pathlib import Path
 from typing import Any
 
@@ -385,8 +386,6 @@ async def _get_user_info(request: Request) -> dict[str, Any]:
         return user_context
 
     except Exception as e:
-        import traceback
-
         logger.error(f"Error retrieving user info: {e}")
         logger.error(f"Traceback: {traceback.format_exc()}")
         return {

nextcloud_mcp_server/auth/viz_routes.py

@@ -15,6 +15,7 @@ import logging
 import time
 from pathlib import Path
 
+import anyio
 import numpy as np
 from jinja2 import Environment, FileSystemLoader
 from starlette.authentication import requires
@@ -396,8 +397,6 @@ async def vector_visualization_search(request: Request) -> JSONResponse:
             coords = pca.fit_transform(vectors)
             return coords, pca
 
-        import anyio
-
         with trace_operation(
             "vector_viz.pca_compute",
             attributes={

nextcloud_mcp_server/config.py

@@ -1,6 +1,7 @@
 import logging
 import logging.config
 import os
+import socket
 from dataclasses import dataclass
 from enum import Enum
 from typing import Any, Optional
@@ -337,7 +338,6 @@ class Settings:
         Returns:
             Collection name string
         """
-        import socket
 
         # Use explicit override if user configured non-default value
         if self.qdrant_collection != "nextcloud_content":

nextcloud_mcp_server/config_validators.py

@@ -9,6 +9,7 @@ See ADR-020 for detailed architecture and deployment mode documentation.
 """
 
 import logging
+import os
 from dataclasses import dataclass
 from enum import Enum
 
@@ -240,8 +241,6 @@ def detect_auth_mode(settings: Settings) -> AuthMode:
     Raises:
         ValueError: If explicit deployment_mode is invalid or conflicts with detected mode
     """
-    import logging
-    import os
 
     logger = logging.getLogger(__name__)
 

nextcloud_mcp_server/document_processors/pymupdf.py

@@ -6,6 +6,8 @@ import tempfile
 from collections.abc import Awaitable, Callable
 from typing import Any, Optional
 
+import anyio
+
 # NOTE: Do NOT call pymupdf.layout.activate() here!
 # It changes the behavior of pymupdf4llm.to_markdown() when page_chunks=True,
 # causing it to return a string instead of a list[dict].
@@ -95,7 +97,6 @@ class PyMuPDFProcessor(DocumentProcessor):
         Raises:
             ProcessorError: If PDF processing fails
         """
-        import anyio
 
         try:
             if progress_callback:

nextcloud_mcp_server/embedding/bm25_provider.py

@@ -3,6 +3,7 @@
 import logging
 from typing import Any
 
+import anyio
 from fastembed import SparseTextEmbedding
 
 logger = logging.getLogger(__name__)
@@ -67,7 +68,6 @@ class BM25SparseEmbeddingProvider:
         Returns:
             Dictionary with 'indices' and 'values' keys for Qdrant sparse vector
         """
-        import anyio
 
         # Run CPU-bound BM25 encoding in thread pool
         return await anyio.to_thread.run_sync(lambda: self.encode(text))  # type: ignore[attr-defined]
@@ -82,7 +82,6 @@ class BM25SparseEmbeddingProvider:
         Returns:
             List of dictionaries with 'indices' and 'values' for each text
         """
-        import anyio
 
         # Run CPU-bound BM25 encoding in thread pool to avoid blocking event loop
         sparse_embeddings = await anyio.to_thread.run_sync(  # type: ignore[attr-defined]

nextcloud_mcp_server/migrations.py

@@ -6,6 +6,7 @@ provides CLI integration.
 """
 
 import logging
+import sqlite3
 from pathlib import Path
 
 from alembic.config import Config
@@ -98,7 +99,6 @@ def get_current_revision(database_path: str | Path | None = None) -> str | None:
     Returns:
         Current revision ID or None if not versioned
     """
-    import sqlite3
 
     if database_path is None:
         database_path = "/app/data/tokens.db"

nextcloud_mcp_server/observability/metrics.py

@@ -14,7 +14,9 @@ and resource usage. Metrics are organized by category:
 - External Dependency Health Metrics
 """
 
+import functools
 import logging
+import time
 
 from prometheus_client import (
     Counter,
@@ -423,8 +425,6 @@ def instrument_tool(func):
     Returns:
         Wrapped function with metrics and tracing instrumentation
     """
-    import functools
-    import time
 
     from nextcloud_mcp_server.observability.tracing import trace_operation
 

nextcloud_mcp_server/search/algorithms.py

@@ -1,9 +1,16 @@
 """Base interfaces and data structures for search algorithms."""
 
+import logging
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
 from typing import Any, Protocol, runtime_checkable
 
+from qdrant_client.models import FieldCondition, Filter, MatchValue
+
+from nextcloud_mcp_server.config import get_settings
+from nextcloud_mcp_server.vector.placeholder import get_placeholder_filter
+from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client
+
 
 @runtime_checkable
 class NextcloudClientProtocol(Protocol):
@@ -78,13 +85,6 @@ async def get_indexed_doc_types(user_id: str) -> set[str]:
         >>> if "note" in types:
         ...     # Search notes
     """
-    import logging
-
-    from qdrant_client.models import FieldCondition, Filter, MatchValue
-
-    from nextcloud_mcp_server.config import get_settings
-    from nextcloud_mcp_server.vector.placeholder import get_placeholder_filter
-    from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client
 
     logger = logging.getLogger(__name__)
     settings = get_settings()

nextcloud_mcp_server/search/context.py

@@ -7,6 +7,9 @@ position markers for better visualization and understanding of search results.
 import logging
 from dataclasses import dataclass
 
+import pymupdf
+import pymupdf4llm
+
 from nextcloud_mcp_server.client import NextcloudClient
 
 logger = logging.getLogger(__name__)
@@ -549,8 +552,6 @@ async def _fetch_document_text(
                     # Extract text from PDF using PyMuPDF
                     # IMPORTANT: Use pymupdf4llm.to_markdown() to match indexing extraction
                     # This ensures character offsets align between indexed chunks and retrieval
-                    import pymupdf
-                    import pymupdf4llm
 
                     logger.debug(f"Extracting text from PDF: {file_path}")
                     pdf_doc = pymupdf.open(stream=file_content, filetype="pdf")

nextcloud_mcp_server/search/pdf_highlighter.py

@@ -10,6 +10,9 @@ varies between indexing and rendering.
 
 import logging
 import re
+import shutil
+import tempfile
+from pathlib import Path
 from typing import Optional
 
 import pymupdf
@@ -77,8 +80,6 @@ class PDFHighlighter:
             Tuple of (full_text, page_boundaries) where page_boundaries is a list of:
             {"page": 1, "start_offset": 0, "end_offset": 1234}
         """
-        import tempfile
-        from pathlib import Path
 
         page_boundaries = []
         text_parts = []
@@ -110,7 +111,6 @@ class PDFHighlighter:
         full_text = "".join(text_parts)
 
         # Clean up temp directory and extracted images
-        import shutil
 
         try:
             shutil.rmtree(temp_dir)
@@ -590,8 +590,6 @@ class PDFHighlighter:
         Returns:
             Tuple of (png_bytes, page_number, highlight_count) or None if failed
         """
-        import tempfile
-        from pathlib import Path
 
         temp_pdf_path = None
         try:

nextcloud_mcp_server/server/oauth_tools.py

@@ -12,6 +12,7 @@ from typing import Optional
 from urllib.parse import urlencode
 
 import httpx
+import jwt
 from mcp.server.auth.middleware.auth_context import get_access_token
 from mcp.server.auth.provider import AccessToken
 from mcp.server.fastmcp import Context
@@ -53,8 +54,6 @@ async def extract_user_id_from_token(ctx: Context) -> str:
     # Try JWT decode first
     if is_jwt:
         try:
-            import jwt
-
             payload = jwt.decode(token, options={"verify_signature": False})
             user_id = payload.get("sub", "unknown")
             logger.info(f"  ✓ JWT decode successful: user_id={user_id}")

nextcloud_mcp_server/server/semantic.py

@@ -1,6 +1,7 @@
 """Semantic search MCP tools using vector database."""
 
 import logging
+import os
 
 import anyio
 from httpx import RequestError
@@ -656,7 +657,6 @@ def configure_semantic_tools(mcp: FastMCP):
         This is useful for determining when vector indexing is complete
         after creating or updating content across all indexed apps.
         """
-        import os
 
         # Check if vector sync is enabled
         vector_sync_enabled = (

nextcloud_mcp_server/server/webdav.py

@@ -1,3 +1,4 @@
+import base64
 import logging
 
 from mcp.server.fastmcp import Context, FastMCP
@@ -120,7 +121,6 @@ def configure_webdav_tools(mcp: FastMCP):
                 pass
 
         # For binary files, return metadata and base64 encoded content
-        import base64
 
         return {
             "path": path,
@@ -156,8 +156,6 @@ def configure_webdav_tools(mcp: FastMCP):
 
         # Handle base64 encoded content
         if content_type and "base64" in content_type.lower():
-            import base64
-
             content_bytes = base64.b64decode(content)
             content_type = content_type.replace(";base64", "")
         else:

nextcloud_mcp_server/vector/document_chunker.py

@@ -3,6 +3,7 @@
 import logging
 from dataclasses import dataclass
 
+import anyio
 from langchain_text_splitters import RecursiveCharacterTextSplitter
 
 logger = logging.getLogger(__name__)
@@ -68,7 +69,6 @@ class DocumentChunker:
         Returns:
             List of chunks with their character positions in the original content
         """
-        import anyio
 
         # Handle empty content - return single empty chunk for backward compatibility
         if not content:

nextcloud_mcp_server/vector/html_processor.py

@@ -1,6 +1,7 @@
 """HTML to Markdown conversion utilities for vector sync."""
 
 import logging
+import re
 
 from markdownify import markdownify as md
 
@@ -43,7 +44,6 @@ def html_to_markdown(html_content: str | None) -> str:
     except Exception as e:
         logger.warning(f"Failed to convert HTML to Markdown: {e}")
         # Fallback: strip all HTML tags as a last resort
-        import re
 
         text = re.sub(r"<[^>]+>", " ", html_content)
         return " ".join(text.split())  # Normalize whitespace

nextcloud_mcp_server/vector/processor.py

@@ -3,6 +3,7 @@
 Processes documents from stream: fetches content, generates embeddings, stores in Qdrant.
 """
 
+import base64
 import logging
 import time
 import uuid
@@ -585,8 +586,6 @@ async def _index_document(
                 "vector_sync.pdf_size": len(content_bytes),
             },
         ):
-            import base64
-
             from nextcloud_mcp_server.search.pdf_highlighter import PDFHighlighter
 
             # Build chunk data for batch processing

nextcloud_mcp_server/vector/scanner.py

@@ -5,6 +5,7 @@ Periodically scans enabled users' content and queues changed documents for proce
 
 import logging
 import os
+import random
 import time
 from dataclasses import dataclass
 
@@ -167,7 +168,6 @@ async def scan_user_documents(
         nc_client: Authenticated Nextcloud client
         initial_sync: If True, send all documents (first-time sync)
     """
-    import random
 
     scan_id = random.randint(1000, 9999)
     logger.info(

tests/client/conftest.py

@@ -1,3 +1,5 @@
+import json
+
 import httpx
 
 # ============================================================================
@@ -22,14 +24,13 @@ def create_mock_response(
     Returns:
         Mock httpx.Response object
     """
-    import json as json_module
 
     if headers is None:
         headers = {}
 
     # If json_data is provided, serialize it to content
     if json_data is not None:
-        content = json_module.dumps(json_data).encode("utf-8")
+        content = json.dumps(json_data).encode("utf-8")
         headers.setdefault("content-type", "application/json")
 
     if content is None:

tests/conftest.py

@@ -1,7 +1,17 @@
+import base64
+import hashlib
+import json
 import logging
 import os
+import re
+import secrets
+import subprocess
+import threading
+import time
 import uuid
+from http.server import BaseHTTPRequestHandler, HTTPServer
 from typing import Any, AsyncGenerator
+from urllib.parse import parse_qs, quote, urlparse
 
 import anyio
 import httpx
@@ -257,7 +267,6 @@ async def nc_mcp_basic_auth_client(
 
     Uses anyio pytest plugin for proper async fixture handling.
     """
-    import base64
 
     credentials = base64.b64encode(b"admin:admin").decode("utf-8")
     auth_header = f"Basic {credentials}"
@@ -342,7 +351,6 @@ async def nc_mcp_oauth_client_with_elicitation(
         logger.info(f"  Schema: {params.schema}")
 
         # Extract OAuth URL from elicitation message
-        import re
 
         url_pattern = r"https?://[^\s]+"
         urls = re.findall(url_pattern, params.message)
@@ -1108,10 +1116,6 @@ def oauth_callback_server():
     # "OAuth tests with browser automation not supported in GitHub Actions CI"
     # )
 
-    import threading
-    from http.server import BaseHTTPRequestHandler, HTTPServer
-    from urllib.parse import parse_qs, urlparse
-
     # Use a dict to store auth codes keyed by state parameter
     # This allows multiple concurrent OAuth flows
     auth_states = {}
@@ -1758,9 +1762,6 @@ async def playwright_oauth_token(
     - Browser fixture provided by pytest-playwright-asyncio
     - See: https://playwright.dev/python/docs/test-runners
     """
-    import secrets
-    import time
-    from urllib.parse import quote
 
     nextcloud_host = os.getenv("NEXTCLOUD_HOST")
     username = os.getenv("NEXTCLOUD_USERNAME")
@@ -2047,9 +2048,6 @@ async def _get_oauth_token_with_scopes(
     Returns:
         OAuth access token string with requested scopes
     """
-    import secrets
-    import time
-    from urllib.parse import quote
 
     nextcloud_host = os.getenv("NEXTCLOUD_HOST")
     username = os.getenv("NEXTCLOUD_USERNAME")
@@ -2417,9 +2415,6 @@ async def _get_oauth_token_for_user(
     Returns:
         OAuth access token string
     """
-    import secrets
-    import time
-    from urllib.parse import quote
 
     nextcloud_host = os.getenv("NEXTCLOUD_HOST")
 
@@ -2560,7 +2555,6 @@ async def all_oauth_tokens(
     Now uses the real callback server with state parameters for reliable
     concurrent token acquisition without race conditions.
     """
-    import time
 
     # Get auth_states dict from callback server
     auth_states, callback_url = oauth_callback_server
@@ -2711,7 +2705,6 @@ async def test_user(nc_client: NextcloudClient):
             user_config = test_user
             await nc_client.users.create_user(**user_config)
     """
-    import uuid
 
     # Generate unique user ID to avoid conflicts
     userid = f"testuser_{uuid.uuid4().hex[:8]}"
@@ -2747,7 +2740,6 @@ async def test_group(nc_client: NextcloudClient):
 
     Returns the group ID.
     """
-    import uuid
 
     # Generate unique group ID to avoid conflicts
     groupid = f"testgroup_{uuid.uuid4().hex[:8]}"
@@ -2882,11 +2874,6 @@ async def _get_keycloak_oauth_token(
     Returns:
         OAuth access token string from Keycloak
     """
-    import base64
-    import hashlib
-    import secrets
-    import time
-    from urllib.parse import quote
 
     # Get auth_states dict from callback server
     auth_states, _ = oauth_callback_server
@@ -3252,8 +3239,6 @@ async def configure_astrolabe_for_mcp_server(nc_client):
             - mcp_server_public_url: Public URL for OAuth token audience validation
             - client_id: Optional OAuth client ID (default: "nextcloudMcpServerUIPublicClient")
     """
-    import json
-    import subprocess
 
     async def _configure(
         mcp_server_internal_url: str,

tests/integration/client/test_document_processing_progress.py

@@ -1,6 +1,7 @@
 """Integration tests for document processing with progress notifications."""
 
 import io
+import os
 
 import pytest
 from PIL import Image
@@ -13,7 +14,6 @@ class TestDocumentProcessingProgress:
 
     async def test_unstructured_processor_with_progress_callback(self, nc_client):
         """Test that UnstructuredProcessor calls progress callback during processing."""
-        import os
 
         # Skip if unstructured is not enabled
         if os.getenv("ENABLE_UNSTRUCTURED", "false").lower() != "true":
@@ -71,7 +71,6 @@ class TestDocumentProcessingProgress:
         self, nc_mcp_client, nc_client
     ):
         """Test that reading a document via WebDAV MCP tool sends progress notifications."""
-        import os
 
         # Skip if document processing is not enabled
         if os.getenv("ENABLE_DOCUMENT_PROCESSING", "false").lower() != "true":
@@ -110,7 +109,6 @@ class TestDocumentProcessingProgress:
 
     async def test_progress_callback_not_required(self, nc_client):
         """Test that processing works without progress callback (backward compatibility)."""
-        import os
 
         if os.getenv("ENABLE_UNSTRUCTURED", "false").lower() != "true":
             pytest.skip("Unstructured processor not enabled")

tests/integration/test_astrolabe_multi_user_background_sync.py

@@ -13,6 +13,8 @@ app password entry → background sync activation → database verification.
 """
 
 import logging
+import re
+import subprocess
 
 import anyio
 import pytest
@@ -151,7 +153,6 @@ async def generate_app_password(
         )
 
     # Validate password format before returning
-    import re
 
     if not re.match(
         r"^[a-zA-Z0-9]{5}-[a-zA-Z0-9]{5}-[a-zA-Z0-9]{5}-[a-zA-Z0-9]{5}-[a-zA-Z0-9]{5}$",
@@ -350,7 +351,6 @@ async def verify_app_password_created(username: str) -> bool:
 
     # Query the database to check for background sync credentials
     # Astrolabe stores app passwords in oc_preferences, not oc_authtoken
-    import subprocess
 
     query = f"""
     SELECT userid, configkey, configvalue

tests/integration/test_sampling.py

@@ -16,6 +16,7 @@ vector database with indexed test data.
 import json
 from unittest.mock import MagicMock
 
+import anyio
 import pytest
 from mcp.types import CreateMessageResult, TextContent
 
@@ -67,7 +68,6 @@ async def test_semantic_search_answer_successful_sampling(
     await require_vector_sync_tools(nc_mcp_client)
 
     # Get initial indexed count before creating note
-    import asyncio
 
     initial_sync = await nc_mcp_client.call_tool(
         "nc_get_vector_sync_status", arguments={}
@@ -118,7 +118,7 @@ Avoid blocking operations in async code.""",
             )
             break
 
-        await asyncio.sleep(wait_interval)
+        await anyio.sleep(wait_interval)
         waited += wait_interval
 
     # Verify sync completed
@@ -247,7 +247,6 @@ async def test_semantic_search_answer_with_limit(nc_mcp_client, temporary_note_f
     )
 
     # Wait for vector indexing to complete
-    import asyncio
 
     max_wait = 30
     wait_interval = 1
@@ -262,7 +261,7 @@ async def test_semantic_search_answer_with_limit(nc_mcp_client, temporary_note_f
         if status_data["status"] == "idle" and status_data["pending_count"] == 0:
             break
 
-        await asyncio.sleep(wait_interval)
+        await anyio.sleep(wait_interval)
         waited += wait_interval
 
     assert waited < max_wait, f"Vector sync did not complete within {max_wait} seconds"
@@ -306,7 +305,6 @@ async def test_semantic_search_answer_score_threshold(
     )
 
     # Wait for vector indexing to complete
-    import asyncio
 
     max_wait = 30
     wait_interval = 1
@@ -321,7 +319,7 @@ async def test_semantic_search_answer_score_threshold(
         if status_data["status"] == "idle" and status_data["pending_count"] == 0:
             break
 
-        await asyncio.sleep(wait_interval)
+        await anyio.sleep(wait_interval)
         waited += wait_interval
 
     assert waited < max_wait, f"Vector sync did not complete within {max_wait} seconds"
@@ -371,7 +369,6 @@ async def test_semantic_search_answer_max_tokens(nc_mcp_client, temporary_note_f
     )
 
     # Wait for vector indexing to complete
-    import asyncio
 
     max_wait = 30
     wait_interval = 1
@@ -386,7 +383,7 @@ async def test_semantic_search_answer_max_tokens(nc_mcp_client, temporary_note_f
         if status_data["status"] == "idle" and status_data["pending_count"] == 0:
             break
 
-        await asyncio.sleep(wait_interval)
+        await anyio.sleep(wait_interval)
         waited += wait_interval
 
     assert waited < max_wait, f"Vector sync did not complete within {max_wait} seconds"

tests/integration/test_semantic_search.py

@@ -10,6 +10,7 @@ Uses SimpleEmbeddingProvider for deterministic, in-process embeddings
 without requiring external services like Ollama.
 """
 
+import math
 import tempfile
 from pathlib import Path
 
@@ -147,7 +148,6 @@ async def test_simple_embedding_provider_deterministic(simple_embedding_provider
     assert len(embedding1) == 384
 
     # Should be normalized (unit length)
-    import math
 
     norm = math.sqrt(sum(x * x for x in embedding1))
     assert abs(norm - 1.0) < 1e-6
@@ -340,7 +340,6 @@ async def test_batch_embedding(simple_embedding_provider: SimpleEmbeddingProvide
     assert all(len(emb) == 384 for emb in embeddings)
 
     # Each should be normalized
-    import math
 
     for emb in embeddings:
         norm = math.sqrt(sum(x * x for x in emb))

tests/load/oauth_metrics.py

@@ -6,6 +6,7 @@ workflow completion rates, and cross-user operation latencies.
 """
 
 import statistics
+import time
 from collections import Counter, defaultdict
 from typing import Any
 
@@ -44,13 +45,11 @@ class OAuthBenchmarkMetrics:
 
     def start(self):
         """Mark the start of the benchmark."""
-        import time
 
         self.start_time = time.time()
 
     def stop(self):
         """Mark the end of the benchmark."""
-        import time
 
         self.end_time = time.time()
 

tests/load/oauth_pool.py

@@ -5,8 +5,12 @@ Manages multiple OAuth-authenticated users for realistic multi-user load testing
 """
 
 import logging
+import secrets
+import string
+import time
 from dataclasses import dataclass
 from typing import Any
+from urllib.parse import quote
 
 import anyio
 import httpx
@@ -333,8 +337,6 @@ class OAuthUserPool:
             TimeoutError: If callback not received within timeout
             ValueError: If token exchange fails
         """
-        import time
-        from urllib.parse import quote
 
         logger.info(f"Starting Playwright OAuth flow for {username}...")
         logger.debug(f"Using state: {state[:16]}...")
@@ -478,8 +480,6 @@ class UserSessionWrapper:
 
 def generate_secure_password(length: int = 20) -> str:
     """Generate a secure random password."""
-    import secrets
-    import string
 
     alphabet = string.ascii_letters + string.digits + "!@#$%^&*()"
     return "".join(secrets.choice(alphabet) for _ in range(length))

tests/load/workloads.py

@@ -4,6 +4,7 @@ Workload definitions for load testing the MCP server.
 Defines realistic operation mixes and individual operation functions.
 """
 
+import json
 import logging
 import random
 import time
@@ -91,8 +92,6 @@ class WorkloadOperations:
             if result and len(result.content) > 0:
                 content = result.content[0]
                 if hasattr(content, "text"):
-                    import json
-
                     note_data = json.loads(content.text)
                     note_id = note_data.get("id")
                     if note_id:
@@ -222,8 +221,6 @@ class MixedWorkload:
                         "nc_notes_get_note", {"note_id": note_id}
                     )
                     if get_result and len(get_result.content) > 0:
-                        import json
-
                         note_data = json.loads(get_result.content[0].text)
                         etag = note_data.get("etag", "")
                         self._warmup_note_ids.append((note_id, etag))

tests/manual/test_nextcloud_impersonate.py

@@ -18,6 +18,7 @@ Usage:
 import asyncio
 import logging
 import os
+import re
 import sys
 
 # Add parent directory to path
@@ -127,7 +128,6 @@ async def main():
         )
 
         # Extract requesttoken from HTML
-        import re
 
         token_match = re.search(r'data-requesttoken="([^"]+)"', settings_response.text)
         if token_match:

tests/server/oauth/test_keycloak_dcr.py

@@ -17,6 +17,7 @@ Architecture:
     MCP Client → Keycloak DCR → Keycloak OAuth → MCP Server → Nextcloud APIs
 """
 
+import json
 import logging
 import os
 import secrets
@@ -623,7 +624,6 @@ async def test_keycloak_dcr_architecture():
     }
 
     logger.info("Keycloak DCR Architecture:")
-    import json
 
     logger.info(json.dumps(architecture, indent=2))
 

tests/server/oauth/test_scope_authorization.py

@@ -11,13 +11,15 @@ Note: Tests use JWT OAuth tokens because scopes are embedded in the token payloa
 enabling efficient scope-based tool filtering without additional API calls.
 """
 
+import logging
+
+import httpx
 import pytest
 
 
 @pytest.mark.integration
 async def test_prm_endpoint():
     """Test that the Protected Resource Metadata endpoint returns correct data."""
-    import httpx
 
     # Test the PRM endpoint directly (RFC 9728 - path includes /mcp resource)
     async with httpx.AsyncClient() as client:
@@ -60,7 +62,6 @@ async def test_basicauth_shows_all_tools(nc_mcp_client):
 @pytest.mark.integration
 async def test_read_only_token_filters_write_tools(nc_mcp_oauth_client_read_only):
     """Test that a token with only read scopes filters out write tools."""
-    import logging
 
     logger = logging.getLogger(__name__)
 
@@ -109,7 +110,6 @@ async def test_read_only_token_filters_write_tools(nc_mcp_oauth_client_read_only
 @pytest.mark.integration
 async def test_write_only_token_filters_read_tools(nc_mcp_oauth_client_write_only):
     """Test that a token with only write scopes filters out read tools."""
-    import logging
 
     logger = logging.getLogger(__name__)
 
@@ -158,7 +158,6 @@ async def test_write_only_token_filters_read_tools(nc_mcp_oauth_client_write_onl
 @pytest.mark.integration
 async def test_full_access_token_shows_all_tools(nc_mcp_oauth_client_full_access):
     """Test that a token with both read and write scopes scopes can see all tools."""
-    import logging
 
     logger = logging.getLogger(__name__)
 
@@ -402,7 +401,6 @@ async def test_jwt_with_no_custom_scopes_returns_zero_tools(
     - OAuth provisioning tools (requiring only 'openid') remain visible
       so users can provision Nextcloud access after authentication
     """
-    import logging
 
     logger = logging.getLogger(__name__)
 
@@ -442,7 +440,6 @@ async def test_jwt_consent_scenarios_read_only(nc_mcp_oauth_client_read_only):
     Simulates user granting only read permission during OAuth consent.
     Expected: Should see read tools but not write tools.
     """
-    import logging
 
     logger = logging.getLogger(__name__)
 
@@ -480,7 +477,6 @@ async def test_jwt_consent_scenarios_write_only(nc_mcp_oauth_client_write_only):
     Simulates user granting only write permission during OAuth consent.
     Expected: Should see write tools but not read-only tools.
     """
-    import logging
 
     logger = logging.getLogger(__name__)
 
@@ -518,7 +514,6 @@ async def test_jwt_consent_scenarios_full_access(nc_mcp_oauth_client_full_access
     Simulates user granting both permissions during OAuth consent.
     Expected: Should see all 90+ tools (both read and write).
     """
-    import logging
 
     logger = logging.getLogger(__name__)
 

tests/server/oauth/test_token_exchange.py

@@ -6,10 +6,12 @@ Tests the critical token exchange pattern that separates:
 """
 
 import os
+import tempfile
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import jwt
 import pytest
+from cryptography.fernet import Fernet
 
 from nextcloud_mcp_server.auth.storage import RefreshTokenStorage
 from nextcloud_mcp_server.auth.token_broker import TokenBrokerService
@@ -21,9 +23,6 @@ pytestmark = pytest.mark.unit
 @pytest.fixture
 async def token_storage():
     """Create test token storage."""
-    import tempfile
-
-    from cryptography.fernet import Fernet
 
     # Generate valid Fernet key
     encryption_key = Fernet.generate_key()

tests/server/test_calendar_todos_mcp.py

@@ -1,5 +1,6 @@
 """Integration tests for Calendar VTODO (task) MCP tools."""
 
+import json
 import logging
 from datetime import datetime, timedelta
 
@@ -41,7 +42,6 @@ async def test_mcp_todo_complete_workflow(
 
         # Extract UID from the result
         result_data = create_result.content[0].text
-        import json
 
         result_json = json.loads(result_data)
         todo_uid = result_json["uid"]
@@ -156,7 +156,6 @@ async def test_mcp_list_todos_with_filters(
             {"calendar_name": calendar_name, "status": "NEEDS-ACTION"},
         )
         assert result.isError is False
-        import json
 
         data = json.loads(result.content[0].text)
         needs_action_todos = [t for t in data["todos"] if t["uid"] in created_uids]
@@ -253,8 +252,6 @@ async def test_mcp_search_todos_across_calendars(
         )
         assert search_result.isError is False
 
-        import json
-
         data = json.loads(search_result.content[0].text)
         assert "todos" in data
 
@@ -388,8 +385,6 @@ async def test_mcp_todo_with_dates(
         )
         assert create_result.isError is False
 
-        import json
-
         result_data = json.loads(create_result.content[0].text)
         todo_uid = result_data["uid"]
 
@@ -432,8 +427,6 @@ async def test_mcp_todo_categories(
         )
         assert create_result.isError is False
 
-        import json
-
         result_data = json.loads(create_result.content[0].text)
         todo_uid = result_data["uid"]
 

tests/unit/test_config.py

@@ -1,5 +1,6 @@
 """Tests for configuration validation."""
 
+import logging
 import os
 from unittest.mock import patch
 
@@ -48,7 +49,6 @@ class TestQdrantConfigValidation:
 
     def test_api_key_warning_in_local_mode(self, caplog):
         """Test that API key in local mode triggers warning."""
-        import logging
 
         caplog.set_level(logging.WARNING, logger="nextcloud_mcp_server.config")
         Settings(
@@ -59,7 +59,6 @@ class TestQdrantConfigValidation:
 
     def test_api_key_no_warning_in_network_mode(self, caplog):
         """Test that API key in network mode doesn't trigger warning."""
-        import logging
 
         caplog.set_level(logging.WARNING, logger="nextcloud_mcp_server.config")
         Settings(
@@ -206,7 +205,6 @@ class TestChunkConfigValidation:
 
     def test_small_chunk_size_warning(self, caplog):
         """Test that chunk size < 512 triggers warning."""
-        import logging
 
         caplog.set_level(logging.WARNING, logger="nextcloud_mcp_server.config")
         Settings(
@@ -221,7 +219,6 @@ class TestChunkConfigValidation:
 
     def test_reasonable_chunk_size_no_warning(self, caplog):
         """Test that chunk size >= 512 doesn't trigger warning."""
-        import logging
 
         caplog.set_level(logging.WARNING, logger="nextcloud_mcp_server.config")
         Settings(

tests/unit/test_hybrid_auth_setup.py

@@ -8,6 +8,7 @@ APIs use OAuth.
 
 from unittest.mock import AsyncMock, MagicMock
 
+import httpx
 import pytest
 
 from nextcloud_mcp_server.app import setup_oauth_config_for_multi_user_basic
@@ -207,7 +208,6 @@ class TestSetupOAuthConfigForMultiUserBasic:
         self, hybrid_auth_settings, mocker
     ):
         """Test handling of OIDC discovery HTTP errors."""
-        import httpx
 
         # Create a mock response with a status error
         mock_response = MagicMock()