Scrapy MCP Server

Instructions

Input Schema

Input Schema (JSON Schema)

Implementation Reference

• extractor/server.py:870-1089 (handler)

  The primary handler function for the 'fill_and_submit_form' tool. It validates inputs, sets up Selenium or Playwright browser, uses FormHandler to fill the form, optionally submits it, and returns a ScrapeResponse with results including final URL and title.
  @app.tool() @timing_decorator async def fill_and_submit_form( url: Annotated[ str, Field( ..., description="包含表单的网页 URL，必须包含协议前缀（http:// 或 https://）", ), ], form_data: Annotated[ Dict[str, Any], Field( ..., description="""表单字段数据，格式为{"选择器": "值"}，支持各种表单元素。 示例：{"#username": "admin", "input[name=password]": "secret", "select[name=country]": "US", "input[type=checkbox]": True}""", ), ], submit: Annotated[bool, Field(default=False, description="是否提交表单")], submit_button_selector: Annotated[ Optional[str], Field( default=None, description="""提交按钮的CSS选择器，如未指定则尝试自动查找。 示例："button[type=submit]"、"#submit-btn\"""", ), ], method: Annotated[ str, Field( default="selenium", description="""自动化方法选择，可选值： "selenium"（使用Selenium WebDriver）、 "playwright"（使用Playwright浏览器自动化）""", ), ], wait_for_element: Annotated[ Optional[str], Field( default=None, description="""表单填写前等待加载的元素CSS选择器。 示例：".form-container"、"#login-form\"""", ), ], ) -> ScrapeResponse: """ Fill and optionally submit a form on a webpage. This tool can handle various form elements including: - Text inputs - Checkboxes and radio buttons - Dropdown selects - File uploads - Form submission Useful for interacting with search forms, contact forms, login forms, etc. Returns: ScrapeResponse object containing success status, form interaction results, and optional submission response. Supports complex form automation workflows. """ try: from .utils import URLValidator # Validate inputs if not URLValidator.is_valid_url(url): return ScrapeResponse( success=False, url=url, method=method, error="Invalid URL format", ) if method not in ["selenium", "playwright"]: return ScrapeResponse( success=False, url=url, method=method, error="Method must be one of: selenium, playwright", ) start_time = time.time() logger.info(f"Form interaction for: {url}") # Apply rate limiting await rate_limiter.wait() # Setup browser based on method if method == "selenium": from selenium import webdriver from selenium.webdriver.chrome.options import Options as ChromeOptions options = ChromeOptions() if settings.browser_headless: options.add_argument("--headless") options.add_argument("--no-sandbox") options.add_argument("--disable-dev-shm-usage") driver = webdriver.Chrome(options=options) try: driver.get(url) # Wait for element if specified if wait_for_element: from selenium.webdriver.common.by import By from selenium.webdriver.support import expected_conditions as EC from selenium.webdriver.support.ui import WebDriverWait WebDriverWait(driver, settings.browser_timeout).until( EC.presence_of_element_located( (By.CSS_SELECTOR, wait_for_element) ) ) # Fill and submit form form_handler = FormHandler(driver) result = await form_handler.fill_form( form_data=form_data, submit=submit, submit_button_selector=submit_button_selector, ) # Get final page info final_url = driver.current_url final_title = driver.title finally: driver.quit() elif method == "playwright": from playwright.async_api import async_playwright playwright = await async_playwright().start() try: browser = await playwright.chromium.launch( headless=settings.browser_headless ) context = await browser.new_context() page = await context.new_page() await page.goto(url, timeout=60000) # Wait for element if specified if wait_for_element: await page.wait_for_selector( wait_for_element, timeout=settings.browser_timeout * 1000, ) # Fill and submit form form_handler = FormHandler(page) result = await form_handler.fill_form( form_data=form_data, submit=submit, submit_button_selector=submit_button_selector, ) # Get final page info final_url = page.url final_title = await page.title() finally: await browser.close() await playwright.stop() duration_ms = int((time.time() - start_time) * 1000) if result.get("success"): metrics_collector.record_request(url, True, duration_ms, f"form_{method}") return ScrapeResponse( success=True, url=url, method=f"form_{method}", data={ "form_results": result, "final_url": final_url, "final_title": final_title, "original_url": url, }, ) else: error_response = ErrorHandler.handle_scraping_error( Exception(result.get("error", "Form interaction failed")), url, f"form_{method}", ) metrics_collector.record_request( url, False, duration_ms, f"form_{method}", error_response["error"]["category"], ) return ScrapeResponse( success=False, url=url, method=f"form_{method}", error=error_response["error"]["message"], ) except Exception as e: duration_ms = ( int((time.time() - start_time) * 1000) if "start_time" in locals() else 0 ) error_response = ErrorHandler.handle_scraping_error(e, url, f"form_{method}") metrics_collector.record_request( url, False, duration_ms, f"form_{method}", error_response["error"]["category"], ) return ScrapeResponse( success=False, url=url, method=f"form_{method}", error=error_response["error"]["message"], )
• extractor/advanced_features.py:498-729 (helper)

  The FormHandler class provides the core utility methods for filling form fields (text, select, checkbox, file) and submitting forms, supporting both Selenium WebDriver and Playwright APIs.
  class FormHandler: """Handle form interactions and submissions.""" def __init__(self, driver_or_page: Any) -> None: self.driver_or_page = driver_or_page # Simple check for Playwright page self.is_playwright = hasattr(driver_or_page, "fill") async def fill_form( self, form_data: Dict[str, Any], submit: bool = False, submit_button_selector: Optional[str] = None, ) -> Dict[str, Any]: """ Fill and optionally submit a form. Args: form_data: Dict mapping field selectors to values submit: Whether to submit the form submit_button_selector: Selector for submit button (if not using default) """ try: results = {} for field_selector, value in form_data.items(): field_result = await self._fill_field(field_selector, value) results[field_selector] = field_result if submit: submit_result = await self._submit_form(submit_button_selector) results["_submit"] = submit_result return {"success": True, "results": results} except Exception as e: logger.error(f"Error filling form: {str(e)}") return {"success": False, "error": str(e)} async def _fill_field(self, selector: str, value: Any) -> Dict[str, Any]: """Fill a single form field.""" try: if self.is_playwright: return await self._fill_field_playwright(selector, value) else: return await self._fill_field_selenium(selector, value) except Exception as e: return {"success": False, "error": str(e)} async def _fill_field_selenium(self, selector: str, value: Any) -> Dict[str, Any]: """Fill field using Selenium.""" try: element = self.driver_or_page.find_element(By.CSS_SELECTOR, selector) tag_name = element.tag_name.lower() input_type = element.get_attribute("type") if tag_name == "select": # Handle select dropdown select = Select(element) if isinstance(value, int): select.select_by_index(value) elif str(value).isdigit(): select.select_by_value(str(value)) else: select.select_by_visible_text(str(value)) elif input_type in ["checkbox", "radio"]: # Handle checkbox/radio if value and not element.is_selected(): element.click() elif not value and element.is_selected(): element.click() elif input_type == "file": # Handle file upload element.send_keys(str(value)) else: # Handle text inputs element.clear() element.send_keys(str(value)) return {"success": True, "value": value} except Exception as e: return {"success": False, "error": str(e)} async def _fill_field_playwright(self, selector: str, value: Any) -> Dict[str, Any]: """Fill field using Playwright.""" try: element = await self.driver_or_page.query_selector(selector) if not element: return {"success": False, "error": "Element not found"} tag_name = await element.evaluate("el => el.tagName.toLowerCase()") input_type = await element.get_attribute("type") if tag_name == "select": # Handle select dropdown if isinstance(value, int): await element.select_option(index=value) else: await element.select_option(label=str(value)) elif input_type in ["checkbox", "radio"]: # Handle checkbox/radio is_checked = await element.is_checked() if value and not is_checked: await element.check() elif not value and is_checked: await element.uncheck() elif input_type == "file": # Handle file upload await element.set_input_files(str(value)) else: # Handle text inputs await element.fill(str(value)) return {"success": True, "value": value} except Exception as e: return {"success": False, "error": str(e)} async def _submit_form( self, submit_button_selector: Optional[str] = None ) -> Dict[str, Any]: """Submit the form.""" try: if self.is_playwright: return await self._submit_form_playwright(submit_button_selector) else: return await self._submit_form_selenium(submit_button_selector) except Exception as e: return {"success": False, "error": str(e)} async def _submit_form_selenium( self, submit_button_selector: Optional[str] = None ) -> Dict[str, Any]: """Submit form using Selenium.""" try: if submit_button_selector: # Use specific submit button submit_button = self.driver_or_page.find_element( By.CSS_SELECTOR, submit_button_selector ) submit_button.click() else: # Try to find submit button automatically submit_selectors = [ "input[type='submit']", "button[type='submit']", "button:contains('Submit')", "input[value*='Submit']", "button:contains('Send')", ] for selector in submit_selectors: try: if "contains" in selector: # Use XPath for text content xpath = "//button[contains(text(), 'Submit')] | //button[contains(text(), 'Send')]" submit_button = self.driver_or_page.find_element( By.XPATH, xpath ) else: submit_button = self.driver_or_page.find_element( By.CSS_SELECTOR, selector ) submit_button.click() break except NoSuchElementException: continue else: # If no submit button found, try submitting the form directly form = self.driver_or_page.find_element(By.TAG_NAME, "form") form.submit() # Wait for page to load after submission await asyncio.sleep(2) return {"success": True, "new_url": self.driver_or_page.current_url} except Exception as e: return {"success": False, "error": str(e)} async def _submit_form_playwright( self, submit_button_selector: Optional[str] = None ) -> Dict[str, Any]: """Submit form using Playwright.""" try: if submit_button_selector: # Use specific submit button await self.driver_or_page.click(submit_button_selector) else: # Try to find submit button automatically submit_selectors = [ "input[type='submit']", "button[type='submit']", "text=Submit", "text=Send", ] for selector in submit_selectors: try: await self.driver_or_page.click(selector) break except Exception: # nosec B112 # Continue trying next submit button selector continue else: # If no submit button found, press Enter on the form await self.driver_or_page.keyboard.press("Enter") # Wait for navigation or response try: await self.driver_or_page.wait_for_load_state( "networkidle", timeout=10000 ) except Exception: # nosec B110 # Ignore timeout or navigation errors - form might submit without page load pass return {"success": True, "new_url": self.driver_or_page.url} except Exception as e: return {"success": False, "error": str(e)}
• extractor/server.py:62-72 (schema)

  Output schema (Pydantic model) used by the fill_and_submit_form tool and other scraping tools to structure responses with success status, data, errors, etc.
  class ScrapeResponse(BaseModel): """Response model for scraping operations.""" success: bool = Field(..., description="操作是否成功") url: str = Field(..., description="被抓取的URL") method: str = Field(..., description="使用的抓取方法") data: Optional[Dict[str, Any]] = Field(default=None, description="抓取到的数据") metadata: Optional[Dict[str, Any]] = Field(default=None, description="页面元数据") error: Optional[str] = Field(default=None, description="错误信息（如果有）") timestamp: datetime = Field(default_factory=datetime.now, description="抓取时间戳")
• extractor/server.py:870-870 (registration)

  The @app.tool() decorator registers the fill_and_submit_form function as an MCP tool in the FastMCP app instance.
  @app.tool()