""" API testing and advanced parsing tools implementation """ import asyncio import json import time from typing import Any, Dict, List from mcp.types import TextContent from ..utils.api_client import get_api_client from ..utils.http_client import get_http_client from ..utils.logging import log_tool_end, log_tool_start from ..utils.plugins import get_plugin_manager async def api_test_tool(args: dict[str, Any]) -> list[TextContent]: """Test API endpoints with various configurations.""" start_time = log_tool_start("api_test", args) try: url = args["url"] method = args.get("method", "GET").upper() headers = args.get("headers", {}) body = args.get("body") body_type = args.get("body_type", "json") params = args.get("params", {}) auth = args.get("auth", {}) timeout = args.get("timeout", 30) follow_redirects = args.get("follow_redirects", True) validate_ssl = args.get("validate_ssl", True) expected_status = args.get("expected_status") response_format = args.get("response_format", "auto") # Prepare authentication auth_headers = {} if auth.get("type") == "basic" and auth.get("username"): import base64 credentials = base64.b64encode( f"{auth['username']}:{auth.get('password', '')}".encode() ).decode() auth_headers["Authorization"] = f"Basic {credentials}" elif auth.get("type") == "bearer" and auth.get("token"): auth_headers["Authorization"] = f"Bearer {auth['token']}" elif auth.get("type") == "api_key" and auth.get("api_key"): header_name = auth.get("api_key_header", "X-API-Key") auth_headers[header_name] = auth["api_key"] # Merge headers headers.update(auth_headers) # Prepare request body request_data = None if body: if body_type == "json": try: request_data = json.loads(body) except json.JSONDecodeError: request_data = body elif body_type == "form": try: request_data = json.loads(body) except json.JSONDecodeError: request_data = body else: request_data = body # Create custom API client for this test api_client = get_api_client() # Make request if method == "GET": response = api_client.get( url, timeout=timeout, headers=headers, params=params, use_cache=False ) elif method == "POST": response = api_client.post(url, data=request_data, headers=headers, timeout=timeout) else: # For other methods, use a generic approach import requests session = requests.Session() session.headers.update(headers) if method == "PUT": response = session.put(url, json=request_data, timeout=timeout) elif method == "DELETE": response = session.delete(url, timeout=timeout) elif method == "PATCH": response = session.patch(url, json=request_data, timeout=timeout) elif method == "HEAD": response = session.head(url, timeout=timeout) elif method == "OPTIONS": response = session.options(url, timeout=timeout) else: response = session.request(method, url, json=request_data, timeout=timeout) # Convert to our format content = "" json_data = None try: if "application/json" in response.headers.get("content-type", ""): json_data = response.json() content = json.dumps(json_data, indent=2, ensure_ascii=False) else: content = response.text except: content = response.text response = { "method": method, "url": url, "status_code": response.status_code, "content_type": response.headers.get("content-type", ""), "content": content, "json_data": json_data, "size": len(content), "response_time_seconds": 0, # Would need timing "response_headers": dict(response.headers), "success": response.status_code < 400, } # Validate response validation_errors = [] if expected_status and response.get("status_code") != expected_status: validation_errors.append( f"Expected status {expected_status}, got {response.get('status_code')}" ) if response_format != "auto": content_type = response.get("content_type", "") if response_format == "json" and "json" not in content_type.lower(): validation_errors.append(f"Expected JSON response, got {content_type}") elif response_format == "xml" and "xml" not in content_type.lower(): validation_errors.append(f"Expected XML response, got {content_type}") elif response_format == "html" and "html" not in content_type.lower(): validation_errors.append(f"Expected HTML response, got {content_type}") # Prepare result result = { "request": { "url": url, "method": method, "headers": headers, "body": body, "params": params, "auth_type": auth.get("type", "none"), }, "response": response, "validation": { "passed": len(validation_errors) == 0, "errors": validation_errors, }, "test_metadata": { "timestamp": time.time(), "timeout": timeout, "validate_ssl": validate_ssl, "follow_redirects": follow_redirects, }, } log_tool_end("api_test", start_time, result, True) return [TextContent(type="text", text=json.dumps(result, indent=2, ensure_ascii=False))] except Exception as e: log_tool_end("api_test", start_time, {}, False, str(e)) return [TextContent(type="text", text=f"API test failed: {e}")] async def api_batch_test_tool(args: dict[str, Any]) -> list[TextContent]: """Run multiple API tests in batch.""" start_time = log_tool_start("api_batch_test", args) try: tests = args.get("tests", []) parallel = args.get("parallel", False) max_concurrent = args.get("max_concurrent", 5) stop_on_failure = args.get("stop_on_failure", False) if not tests: return [TextContent(type="text", text="No tests provided")] results = [] if parallel: # Run tests in parallel semaphore = asyncio.Semaphore(max_concurrent) async def run_single_test(test_config): async with semaphore: try: # Create test args test_args = { "url": test_config["url"], "method": test_config.get("method", "GET"), "expected_status": test_config.get("expected_status"), "timeout": test_config.get("timeout", 10), } # Add optional parameters for key in ["headers", "body", "body_type", "params", "auth"]: if key in test_config: test_args[key] = test_config[key] result = await api_test_tool(test_args) return { "test_name": test_config["name"], "result": json.loads(result[0].text), "success": True, } except Exception as e: return { "test_name": test_config["name"], "error": str(e), "success": False, } # Run all tests concurrently tasks = [run_single_test(test) for test in tests] batch_results = await asyncio.gather(*tasks) for i, result in enumerate(batch_results): results.append(result) if stop_on_failure and not result.get("success", False): break else: # Run tests sequentially for test_config in tests: try: test_args = { "url": test_config["url"], "method": test_config.get("method", "GET"), "expected_status": test_config.get("expected_status"), "timeout": test_config.get("timeout", 10), } for key in ["headers", "body", "body_type", "params", "auth"]: if key in test_config: test_args[key] = test_config[key] result = await api_test_tool(test_args) results.append( { "test_name": test_config["name"], "result": json.loads(result[0].text), "success": True, } ) except Exception as e: results.append( { "test_name": test_config["name"], "error": str(e), "success": False, } ) if stop_on_failure and not results[-1].get("success", False): break # Calculate summary total_tests = len(results) successful_tests = sum(1 for r in results if r.get("success", False)) failed_tests = total_tests - successful_tests summary = { "batch_summary": { "total_tests": total_tests, "successful": successful_tests, "failed": failed_tests, "success_rate": successful_tests / total_tests if total_tests > 0 else 0, "parallel_execution": parallel, "stopped_on_failure": stop_on_failure, }, "test_results": results, } log_tool_end("api_batch_test", start_time, summary, True) return [TextContent(type="text", text=json.dumps(summary, indent=2, ensure_ascii=False))] except Exception as e: log_tool_end("api_batch_test", start_time, {}, False, str(e)) return [TextContent(type="text", text=f"Batch API test failed: {e}")] async def parse_advanced_tool(args: dict[str, Any]) -> list[TextContent]: """Advanced content parsing with custom schemas.""" start_time = log_tool_start("parse_advanced", args) try: url = args.get("url") html = args.get("html") schema = args.get("schema", {}) output_format = args.get("output_format", "json") validate_schema = args.get("validate_schema", False) if not html and not url: return [TextContent(type="text", text="Either 'url' or 'html' parameter is required")] # Get HTML content if url and not html: api_client = get_api_client() response = api_client.get(url, use_cache=True) if not response.get("success"): return [ TextContent( type="text", text=f"Failed to fetch URL: {response.get('error')}", ) ] html = response.get("content", "") if not html: return [TextContent(type="text", text="No HTML content to parse")] # Parse HTML from bs4 import BeautifulSoup soup = BeautifulSoup(html, "lxml") extracted_data = {} fields = schema.get("fields", []) for field in fields: field_name = field["name"] selector = field["selector"] attribute = field.get("attribute") multiple = field.get("multiple", False) required = field.get("required", False) try: elements = soup.select(selector) if not elements: if required: extracted_data[field_name] = f"REQUIRED_FIELD_MISSING: {selector}" continue if multiple: values = [] for element in elements: if attribute: if attribute == "text": values.append(element.get_text(strip=True)) else: values.append(element.get(attribute, "")) else: values.append(element.get_text(strip=True)) extracted_data[field_name] = values else: element = elements[0] if attribute: if attribute == "text": extracted_data[field_name] = element.get_text(strip=True) else: extracted_data[field_name] = element.get(attribute, "") else: extracted_data[field_name] = element.get_text(strip=True) except Exception as e: extracted_data[field_name] = f"EXTRACTION_ERROR: {str(e)}" # Apply transformations transformations = schema.get("transformations", {}) for field_name, transform in transformations.items(): if field_name in extracted_data: value = extracted_data[field_name] if transform.get("type") == "trim" and isinstance(value, str): extracted_data[field_name] = value.strip() elif transform.get("type") == "lowercase" and isinstance(value, str): extracted_data[field_name] = value.lower() elif transform.get("type") == "uppercase" and isinstance(value, str): extracted_data[field_name] = value.upper() elif transform.get("type") == "number" and isinstance(value, str): try: extracted_data[field_name] = float(value.replace(",", "")) except: pass # Keep original value # Regex pattern extraction pattern = transform.get("pattern") if pattern and isinstance(value, str): import re match = re.search(pattern, value) if match: replacement = transform.get("replace", "") if replacement: extracted_data[field_name] = re.sub(pattern, replacement, value) else: extracted_data[field_name] = ( match.group(1) if match.groups() else match.group(0) ) # Validate schema if requested validation_errors = [] if validate_schema: for field in fields: if field.get("required", False): field_name = field["name"] if field_name not in extracted_data or not extracted_data[field_name]: validation_errors.append( f"Required field '{field_name}' is missing or empty" ) # Format output if output_format == "json": result = { "extracted_data": extracted_data, "schema_validation": { "passed": len(validation_errors) == 0, "errors": validation_errors, }, "metadata": { "total_fields": len(fields), "extracted_fields": len(extracted_data), "extraction_success_rate": len(extracted_data) / len(fields) if fields else 0, "source": "url" if url else "html", }, } output = json.dumps(result, indent=2, ensure_ascii=False) elif output_format == "csv": import csv import io if isinstance(extracted_data, dict): # Single record output = io.StringIO() writer = csv.DictWriter(output, fieldnames=extracted_data.keys()) writer.writeheader() writer.writerow(extracted_data) output = output.getvalue() else: # Multiple records output = "Multiple records not supported in CSV format" else: output = json.dumps(extracted_data, indent=2, ensure_ascii=False) log_tool_end( "parse_advanced", start_time, {"extracted_fields": len(extracted_data)}, True, ) return [TextContent(type="text", text=output)] except Exception as e: log_tool_end("parse_advanced", start_time, {}, False, str(e)) return [TextContent(type="text", text=f"Advanced parsing failed: {e}")] async def extract_dynamic_content_tool(args: dict[str, Any]) -> list[TextContent]: """Extract dynamically loaded content using browser automation.""" start_time = log_tool_start("extract_dynamic_content", args) try: url = args["url"] wait_for = args.get("wait_for") wait_time = args.get("wait_time", 2) interactions = args.get("interactions", []) extraction_rules = args.get("extraction_rules", {}) take_screenshot = args.get("take_screenshot", False) # Get browser client http_client = await get_http_client() # Navigate to page page = await http_client.ensure_session() response = await page.goto(url, wait_until="domcontentloaded") # Wait for specific element if specified if wait_for: try: await page.wait_for_selector(wait_for, timeout=10000) except Exception as e: return [TextContent(type="text", text=f"Wait for selector failed: {e}")] # Perform interactions for interaction in interactions: try: if interaction["type"] == "click": selector = interaction.get("selector") if selector: await page.click(selector, timeout=5000) elif interaction["type"] == "scroll": # Scroll to bottom or specific element await page.evaluate("window.scrollTo(0, document.body.scrollHeight)") elif interaction["type"] == "wait": await asyncio.sleep(interaction.get("duration", 1)) except Exception as e: return [TextContent(type="text", text=f"Interaction failed: {e}")] # Additional wait time if wait_time > 0: await asyncio.sleep(wait_time) # Take screenshot if requested screenshot_data = None if take_screenshot: try: screenshot_bytes = await page.screenshot(full_page=True) import base64 screenshot_data = base64.b64encode(screenshot_bytes).decode() except Exception as e: screenshot_data = f"Screenshot failed: {e}" # Extract content html = await page.content() # Use advanced parsing if rules provided if extraction_rules: parse_args = { "html": html, "schema": extraction_rules, "output_format": "json", } parse_result = await parse_advanced_tool(parse_args) extracted_data = json.loads(parse_result[0].text) else: # Simple HTML extraction extracted_data = { "html": html, "title": await page.title(), "url": page.url, } # Prepare result result = { "url": url, "final_url": page.url, "title": await page.title(), "extracted_data": extracted_data, "interactions_performed": len(interactions), "wait_time": wait_time, "metadata": { "dynamic_content": True, "browser_automation": True, "screenshot_taken": take_screenshot, }, } if screenshot_data: result["screenshot"] = screenshot_data log_tool_end("extract_dynamic_content", start_time, result, True) return [TextContent(type="text", text=json.dumps(result, indent=2, ensure_ascii=False))] except Exception as e: log_tool_end("extract_dynamic_content", start_time, {}, False, str(e)) return [TextContent(type="text", text=f"Dynamic content extraction failed: {e}")] async def load_plugin_tool(args: dict[str, Any]) -> list[TextContent]: """Load a plugin by name.""" start_time = log_tool_start("load_plugin", args) try: plugin_name = args["plugin_name"] config = args.get("config", {}) plugin_manager = get_plugin_manager() success = plugin_manager.load_plugin(plugin_name, config) result = {"plugin_name": plugin_name, "success": success, "config": config} if success: plugin = plugin_manager.get_plugin(plugin_name) if plugin: result["plugin_info"] = { "name": plugin.name, "version": plugin.version, "enabled": plugin.is_enabled(), "type": type(plugin).__name__, } log_tool_end("load_plugin", start_time, result, success) return [TextContent(type="text", text=json.dumps(result, indent=2, ensure_ascii=False))] except Exception as e: log_tool_end("load_plugin", start_time, {}, False, str(e)) return [TextContent(type="text", text=f"Plugin loading failed: {e}")] async def unload_plugin_tool(args: dict[str, Any]) -> list[TextContent]: """Unload a plugin by name.""" start_time = log_tool_start("unload_plugin", args) try: plugin_name = args["plugin_name"] plugin_manager = get_plugin_manager() success = plugin_manager.unload_plugin(plugin_name) result = {"plugin_name": plugin_name, "success": success} log_tool_end("unload_plugin", start_time, result, success) return [TextContent(type="text", text=json.dumps(result, indent=2, ensure_ascii=False))] except Exception as e: log_tool_end("unload_plugin", start_time, {}, False, str(e)) return [TextContent(type="text", text=f"Plugin unloading failed: {e}")] async def list_plugins_tool(args: dict[str, Any]) -> list[TextContent]: """List all loaded plugins.""" start_time = log_tool_start("list_plugins", args) try: plugin_manager = get_plugin_manager() plugins = plugin_manager.list_plugins() result = {"loaded_plugins": plugins, "total_count": len(plugins)} log_tool_end("list_plugins", start_time, result, True) return [TextContent(type="text", text=json.dumps(result, indent=2, ensure_ascii=False))] except Exception as e: log_tool_end("list_plugins", start_time, {}, False, str(e)) return [TextContent(type="text", text=f"Plugin listing failed: {e}")] async def discover_plugins_tool(args: dict[str, Any]) -> list[TextContent]: """Discover available plugins.""" start_time = log_tool_start("discover_plugins", args) try: plugin_manager = get_plugin_manager() available_plugins = plugin_manager.discover_plugins() result = { "available_plugins": available_plugins, "total_count": len(available_plugins), "plugin_directories": [str(d) for d in plugin_manager.plugin_dirs], } log_tool_end("discover_plugins", start_time, result, True) return [TextContent(type="text", text=json.dumps(result, indent=2, ensure_ascii=False))] except Exception as e: log_tool_end("discover_plugins", start_time, {}, False, str(e)) return [TextContent(type="text", text=f"Plugin discovery failed: {e}")] async def get_plugin_tools_tool(args: dict[str, Any]) -> list[TextContent]: """Get all tools provided by loaded plugins.""" start_time = log_tool_start("get_plugin_tools", args) try: plugin_manager = get_plugin_manager() plugin_tools = plugin_manager.get_tools_from_plugins() result = { "plugin_tools": list(plugin_tools.keys()), "total_count": len(plugin_tools), "tools": [ { "name": name, "module": func.__module__, "doc": func.__doc__ or "No description", } for name, func in plugin_tools.items() ], } log_tool_end("get_plugin_tools", start_time, result, True) return [TextContent(type="text", text=json.dumps(result, indent=2, ensure_ascii=False))] except Exception as e: log_tool_end("get_plugin_tools", start_time, {}, False, str(e)) return [TextContent(type="text", text=f"Plugin tools retrieval failed: {e}")]