Phase 4: Code quality - DRY refactoring and constants

- Add _build_webhook_urls() and _validate_token_exists() helpers to webhook_service.py

- Add named constants (DEFAULT_REQUEST_LIMIT, DEFAULT_TIMEOUT_SECONDS, POLL_INTERVAL_SECONDS) to request_service.py

- Add proper type hints to tool_handlers.py _get_handler() method

- Eliminates ~80+ lines of duplicate code

Author: zebbern

handlers/tool_handlers.py

@@ -8,11 +8,11 @@
 from __future__ import annotations
 
 import logging
-from typing import Any
+from typing import Any, Callable, Coroutine
 
 from mcp.types import TextContent
 
-from models.schemas import WebhookConfig, SearchFilters, DeleteFilters
+from models.schemas import WebhookConfig, SearchFilters, DeleteFilters, ToolResult
 from services.webhook_service import WebhookService
 from services.request_service import RequestService
 from services.bugbounty_service import BugBountyService
@@ -84,7 +84,9 @@ async def handle(
                 "An unexpected error occurred. Please report this issue."
             )
 
-    def _get_handler(self, name: str):
+    def _get_handler(
+        self, name: str
+    ) -> Callable[[dict[str, Any]], Coroutine[Any, Any, ToolResult]] | None:
         """Get the handler method for a tool name.
         
         Args:
@@ -93,7 +95,7 @@ def _get_handler(self, name: str):
         Returns:
             Handler coroutine or None if not found
         """
-        handlers = {
+        handlers: dict[str, Callable[[dict[str, Any]], Coroutine[Any, Any, ToolResult]]] = {
             "create_webhook": self._handle_create_webhook,
             "create_webhook_with_config": self._handle_create_webhook_with_config,
             "send_to_webhook": self._handle_send_to_webhook,

services/request_service.py

@@ -14,6 +14,11 @@
 from models.schemas import SearchFilters, DeleteFilters, ToolResult
 from utils.http_client import WebhookHttpClient
 
+# Constants for request handling
+DEFAULT_REQUEST_LIMIT = 10
+DEFAULT_TIMEOUT_SECONDS = 60
+POLL_INTERVAL_SECONDS = 2.0
+
 
 class RequestService:
     """Service for webhook request operations.
@@ -37,7 +42,7 @@ def __init__(self, client: WebhookHttpClient) -> None:
     async def get_all(
         self,
         webhook_token: str,
-        limit: int = 10,
+        limit: int = DEFAULT_REQUEST_LIMIT,
         request_type: str | None = None,
     ) -> ToolResult:
         """Get all requests sent to a webhook.
@@ -232,7 +237,7 @@ def _format_request(req: dict[str, Any]) -> dict[str, Any]:
     async def wait_for_request(
         self,
         webhook_token: str,
-        timeout_seconds: int = 60,
+        timeout_seconds: int = DEFAULT_TIMEOUT_SECONDS,
         request_type: str | None = None,
     ) -> ToolResult:
         """Wait for a new HTTP request to be received by the webhook.
@@ -251,7 +256,7 @@ async def wait_for_request(
         from utils.http_client import WebhookApiError
 
         start_time = time.time()
-        poll_interval = 2.0  # Poll every 2 seconds
+        poll_interval = POLL_INTERVAL_SECONDS
         max_retries = 3
         retry_count = 0
 
@@ -321,7 +326,7 @@ async def wait_for_request(
     async def wait_for_email(
         self,
         webhook_token: str,
-        timeout_seconds: int = 60,
+        timeout_seconds: int = DEFAULT_TIMEOUT_SECONDS,
         extract_links: bool = True,
     ) -> ToolResult:
         """Wait for an email to be received at the webhook's email address.
@@ -342,7 +347,7 @@ async def wait_for_email(
         from utils.http_client import WebhookApiError
 
         start_time = time.time()
-        poll_interval = 2.0  # Poll every 2 seconds
+        poll_interval = POLL_INTERVAL_SECONDS
         max_retries = 3
         retry_count = 0
 

services/webhook_service.py

@@ -32,6 +32,56 @@ def __init__(self, client: WebhookHttpClient) -> None:
         """
         self._client = client
 
+    def _build_webhook_urls(self, token: str, alias: str | None = None) -> dict[str, str]:
+        """Build all URL variants for a webhook token.
+        
+        Args:
+            token: The webhook UUID
+            alias: Optional custom alias
+            
+        Returns:
+            Dict with url, subdomain_url, api_url, email, and dns keys
+        """
+        identifier = alias if alias else token
+        return {
+            "url": f"{WEBHOOK_SITE_API}/{identifier}",
+            "subdomain_url": f"https://{token}.webhook.site",
+            "api_url": f"{WEBHOOK_SITE_API}/token/{token}",
+            "email": f"{token}@email.webhook.site",
+            "dns": f"{token}.dnshook.site",
+        }
+    
+    async def _validate_token_exists(self, webhook_token: str) -> ToolResult | None:
+        """Validate that a webhook token exists.
+        
+        Args:
+            webhook_token: The webhook UUID to validate
+            
+        Returns:
+            None if token is valid, ToolResult with error if invalid
+        """
+        try:
+            await self._client.get(f"/token/{webhook_token}")
+            return None  # Token is valid
+        except WebhookApiError as e:
+            if e.status_code == 404:
+                return ToolResult(
+                    success=False,
+                    message=f"Token '{webhook_token}' not found or expired",
+                    data=None
+                )
+            return ToolResult(
+                success=False,
+                message=f"Failed to validate token: {str(e)}",
+                data=None
+            )
+        except Exception as e:
+            return ToolResult(
+                success=False,
+                message=f"Failed to validate token: {str(e)}",
+                data=None
+            )
+    
     async def create(self) -> ToolResult:
         """Create a new webhook with default settings.
         
@@ -44,23 +94,15 @@ async def create(self) -> ToolResult:
 
         token = data.get("uuid")
         alias = data.get("alias")
-        url = f"{WEBHOOK_SITE_API}/{alias if alias else token}"
-        subdomain_url = f"https://{token}.webhook.site"
-        email = f"{token}@email.webhook.site"
-        api_url = f"{WEBHOOK_SITE_API}/token/{token}"
-        dns = f"{token}.dnshook.site"
+        urls = self._build_webhook_urls(token, alias)
 
         return ToolResult(
             success=True,
-            message=f"Webhook created! Send requests to: {url}",
+            message=f"Webhook created! Send requests to: {urls['url']}",
             data={
                 "token": token,
                 "alias": alias,
-                "url": url,
-                "subdomain_url": subdomain_url,
-                "api_url": api_url,
-                "email": email,
-                "dns": dns,
+                **urls,
                 "expires_at": data.get("expires_at"),
                 "default_status": data.get("default_status"),
                 "default_content": data.get("default_content"),
@@ -87,23 +129,15 @@ async def create_with_config(self, config: WebhookConfig) -> ToolResult:
 
         token = data.get("uuid")
         alias = data.get("alias")
-        url = f"{WEBHOOK_SITE_API}/{alias if alias else token}"
-        subdomain_url = f"https://{token}.webhook.site"
-        email = f"{token}@email.webhook.site"
-        api_url = f"{WEBHOOK_SITE_API}/token/{token}"
-        dns = f"{token}.dnshook.site"
+        urls = self._build_webhook_urls(token, alias)
 
         return ToolResult(
             success=True,
-            message=f"Webhook created with custom config! URL: {url}",
+            message=f"Webhook created with custom config! URL: {urls['url']}",
             data={
                 "token": token,
                 "alias": alias,
-                "url": url,
-                "subdomain_url": subdomain_url,
-                "api_url": api_url,
-                "email": email,
-                "dns": dns,
+                **urls,
                 "default_status": data.get("default_status"),
                 "default_content": data.get("default_content"),
                 "default_content_type": data.get("default_content_type"),
@@ -241,27 +275,9 @@ async def get_url(self, webhook_token: str, validate: bool = False) -> ToolResul
             ToolResult with token and full URL
         """
         if validate:
-            try:
-                await self._client.get(f"/token/{webhook_token}")
-                # If we get here, token is valid
-            except WebhookApiError as e:
-                if e.status_code == 404:
-                    return ToolResult(
-                        success=False,
-                        message=f"Token '{webhook_token}' not found or expired",
-                        data=None
-                    )
-                return ToolResult(
-                    success=False,
-                    message=f"Failed to validate token: {str(e)}",
-                    data=None
-                )
-            except Exception as e:
-                return ToolResult(
-                    success=False,
-                    message=f"Failed to validate token: {str(e)}",
-                    data=None
-                )
+            validation_error = await self._validate_token_exists(webhook_token)
+            if validation_error:
+                return validation_error
 
         url = f"{WEBHOOK_SITE_API}/{webhook_token}"
 
@@ -287,27 +303,9 @@ async def get_email(self, webhook_token: str, validate: bool = False) -> ToolRes
             ToolResult with token, email address, and URL
         """
         if validate:
-            try:
-                await self._client.get(f"/token/{webhook_token}")
-                # If we get here, token is valid
-            except WebhookApiError as e:
-                if e.status_code == 404:
-                    return ToolResult(
-                        success=False,
-                        message=f"Token '{webhook_token}' not found or expired",
-                        data=None
-                    )
-                return ToolResult(
-                    success=False,
-                    message=f"Failed to validate token: {str(e)}",
-                    data=None
-                )
-            except Exception as e:
-                return ToolResult(
-                    success=False,
-                    message=f"Failed to validate token: {str(e)}",
-                    data=None
-                )
+            validation_error = await self._validate_token_exists(webhook_token)
+            if validation_error:
+                return validation_error
 
         email = f"{webhook_token}@email.webhook.site"
         url = f"{WEBHOOK_SITE_API}/{webhook_token}"
@@ -336,27 +334,9 @@ async def get_dns(self, webhook_token: str, validate: bool = False) -> ToolResul
             ToolResult with token, DNS domain, example subdomain usage, and URL
         """
         if validate:
-            try:
-                await self._client.get(f"/token/{webhook_token}")
-                # If we get here, token is valid
-            except WebhookApiError as e:
-                if e.status_code == 404:
-                    return ToolResult(
-                        success=False,
-                        message=f"Token '{webhook_token}' not found or expired",
-                        data=None
-                    )
-                return ToolResult(
-                    success=False,
-                    message=f"Failed to validate token: {str(e)}",
-                    data=None
-                )
-            except Exception as e:
-                return ToolResult(
-                    success=False,
-                    message=f"Failed to validate token: {str(e)}",
-                    data=None
-                )
+            validation_error = await self._validate_token_exists(webhook_token)
+            if validation_error:
+                return validation_error
 
         dns_domain = f"{webhook_token}.dnshook.site"
         url = f"{WEBHOOK_SITE_API}/{webhook_token}"