move to uv and ruff

Author: soodoku

pyproject.toml

@@ -105,6 +105,10 @@ ignore = [
     "E501",  # line too long, handled by formatter
     "B008",  # do not perform function calls in argument defaults
     "C901",  # too complex
+    "B904",  # raise ... from err - not needed for our error patterns
+    "E722",  # bare except - allowed in test cleanup
+    "E402",  # module level import not at top - conditional imports needed
+    "B023",  # function definition does not bind loop variable - false positive
 ]
 
 [tool.ruff.format]

rmcp/bidirectional.py

@@ -22,7 +22,7 @@
 import json
 import logging
 import uuid
-from typing import Any, Dict, Optional, Set
+from typing import Any
 
 from .core.context import Context
 from .registries.tools import tool
@@ -39,15 +39,15 @@ class CallbackManager:
     """
 
     def __init__(self):
-        self.active_callbacks: Dict[str, Dict[str, Any]] = {}
-        self.callback_permissions: Dict[str, Set[str]] = {}
+        self.active_callbacks: dict[str, dict[str, Any]] = {}
+        self.callback_permissions: dict[str, set[str]] = {}
         self.callback_timeout = 300.0  # 5 minutes default
 
     def register_callback(
         self,
         callback_id: str,
         context: Context,
-        allowed_tools: Optional[Set[str]] = None,
+        allowed_tools: set[str] | None = None,
     ) -> None:
         """Register a callback session for R → Python communication."""
         self.active_callbacks[callback_id] = {
@@ -66,8 +66,8 @@ def unregister_callback(self, callback_id: str) -> None:
             logger.info(f"Unregistered callback session: {callback_id}")
 
     async def handle_callback(
-        self, callback_id: str, tool_name: str, parameters: Dict[str, Any]
-    ) -> Dict[str, Any]:
+        self, callback_id: str, tool_name: str, parameters: dict[str, Any]
+    ) -> dict[str, Any]:
         """Handle a callback request from R."""
         if callback_id not in self.active_callbacks:
             return {
@@ -99,7 +99,6 @@ async def handle_callback(
             context = callback_info["context"]
 
             # Import here to avoid circular imports
-            from .core.server import MCPServer
 
             server = context._server
 
@@ -174,8 +173,8 @@ def get_callback_manager() -> CallbackManager:
     },
 )
 async def create_r_callback_session(
-    context: Context, params: Dict[str, Any]
-) -> Dict[str, Any]:
+    context: Context, params: dict[str, Any]
+) -> dict[str, Any]:
     """
     Create a callback session that allows R scripts to call back to Python MCP tools.
 
@@ -235,7 +234,7 @@ async def create_r_callback_session(
         "required": ["callback_id", "tool_name", "parameters"],
     },
 )
-async def handle_r_callback(context: Context, params: Dict[str, Any]) -> Dict[str, Any]:
+async def handle_r_callback(context: Context, params: dict[str, Any]) -> dict[str, Any]:
     """
     Handle a callback request from an R script.
 
@@ -269,14 +268,14 @@ def create_r_callback_utilities() -> str:
 # Initialize callback system
 rmcp_init_callback <- function(callback_config) {
   .rmcp_callback_config <<- callback_config
-  
+
   # Create temporary file for callback communication
   .rmcp_callback_file <<- tempfile(pattern = "rmcp_callback_", fileext = ".json")
-  
+
   cat("RMCP callback system initialized\\n")
   cat("Callback ID:", callback_config$callback_id, "\\n")
   cat("Allowed tools:", paste(callback_config$allowed_tools, collapse = ", "), "\\n")
-  
+
   return(TRUE)
 }
 
@@ -285,28 +284,28 @@ def create_r_callback_utilities() -> str:
   if (is.null(.rmcp_callback_config)) {
     stop("Callback system not initialized. Call rmcp_init_callback() first.")
   }
-  
+
   # Check if tool is allowed
-  if (length(.rmcp_callback_config$allowed_tools) > 0 && 
+  if (length(.rmcp_callback_config$allowed_tools) > 0 &&
       !tool_name %in% .rmcp_callback_config$allowed_tools) {
     stop(paste("Tool", tool_name, "not allowed for callbacks"))
   }
-  
+
   # Prepare callback request
   callback_request <- list(
     callback_id = .rmcp_callback_config$callback_id,
     tool_name = tool_name,
     parameters = parameters,
     timestamp = as.numeric(Sys.time())
   )
-  
+
   # Write request to file
   write(toJSON(callback_request, auto_unbox = TRUE), .rmcp_callback_file)
-  
+
   # Signal Python to handle callback (this is a simplified version)
   # In practice, this would use a more sophisticated IPC mechanism
   cat("RMCP_CALLBACK:", .rmcp_callback_file, "\\n", file = stderr())
-  
+
   # For now, return a placeholder response
   # Real implementation would wait for Python response
   return(list(
@@ -327,10 +326,10 @@ def create_r_callback_utilities() -> str:
       timestamp = as.numeric(Sys.time()),
       callback_id = .rmcp_callback_config$callback_id
     )
-    
+
     cat("RMCP_PROGRESS_CALLBACK:", toJSON(progress_data, auto_unbox = TRUE), "\\n", file = stderr())
   }
-  
+
   # Also call regular progress function
   if (exists("rmcp_progress")) {
     rmcp_progress(message, current, total)
@@ -359,10 +358,10 @@ def create_r_callback_utilities() -> str:
   if (!is.null(.rmcp_callback_file) && file.exists(.rmcp_callback_file)) {
     unlink(.rmcp_callback_file)
   }
-  
+
   .rmcp_callback_config <<- NULL
   .rmcp_callback_file <<- NULL
-  
+
   cat("RMCP callback system cleaned up\\n")
 }
 
@@ -398,8 +397,8 @@ def create_r_callback_utilities() -> str:
     },
 )
 async def setup_r_bidirectional(
-    context: Context, params: Dict[str, Any]
-) -> Dict[str, Any]:
+    context: Context, params: dict[str, Any]
+) -> dict[str, Any]:
     """
     Set up bidirectional communication in an R session.
 
@@ -428,19 +427,19 @@ async def setup_r_bidirectional(
         setup_script = f"""
         # Load RMCP callback utilities
         {create_r_callback_utilities()}
-        
+
         # Initialize callback system
         callback_config <- {json.dumps(callback_config)}
         rmcp_init_callback(callback_config)
-        
+
         # Test callback system
         cat("Bidirectional communication ready\\n")
-        
+
         result <- list(
             bidirectional_enabled = TRUE,
             callback_id = callback_config$callback_id,
             allowed_tools = callback_config$allowed_tools,
-            session_id = "{session_id or 'default'}"
+            session_id = "{session_id or "default"}"
         )
         """
 
@@ -469,8 +468,8 @@ async def setup_r_bidirectional(
     input_schema={"type": "object", "properties": {}},
 )
 async def list_callback_sessions(
-    context: Context, params: Dict[str, Any]
-) -> Dict[str, Any]:
+    context: Context, params: dict[str, Any]
+) -> dict[str, Any]:
     """
     List all active callback sessions.
 

rmcp/cli.py

@@ -597,7 +597,7 @@ def _load_config(config_file: str) -> dict:
     import json
 
     try:
-        with open(config_file, "r") as f:
+        with open(config_file) as f:
             return json.load(f)
     except Exception as e:
         logger.error(f"Failed to load config file {config_file}: {e}")

rmcp/config/defaults.py

@@ -6,10 +6,10 @@
 """
 
 from pathlib import Path
-from typing import Any, Dict
+from typing import Any
 
 # Default configuration as a dictionary for easy serialization
-DEFAULT_CONFIG: Dict[str, Any] = {
+DEFAULT_CONFIG: dict[str, Any] = {
     "http": {
         "host": "localhost",
         "port": 8000,

rmcp/config/loader.py

@@ -57,11 +57,10 @@
 """
 
 import copy
-import functools
 import json
 import os
 from pathlib import Path
-from typing import Any, Dict, Optional, Union
+from typing import Any
 
 try:
     import jsonschema
@@ -127,12 +126,12 @@ def __init__(self):
         Creates a new configuration loader with empty cache.
         The cache will be populated on first load_config() call.
         """
-        self._config_cache: Optional[RMCPConfig] = None
+        self._config_cache: RMCPConfig | None = None
 
     def load_config(
         self,
-        config_file: Optional[Union[str, Path]] = None,
-        overrides: Optional[Dict[str, Any]] = None,
+        config_file: str | Path | None = None,
+        overrides: dict[str, Any] | None = None,
         validate: bool = True,
     ) -> RMCPConfig:
         """
@@ -175,8 +174,8 @@ def load_config(
         return self._dict_to_config(config_dict)
 
     def _load_config_file(
-        self, config_file: Optional[Union[str, Path]] = None
-    ) -> Optional[Dict[str, Any]]:
+        self, config_file: str | Path | None = None
+    ) -> dict[str, Any] | None:
         """Load configuration from JSON file."""
         config_paths = []
 
@@ -190,10 +189,10 @@ def _load_config_file(
         for config_path in config_paths:
             if config_path.exists() and config_path.is_file():
                 try:
-                    with open(config_path, "r", encoding="utf-8") as f:
+                    with open(config_path, encoding="utf-8") as f:
                         config_data = json.load(f)
                     return config_data
-                except (json.JSONDecodeError, IOError) as e:
+                except (OSError, json.JSONDecodeError) as e:
                     if config_file:
                         # If explicitly specified, raise error
                         raise ConfigError(
@@ -207,7 +206,7 @@ def _load_config_file(
 
         return None
 
-    def _load_environment_config(self) -> Dict[str, Any]:
+    def _load_environment_config(self) -> dict[str, Any]:
         """Load configuration from environment variables."""
         env_config = {}
 
@@ -242,7 +241,7 @@ def _convert_env_value(self, env_var: str, value: str) -> Any:
         # String value
         return value
 
-    def _set_nested_value(self, config_dict: Dict[str, Any], path: str, value: Any):
+    def _set_nested_value(self, config_dict: dict[str, Any], path: str, value: Any):
         """Set a nested dictionary value using dot notation."""
         keys = path.split(".")
         current = config_dict
@@ -255,8 +254,8 @@ def _set_nested_value(self, config_dict: Dict[str, Any], path: str, value: Any):
         current[keys[-1]] = value
 
     def _merge_config(
-        self, base: Dict[str, Any], override: Dict[str, Any]
-    ) -> Dict[str, Any]:
+        self, base: dict[str, Any], override: dict[str, Any]
+    ) -> dict[str, Any]:
         """Recursively merge configuration dictionaries."""
         result = copy.deepcopy(base)
 
@@ -272,14 +271,14 @@ def _merge_config(
 
         return result
 
-    def _validate_config(self, config_dict: Dict[str, Any]):
+    def _validate_config(self, config_dict: dict[str, Any]):
         """Validate configuration against JSON schema."""
         try:
             jsonschema.validate(config_dict, CONFIG_SCHEMA)
         except jsonschema.ValidationError as e:
             raise ConfigError(f"Configuration validation failed: {e.message}")
 
-    def _dict_to_config(self, config_dict: Dict[str, Any]) -> RMCPConfig:
+    def _dict_to_config(self, config_dict: dict[str, Any]) -> RMCPConfig:
         """Convert configuration dictionary to typed RMCPConfig object."""
         try:
             http_config = HTTPConfig(**config_dict.get("http", {}))
@@ -302,7 +301,7 @@ def _dict_to_config(self, config_dict: Dict[str, Any]) -> RMCPConfig:
 
 # Global configuration instance
 _config_loader = ConfigLoader()
-_global_config: Optional[RMCPConfig] = None
+_global_config: RMCPConfig | None = None
 
 
 def get_config(reload: bool = False) -> RMCPConfig:
@@ -324,8 +323,8 @@ def get_config(reload: bool = False) -> RMCPConfig:
 
 
 def load_config(
-    config_file: Optional[Union[str, Path]] = None,
-    overrides: Optional[Dict[str, Any]] = None,
+    config_file: str | Path | None = None,
+    overrides: dict[str, Any] | None = None,
     validate: bool = True,
 ) -> RMCPConfig:
     """