Validate document polling

Author: Git-Uzair

src/adeu/mcp_components/shared.py

@@ -7,8 +7,8 @@
 from typing import List
 
 # Centralized MCP Configuration
-FRONTEND_URL = os.environ.get("ADEU_FRONTEND_URL", "https://app.adeu.ai")
-BACKEND_URL = os.environ.get("ADEU_BACKEND_URL", "https://app.adeu.ai")
+FRONTEND_URL = os.environ.get("ADEU_FRONTEND_URL", "http://localhost:5173")
+BACKEND_URL = os.environ.get("ADEU_BACKEND_URL", "http://localhost:8000")
 MARKDOWN_UI_URI = "ui://adeu/markdown-ui"
 EMAIL_UI_URI = "ui://adeu/email-ui"
 

src/adeu/mcp_components/tools/validation.py

@@ -1,9 +1,10 @@
 # FILE: src/adeu/mcp_components/tools/validation.py
+import asyncio
 import json
 import urllib.error
 import urllib.request
 from pathlib import Path
-from typing import Annotated, List
+from typing import Annotated, List, Optional
 
 from fastmcp import Context
 from fastmcp.dependencies import Depends
@@ -21,118 +22,150 @@
 
 @tool(
     description=(
-        "Analyzes documents to find inconsistencies, contradictions, and risk assessments. "
-        "Always present the complete report to the user, including every verbatim evidence quote "
-        "exactly as returned, without summarizing or omitting any findings."
-        "Run this on validation request for files or directories."
+        "Validates documents for inconsistencies, contradictions, and risk assessments. "
+        "To START a new validation, provide 'file_paths'. This will immediately return a task_id. "
+        "To CHECK the status of a validation, call this tool AGAIN and provide ONLY the 'task_id'. "
+        "The checking process will poll for up to 50 seconds. If it times out, continue checking."
     ),
     timeout=300.0,
     annotations={"openWorldHint": True},
     meta={"ui": {"resourceUri": MARKDOWN_UI_URI}},
 )
 async def validate_documents(
-    file_paths: Annotated[List[str], "List of absolute paths to documents (DOCX, PDF) OR directories."],
     ctx: Context,
+    file_paths: Annotated[
+        Optional[List[str]],
+        "List of absolute paths to documents (DOCX, PDF) OR directories to start a new job.",
+    ] = None,
+    task_id: Annotated[Optional[int], "If resuming a pending check, provide the task ID here."] = None,
     api_key: str = Depends(get_cloud_auth_token),
 ) -> ToolResult:
-    await ctx.info("Starting document validation", extra={"provided_paths": file_paths})
-
-    if not file_paths:
-        await ctx.warning("No file paths provided by client")
-        raise ToolError("You must provide at least 1 file path or directory to perform document validation.")
-
-    resolved_files: list[Path] = []
-    valid_extensions = {".docx", ".pdf"}
-
-    await ctx.debug("Resolving files and directories")
-    for path_str in file_paths:
-        p = Path(path_str)
-        if not p.exists():
-            await ctx.error("Path not found on disk", extra={"missing_path": path_str})
-            raise ToolError(f"Path not found on local disk: {path_str}")
-
-        if p.is_dir():
-            for child in p.iterdir():
-                if child.is_file() and child.suffix.lower() in valid_extensions:
-                    resolved_files.append(child)
-        elif p.is_file():
-            if p.suffix.lower() not in valid_extensions:
-                await ctx.warning("Unsupported file type skipped", extra={"file": path_str})
-                raise ToolError(f"Unsupported file type for {path_str}. Only .docx and .pdf are supported.")
-            resolved_files.append(p)
-
-    resolved_files = list(set(resolved_files))
-
-    if not resolved_files:
-        await ctx.error("No valid documents found in provided paths")
-        raise ToolError("No supported documents (.docx or .pdf) were found in the provided paths.")
-
-    await ctx.info(
-        f"Resolved {len(resolved_files)} file(s) for validation",
-        extra={"files": [p.name for p in resolved_files]},
-    )
-
-    files_data = []
-    for p in resolved_files:
-        try:
+    if not file_paths and not task_id:
+        raise ToolError(
+            "You must provide either 'file_paths' to start a new validation, or 'task_id' to check an existing one."
+        )
+
+    # ==========================================
+    # PHASE 1: INIT (Upload and get task_id)
+    # ==========================================
+    if file_paths:
+        await ctx.info(
+            "Starting new document validation task",
+            extra={"provided_paths": file_paths},
+        )
+        resolved_files: list[Path] = []
+        valid_extensions = {".docx", ".pdf"}
+
+        for path_str in file_paths:
+            p = Path(path_str)
+            if not p.exists():
+                raise ToolError(f"Path not found on local disk: {path_str}")
+
+            if p.is_dir():
+                for child in p.iterdir():
+                    if child.is_file() and child.suffix.lower() in valid_extensions:
+                        resolved_files.append(child)
+            elif p.is_file():
+                if p.suffix.lower() not in valid_extensions:
+                    raise ToolError(f"Unsupported file type for {path_str}. Only .docx and .pdf are supported.")
+                resolved_files.append(p)
+
+        resolved_files = list(set(resolved_files))
+        if not resolved_files:
+            raise ToolError("No supported documents (.docx or .pdf) were found in the provided paths.")
+
+        files_data = []
+        for p in resolved_files:
             with open(p, "rb") as f:
                 files_data.append(("files", p.name, f.read()))
-        except Exception as e:
-            await ctx.error("Failed to read file", extra={"filename": p.name, "error": str(e)})
-            raise ToolError(f"Failed to read file {p.name}: {str(e)}") from e
-
-    await ctx.debug("Encoding multipart/form-data payload")
-    body, content_type = _encode_multipart_formdata(files_data)
-    url = f"{BACKEND_URL}/api/v1/documents/validate"
-
-    req = urllib.request.Request(
-        url,
-        data=body,
-        headers={
-            "Authorization": f"Bearer {api_key}",
-            "Content-Type": content_type,
-            "Accept": "application/json",
-        },
-        method="POST",
-    )
-
-    try:
-        await ctx.info(
-            "Sending validation request to Adeu Cloud",
-            extra={"url": url, "payload_size_bytes": len(body)},
+
+        body, content_type = _encode_multipart_formdata(files_data)
+        url = f"{BACKEND_URL}/api/v1/documents/validate"
+
+        req = urllib.request.Request(
+            url,
+            data=body,
+            headers={
+                "Authorization": f"Bearer {api_key}",
+                "Content-Type": content_type,
+                "Accept": "application/json",
+            },
+            method="POST",
         )
-        with urllib.request.urlopen(req) as response:
-            data = json.loads(response.read().decode("utf-8"))
 
-            await ctx.debug("Received successful response from cloud API")
+        try:
+            response = await asyncio.to_thread(urllib.request.urlopen, req)
+            data = json.loads(response.read().decode("utf-8"))
+            new_task_id = data.get("task_id")
 
-            # The backend now provides the fully formatted markdown report
-            markdown_report = data.get("report_markdown", "No report generated.")
-            report_title = (
-                resolved_files[0].name
-                if len(resolved_files) == 1
-                else f"Validation Report ({len(resolved_files)} files)"
-            )
-            return ToolResult(
-                content=markdown_report,
-                structured_content={"markdown": markdown_report, "title": report_title},
+            msg = (
+                f"Validation task started successfully. Task ID: {new_task_id}. "
+                f"Please call `validate_documents` again immediately with "
+                f"task_id={new_task_id} to monitor the progress."
             )
-
-    except urllib.error.HTTPError as e:
-        if e.code == 401:
-            await ctx.warning("Cloud authentication expired during validation")
-            DesktopAuthManager.clear_api_key()
-            raise ToolError("Your authentication expired. Please call `login_to_adeu_cloud` to re-authenticate.") from e
-        elif e.code == 403:
-            await ctx.warning("Authorization Error: User lacks access to this tool")
-            raise ToolError("Authorization Error: You do not have access to use this tool.") from e
-
-        error_body = e.read().decode("utf-8")
-        await ctx.error(
-            "Cloud validation API failure",
-            extra={"status_code": e.code, "body": error_body},
+            await ctx.info(f"Task started: {new_task_id}")
+            return ToolResult(content=msg, structured_content={"status": "pending", "message": msg})
+
+        except urllib.error.HTTPError as e:
+            if e.code == 401:
+                DesktopAuthManager.clear_api_key()
+                raise ToolError(
+                    "Your authentication expired. Please call `login_to_adeu_cloud` to re-authenticate."
+                ) from e
+            error_body = e.read().decode("utf-8")
+            raise ToolError(f"Cloud analysis failed (HTTP {e.code}): {error_body}") from e
+        except Exception as e:
+            raise ToolError(f"Unexpected error: {str(e)}") from e
+
+    # ==========================================
+    # PHASE 2: POLL (Wait for completion)
+    # ==========================================
+    poll_url = f"{BACKEND_URL}/api/v1/documents/validate/{task_id}"
+
+    # Poll up to 10 times (5 seconds each) = 50 seconds total
+    for attempt in range(10):
+        req = urllib.request.Request(
+            poll_url,
+            headers={
+                "Authorization": f"Bearer {api_key}",
+                "Accept": "application/json",
+            },
         )
-        raise ToolError(f"Cloud analysis failed (HTTP {e.code}): {error_body}") from e
-    except Exception as e:
-        await ctx.error("Unexpected error communicating with Adeu Cloud", extra={"error": str(e)})
-        raise ToolError(f"Failed to communicate with Adeu Cloud: {str(e)}") from e
+
+        try:
+            response = await asyncio.to_thread(urllib.request.urlopen, req)
+            data = json.loads(response.read().decode("utf-8"))
+            status = data.get("status")
+
+            if status == "COMPLETED":
+                markdown_report = data.get("report_markdown", "No report generated.")
+                return ToolResult(
+                    content=markdown_report,
+                    structured_content={
+                        "markdown": markdown_report,
+                        "title": f"Validation Report #{task_id}",
+                        "status": "completed",
+                    },
+                )
+
+            if status == "FAILED":
+                error_msg = data.get("error", "Unknown internal error")
+                raise ToolError(f"Validation task failed on the server: {error_msg}")
+
+            await ctx.debug(f"Task {task_id} status is {status}. Attempt {attempt + 1}/10. Sleeping 5s.")
+
+        except urllib.error.HTTPError as e:
+            if e.code == 401:
+                DesktopAuthManager.clear_api_key()
+                raise ToolError("Your authentication expired. Please re-authenticate.") from e
+            error_body = e.read().decode("utf-8")
+            raise ToolError(f"Failed to check task status (HTTP {e.code}): {error_body}") from e
+        except Exception as e:
+            raise ToolError(f"Unexpected error checking task status: {str(e)}") from e
+
+        # Sleep 5 seconds before the next poll
+        await asyncio.sleep(5)
+
+    # If we reach here, the 50s timeout has been reached but it's still pending
+    msg = f"Task {task_id} is still processing. Please call `validate_documents` again with task_id={task_id}."
+    return ToolResult(content=msg, structured_content={"status": "pending", "message": msg})