fix: use new

Author: mmilanta

src/mcp_scan/MCPScanner.py

@@ -170,8 +170,6 @@ async def scan_server(self, server: ServerScanResult, inspect_only: bool = False
             )
 
             if not inspect_only:
-                logger.debug("Verifying server: %s", server.name)
-                result = await verify_server(result, base_url=self.base_url)
                 logger.debug("Checking if server has changed: %s", server.name)
                 result = await self.check_server_changed(result)
                 logger.debug("Checking whitelist for server: %s", server.name)
@@ -189,6 +187,9 @@ async def scan_path(self, path: str, inspect_only: bool = False) -> ScanPathResu
         for i, server in enumerate(path_result.servers):
             logger.debug("Scanning server %d/%d: %s", i + 1, len(path_result.servers), server.name)
             path_result.servers[i] = await self.scan_server(server, inspect_only)
+
+        logger.debug("Verifying server: %s", server.name)
+        path_result = await verify_server(path_result, base_url=self.base_url)
         path_result.cross_ref_result = await self.check_cross_references(path_result)
         await self.emit("path_scanned", path_result)
         return path_result

src/mcp_scan/models.py

@@ -153,22 +153,6 @@ class ServerScanResult(BaseModel):
     result: list[EntityScanResult] | None = None
     error: ScanError | None = None
 
-    @model_serializer
-    def serialize(self, _info):
-        entities_with_result = self.entities_with_result
-        prompts_with_result = entities_with_result[: len(self.prompts)]
-        resources_with_result = entities_with_result[len(self.prompts) : len(self.prompts) + len(self.resources)]
-        tools_with_result = entities_with_result[len(self.prompts) + len(self.resources) :]
-
-        return {
-            "name": self.name,
-            "server": self.server,
-            "prompts": prompts_with_result,
-            "resources": resources_with_result,
-            "tools": tools_with_result,
-            "error": self.error,
-        }
-
     @property
     def entities(self) -> list[Entity]:
         return self.prompts + self.resources + self.tools
@@ -191,3 +175,7 @@ class ScanPathResult(BaseModel):
     servers: list[ServerScanResult] = []
     error: ScanError | None = None
     cross_ref_result: CrossRefResult | None = None
+
+    @property
+    def entities(self) -> list[Entity]:
+        return sum([server.entities for server in self.servers], start=[],)

src/mcp_scan/verify_api.py

@@ -1,43 +1,22 @@
 import ast
 import json
-
+import rich
 import aiohttp
 
-from .models import EntityScanResult, ServerScanResult, entity_type_to_str
+from .models import EntityScanResult, ScanPathResult
 
 
-async def verify_server(server_scan_result: ServerScanResult, base_url: str) -> ServerScanResult:
-    result = server_scan_result.model_copy(deep=True)
-    if len(server_scan_result.entities) == 0:
-        return result
-    messages = [
-        {
-            "role": "system",
-            "content": (
-                f"{entity_type_to_str(entity).capitalize()} Name:{entity.name}\n"
-                f"{entity_type_to_str(entity).capitalize()} Description:{entity.description}"
-            ),
-        }
-        for entity in server_scan_result.entities
-    ]
+async def verify_server(scan_path: ScanPathResult, base_url: str) -> ScanPathResult:
+    if len(scan_path.entities) == 0:
+        return scan_path.model_copy(deep=True)
     url = base_url[:-1] if base_url.endswith("/") else base_url
-    url = url + "/api/v1/public/mcp"
+    url = url + "/api/v2/public/mcp"
     headers = {"Content-Type": "application/json"}
-    data = {
-        "messages": messages,
-    }
     try:
         async with aiohttp.ClientSession() as session:
-            async with session.post(url, headers=headers, data=json.dumps(data)) as response:
+            async with session.post(url, headers=headers, data=scan_path.model_dump_json()) as response:
                 if response.status == 200:
-                    response_content: dict = await response.json()
-                    result.result = [EntityScanResult(verified=True) for _ in messages]
-                    for error in response_content.get("errors", []):
-                        key = ast.literal_eval(error["key"])
-                        idx = key[1][0]
-                        result.result[idx].verified = False
-                        result.result[idx].status = "failed - " + " ".join(error["args"])
-                    return result
+                    return ScanPathResult.model_validate_json(await response.read())
                 else:
                     raise Exception(f"Error: {response.status} - {await response.text()}")
     except Exception as e:
@@ -46,5 +25,7 @@ async def verify_server(server_scan_result: ServerScanResult, base_url: str) ->
             errstr = errstr.splitlines()[0]
         except Exception:
             errstr = ""
-        result.result = [EntityScanResult(status="could not reach verification server " + errstr) for _ in messages]
+        result = scan_path.model_copy(deep=True)
+        for server in result.servers:
+            server.result = [EntityScanResult(status="could not reach verification server " + errstr) for _ in server.entities]
         return result