fix/struggle in serializing exceptions (#57)

* fix: skip serialization of exceptions

* fix: minor

Author: mmilanta

src/mcp_scan/MCPScanner.py

@@ -128,29 +128,33 @@ async def get_servers_from_path(self, path: str) -> ScanPathResult:
 
     async def check_server_changed(self, server: ServerScanResult) -> ServerScanResult:
         logger.debug("Checking for changes in server: %s", server.name)
-        result = server.model_copy(deep=True)
+        output_server = server.clone()
+        if output_server.result is None:
+            raise ValueError("Server result is None, cannot check for changes")
         for i, (entity, entity_result) in enumerate(server.entities_with_result):
             if entity_result is None:
                 continue
             c, messages = self.storage_file.check_and_update(server.name or "", entity, entity_result.verified)
-            result.result[i].changed = c
+            output_server.result[i].changed = c
             if c:
                 logger.info("Entity %s in server %s has changed", entity.name, server.name)
-                result.result[i].messages.extend(messages)
-        return result
+                output_server.result[i].messages.extend(messages)
+        return output_server
 
     async def check_whitelist(self, server: ServerScanResult) -> ServerScanResult:
         logger.debug("Checking whitelist for server: %s", server.name)
-        result = server.model_copy()
+        output_server = server.clone()
+        if output_server.result is None:
+            raise ValueError("Server result is None, cannot check for changes")
         for i, (entity, entity_result) in enumerate(server.entities_with_result):
             if entity_result is None:
                 continue
             if self.storage_file.is_whitelisted(entity):
                 logger.debug("Entity %s is whitelisted", entity.name)
-                result.result[i].whitelisted = True
+                output_server.result[i].whitelisted = True
             else:
-                result.result[i].whitelisted = False
-        return result
+                output_server.result[i].whitelisted = False
+        return output_server
 
     async def emit(self, signal: str, data: Any):
         logger.debug("Emitting signal: %s", signal)
@@ -159,7 +163,7 @@ async def emit(self, signal: str, data: Any):
 
     async def scan_server(self, server: ServerScanResult, inspect_only: bool = False) -> ServerScanResult:
         logger.info("Scanning server: %s, inspect_only: %s", server.name, inspect_only)
-        result = server.model_copy(deep=True)
+        result = server.clone()
         try:
             result.signature = await check_server_with_timeout(
                 server.server, self.server_timeout, self.suppress_mcpserver_io

src/mcp_scan/models.py

@@ -129,6 +129,16 @@ def serialize_exception(self, exception: Exception | None, _info) -> str | None:
     def text(self) -> str:
         return self.message or (str(self.exception) or "")
 
+    def clone(self) -> "ScanError":
+        """
+        Create a copy of the ScanError instance. This is not the same as `model_copy(deep=True)`, because it does not
+        clone the exception. This is crucial to avoid issues with serialization of exceptions.
+        """
+        return ScanError(
+            message=self.message,
+            exception=self.exception,
+        )
+
 
 class EntityScanResult(BaseModel):
     model_config = ConfigDict()
@@ -190,6 +200,15 @@ def entities_with_result(self) -> list[tuple[Entity, EntityScanResult | None]]:
         else:
             return [(entity, None) for entity in self.entities]
 
+    def clone(self) -> "ServerScanResult":
+        """
+        Create a copy of the ServerScanResult instance. This is not the same as `model_copy(deep=True)`, because it does not
+        clone the error. This is crucial to avoid issues with serialization of exceptions.
+        """
+        output = self.model_copy(deep=True, update={"error": None})
+        output.error = self.error.clone() if self.error else None
+        return output
+
 
 class ScanPathResult(BaseModel):
     model_config = ConfigDict()
@@ -202,6 +221,15 @@ class ScanPathResult(BaseModel):
     def entities(self) -> list[Entity]:
         return list(chain.from_iterable(server.entities for server in self.servers))
 
+    def clone(self) -> "ScanPathResult":
+        """
+        Create a copy of the ScanPathResult instance. This is not the same as `model_copy(deep=True)`, because it does not
+        clone the error. This is crucial to avoid issues with serialization of exceptions.
+        """
+        output = self.model_copy(deep=True, update={"error": None})
+        output.error = self.error.clone() if self.error else None
+        return output
+
 
 def entity_to_tool(
     entity: Entity,

src/mcp_scan/verify_api.py

@@ -19,7 +19,7 @@
 
 
 async def verify_scan_path_public_api(scan_path: ScanPathResult, base_url: str) -> ScanPathResult:
-    output_path = scan_path.model_copy(deep=True)
+    output_path = scan_path.clone()
     url = base_url[:-1] if base_url.endswith("/") else base_url
     url = url + "/api/v1/public/mcp-scan"
     headers = {"Content-Type": "application/json"}
@@ -64,7 +64,7 @@ def get_policy() -> str:
 
 
 async def verify_scan_path_locally(scan_path: ScanPathResult) -> ScanPathResult:
-    output_path = scan_path.model_copy(deep=True)
+    output_path = scan_path.clone()
     tools_to_scan: list[Tool] = []
     for server in scan_path.servers:
         # None server signature are servers which are not reachable.