v2.1.0: Add 3 new batch/utility tools

- send_multiple_requests: Batch send for load testing

- clone_webhook: Copy webhook with all settings

- export_webhook_data: Export requests to JSON

- Updated README with new tools (24 total)

Author: zebbern

CHANGELOG.md

@@ -5,6 +5,21 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [2.1.0] - 2026-01-26
+
+### Added
+
+- **3 New Tools** (21 → 24 total):
+  - `send_multiple_requests` - Send batch of requests for load testing
+  - `clone_webhook` - Copy webhook with all settings to a new token
+  - `export_webhook_data` - Export all requests to JSON format
+- `post_raw()` method in HTTP client for absolute URLs
+
+### Changed
+
+- Updated README badge (21 → 24 tools)
+- Added "Batch & Utility" tools section to README
+
 ## [2.0.7] - 2026-01-26
 
 ### Changed

README.md

@@ -2,7 +2,7 @@
 
 [![PyPI](https://img.shields.io/pypi/v/webhook-mcp-server.svg)](https://pypi.org/project/webhook-mcp-server/)
 [![Python](https://img.shields.io/badge/python-3.10+-blue.svg)](https://www.python.org/)
-[![MCP](https://img.shields.io/badge/MCP-21%20tools-brightgreen.svg)](https://modelcontextprotocol.io/)
+[![MCP](https://img.shields.io/badge/MCP-24%20tools-brightgreen.svg)](https://modelcontextprotocol.io/)
 [![License](https://img.shields.io/badge/license-MIT-blue.svg)](LICENSE)
 
 A Model Context Protocol (MCP) server for [webhook.site](https://webhook.site) - instantly capture HTTP requests, emails, and DNS lookups. Perfect for testing webhooks, debugging API callbacks, security testing, and bug bounty hunting.
@@ -134,6 +134,14 @@ Add to `claude_desktop_config.json`:
 | `check_for_callbacks`        | Quick check for OOB callbacks                        |
 | `extract_links_from_request` | Extract URLs from captured requests                  |
 
+### Batch & Utility
+
+| Tool                     | Description                                    |
+| ------------------------ | ---------------------------------------------- |
+| `send_multiple_requests` | Send batch of requests for load testing        |
+| `clone_webhook`          | Copy webhook with all settings                 |
+| `export_webhook_data`    | Export all requests to JSON                    |
+
 ---
 
 ## Examples

handlers/tool_handlers.py

@@ -118,6 +118,10 @@ def _get_handler(
             "generate_xss_callback": self._handle_generate_xss_callback,
             "generate_canary_token": self._handle_generate_canary_token,
             "extract_links_from_request": self._handle_extract_links_from_request,
+            # Batch & utility tools
+            "send_multiple_requests": self._handle_send_multiple_requests,
+            "clone_webhook": self._handle_clone_webhook,
+            "export_webhook_data": self._handle_export_webhook_data,
         }
         return handlers.get(name)
 
@@ -370,3 +374,31 @@ async def _handle_extract_links_from_request(self, arguments: dict[str, Any]):
             request_id=arguments.get("request_id"),
             filter_domain=arguments.get("filter_domain"),
         )
+    
+    # =========================================================================
+    # Batch & Utility Tool Handlers
+    # =========================================================================
+    
+    async def _handle_send_multiple_requests(self, arguments: dict[str, Any]) -> ToolResult:
+        """Handle send_multiple_requests tool."""
+        self._validate_webhook_token(arguments["webhook_token"])
+        return await self._request_service.send_multiple(
+            webhook_token=arguments["webhook_token"],
+            payloads=arguments["payloads"],
+            delay_ms=arguments.get("delay_ms", 0),
+        )
+    
+    async def _handle_clone_webhook(self, arguments: dict[str, Any]) -> ToolResult:
+        """Handle clone_webhook tool."""
+        self._validate_webhook_token(arguments["source_token"])
+        return await self._webhook_service.clone_webhook(
+            source_token=arguments["source_token"],
+        )
+    
+    async def _handle_export_webhook_data(self, arguments: dict[str, Any]) -> ToolResult:
+        """Handle export_webhook_data tool."""
+        self._validate_webhook_token(arguments["webhook_token"])
+        return await self._request_service.export_requests(
+            webhook_token=arguments["webhook_token"],
+            limit=arguments.get("limit", 100),
+        )

models/schemas.py

@@ -627,4 +627,64 @@ def to_json(self) -> str:
             "required": ["webhook_token"]
         }
     ),
+    # =========================================================================
+    # BATCH & UTILITY TOOLS
+    # =========================================================================
+    Tool(
+        name="send_multiple_requests",
+        description="Send multiple requests to a webhook in batch. Useful for load testing or sending multiple test payloads at once.",
+        inputSchema={
+            "type": "object",
+            "properties": {
+                "webhook_token": {
+                    "type": "string",
+                    "description": "The webhook token (UUID) from webhook.site"
+                },
+                "payloads": {
+                    "type": "array",
+                    "items": {"type": "object"},
+                    "description": "Array of JSON payloads to send to the webhook"
+                },
+                "delay_ms": {
+                    "type": "integer",
+                    "description": "Delay between requests in milliseconds (default: 0)",
+                    "default": 0
+                }
+            },
+            "required": ["webhook_token", "payloads"]
+        }
+    ),
+    Tool(
+        name="clone_webhook",
+        description="Clone an existing webhook with all its settings. Creates a new webhook with the same configuration (status, content, CORS, timeout) as the source.",
+        inputSchema={
+            "type": "object",
+            "properties": {
+                "source_token": {
+                    "type": "string",
+                    "description": "The webhook token (UUID) to clone"
+                }
+            },
+            "required": ["source_token"]
+        }
+    ),
+    Tool(
+        name="export_webhook_data",
+        description="Export all captured requests from a webhook to JSON format. Includes full request details: headers, body, IP, timestamp, user agent.",
+        inputSchema={
+            "type": "object",
+            "properties": {
+                "webhook_token": {
+                    "type": "string",
+                    "description": "The webhook token (UUID) from webhook.site"
+                },
+                "limit": {
+                    "type": "integer",
+                    "description": "Maximum number of requests to export (default: 100)",
+                    "default": 100
+                }
+            },
+            "required": ["webhook_token"]
+        }
+    ),
 ]

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "webhook-mcp-server"
-version = "2.0.7"
+version = "2.1.0"
 description = "MCP Server for webhook.site API integration with layered architecture"
 readme = "README.md"
 license = {text = "MIT"}

services/request_service.py

@@ -441,6 +441,118 @@ async def wait_for_email(
             }
         )
 
+    async def send_multiple(
+        self,
+        webhook_token: str,
+        payloads: list[dict[str, Any]],
+        delay_ms: int = 0,
+    ) -> ToolResult:
+        """Send multiple requests to a webhook in batch.
+        
+        Useful for load testing or sending test payloads.
+        
+        Args:
+            webhook_token: The webhook UUID
+            payloads: List of JSON payloads to send
+            delay_ms: Delay between requests in milliseconds (default: 0)
+            
+        Returns:
+            ToolResult with success/failure counts
+        """
+        from utils.http_client import WEBHOOK_SITE_API
+        
+        results = []
+        success_count = 0
+        fail_count = 0
+        
+        for i, payload in enumerate(payloads):
+            try:
+                response = await self._client.post_raw(
+                    f"{WEBHOOK_SITE_API}/{webhook_token}",
+                    json=payload
+                )
+                results.append({
+                    "index": i,
+                    "success": True,
+                    "status_code": response.status_code
+                })
+                success_count += 1
+            except Exception as e:
+                results.append({
+                    "index": i,
+                    "success": False,
+                    "error": str(e)
+                })
+                fail_count += 1
+            
+            # Add delay between requests if specified
+            if delay_ms > 0 and i < len(payloads) - 1:
+                await asyncio.sleep(delay_ms / 1000.0)
+        
+        return ToolResult(
+            success=fail_count == 0,
+            message=f"Sent {success_count}/{len(payloads)} requests successfully",
+            data={
+                "total": len(payloads),
+                "success_count": success_count,
+                "fail_count": fail_count,
+                "results": results
+            }
+        )
+    
+    async def export_requests(
+        self,
+        webhook_token: str,
+        limit: int = 100,
+        format: str = "json",
+    ) -> ToolResult:
+        """Export all requests from a webhook to JSON.
+        
+        Args:
+            webhook_token: The webhook UUID
+            limit: Maximum number of requests to export (default: 100)
+            format: Export format (currently only 'json' supported)
+            
+        Returns:
+            ToolResult with exported data
+        """
+        data = await self._client.get(
+            f"/token/{webhook_token}/requests",
+            params={"per_page": limit},
+        )
+        
+        requests_data = data.get("data", [])
+        
+        # Format each request with full details
+        export_data = []
+        for req in requests_data:
+            export_data.append({
+                "uuid": req.get("uuid"),
+                "type": req.get("type"),
+                "method": req.get("method"),
+                "url": req.get("url"),
+                "ip": req.get("ip"),
+                "hostname": req.get("hostname"),
+                "headers": req.get("headers", {}),
+                "query": req.get("query", {}),
+                "content": req.get("content"),
+                "text_content": req.get("text_content"),
+                "created_at": req.get("created_at"),
+                "user_agent": self._extract_header(req, "User-Agent"),
+                "content_type": self._extract_header(req, "Content-Type"),
+            })
+        
+        return ToolResult(
+            success=True,
+            message=f"Exported {len(export_data)} requests",
+            data={
+                "webhook_token": webhook_token,
+                "export_format": format,
+                "request_count": len(export_data),
+                "requests": export_data
+            }
+        )
+    
     @staticmethod
     def _extract_header(req: dict[str, Any], header_name: str) -> str:
         """Extract a header value from a request."""

services/webhook_service.py

@@ -351,3 +351,72 @@ async def get_dns(self, webhook_token: str, validate: bool = False) -> ToolResul
                 "url": url,
             }
         )
+
+    async def clone_webhook(self, source_token: str) -> ToolResult:
+        """Clone an existing webhook with all its settings.
+        
+        Creates a new webhook with the same configuration as the source.
+        
+        Args:
+            source_token: The webhook UUID to clone
+            
+        Returns:
+            ToolResult with new webhook token and copied settings
+        """
+        try:
+            # Get source webhook info
+            response = await self._client.get(f"/token/{source_token}")
+            response.raise_for_status()
+            source_data = response.json()
+            
+            # Create new webhook with same config
+            config = WebhookConfig(
+                default_status=source_data.get("default_status"),
+                default_content=source_data.get("default_content"),
+                default_content_type=source_data.get("default_content_type"),
+                timeout=source_data.get("timeout"),
+                cors=source_data.get("cors"),
+            )
+            
+            # Create the cloned webhook
+            create_response = await self._client.post("/token", json=config.to_payload())
+            create_response.raise_for_status()
+            new_data = create_response.json()
+            
+            new_token = new_data.get("uuid")
+            urls = self._build_webhook_urls(new_token)
+            
+            return ToolResult(
+                success=True,
+                message=f"Webhook cloned! New URL: {urls['url']}",
+                data={
+                    "source_token": source_token,
+                    "new_token": new_token,
+                    **urls,
+                    "cloned_settings": {
+                        "default_status": source_data.get("default_status"),
+                        "default_content": source_data.get("default_content"),
+                        "default_content_type": source_data.get("default_content_type"),
+                        "timeout": source_data.get("timeout"),
+                        "cors": source_data.get("cors"),
+                    }
+                }
+            )
+        except WebhookApiError as e:
+            if e.status_code == 404:
+                return ToolResult(
+                    success=False,
+                    message=f"Source token '{source_token}' not found or expired",
+                    data=None
+                )
+            return ToolResult(
+                success=False,
+                message=f"Failed to clone webhook: {str(e)}",
+                data=None
+            )
+        except Exception as e:
+            return ToolResult(
+                success=False,
+                message=f"Failed to clone webhook: {str(e)}",
+                data=None
+            )

utils/http_client.py

@@ -226,3 +226,34 @@ async def delete(
             return response.status_code
         except httpx.RequestError as e:
             raise WebhookApiError(f"DELETE request failed: {str(e)}") from e
+    
+    async def post_raw(
+        self,
+        url: str,
+        json: dict[str, Any] | None = None,
+        headers: dict[str, str] | None = None,
+    ) -> httpx.Response:
+        """Perform POST request to an absolute URL (not using base_url).
+        
+        Useful for posting to webhook endpoints directly.
+        
+        Args:
+            url: Full URL to POST to
+            json: JSON body data
+            headers: Additional headers
+            
+        Returns:
+            Full httpx Response object
+            
+        Raises:
+            WebhookApiError: On HTTP or API errors
+        """
+        try:
+            response = await self.client.post(
+                url,
+                json=json,
+                headers=headers,
+            )
+            return response
+        except httpx.RequestError as e:
+            raise WebhookApiError(f"POST request failed: {str(e)}") from e