Nexus MCP Server

Instructions

Implementation Reference

• nexus_server.py:97-232 (handler)

  The complete implementation of the 'nexus_read' tool handler, including registration via @mcp.tool() decorator. It fetches the given URL, parses HTML with BeautifulSoup, and extracts content based on the 'focus' parameter: 'general' for full text, 'code' for headers/code/tables, 'auto' detects type.
  @mcp.tool() async def nexus_read(url: str, focus: str = "auto") -> str: """ Reads a URL with intelligent parsing logic. Args: url: The URL to visit. focus: 'general' = Returns clean article text (Exa style). 'code' = Returns only headers, code blocks, and tables (Ref style). 'auto' = Detects if it's a doc site and switches to 'code' mode. Returns: Parsed and cleaned content from the URL. """ logger.info(f"Read requested - URL: '{url}', Focus: {focus}") # Validate inputs if not url or not url.strip(): error_msg = "URL cannot be empty" logger.error(error_msg) return f"Error: {error_msg}" if focus not in ["auto", "general", "code"]: error_msg = f"Invalid focus '{focus}'. Must be 'auto', 'general', or 'code'" logger.error(error_msg) return f"Error: {error_msg}" url = url.strip() # Validate URL format if not url.startswith(("http://", "https://")): error_msg = "URL must start with http:// or https://" logger.error(error_msg) return f"Error: {error_msg}" # Auto-detection logic original_focus = focus if focus == "auto": technical_indicators = ["docs", "api", "reference", "github", "guide", "documentation"] if any(ind in url.lower() for ind in technical_indicators): focus = "code" logger.debug(f"Auto-detected technical site, switching to code focus") else: focus = "general" logger.debug(f"Auto-detected general site, using general focus") async with httpx.AsyncClient( follow_redirects=True, headers={"User-Agent": "NexusMCP/1.0"}, timeout=REQUEST_TIMEOUT ) as client: try: response = await client.get(url) response.raise_for_status() logger.debug(f"Successfully fetched URL: {url} (Status: {response.status_code})") soup = bs4.BeautifulSoup(response.text, 'html.parser') # Pre-cleaning (Remove junk common to all modes) for trash in soup(["script", "style", "nav", "footer", "iframe", "svg", "noscript"]): trash.decompose() output = [] output.append(f"=== SOURCE: {url} ===") output.append(f"=== MODE: {focus.upper()} ===\n") if focus == "code": # REF MODE (High Signal / Low Noise) relevant_tags = soup.find_all(['h1', 'h2', 'h3', 'h4', 'pre', 'code', 'table']) for tag in relevant_tags: if tag.name in ['h1', 'h2', 'h3', 'h4']: header_text = tag.get_text(strip=True) if header_text: output.append(f"\n## {header_text}") elif tag.name == 'pre': code_text = tag.get_text() if code_text.strip(): output.append(f"```\n{code_text}\n```") elif tag.name == 'code' and tag.parent.name != 'pre': code_text = tag.get_text(strip=True) if code_text: output.append(f"`{code_text}`") elif tag.name == 'table': # Enhanced table extraction try: rows = tag.find_all('tr') if rows: output.append("\n[Table]") for row in rows[:10]: # Limit to first 10 rows cells = row.find_all(['td', 'th']) cell_texts = [cell.get_text(strip=True) for cell in cells] if cell_texts: output.append(" | ".join(cell_texts)) except Exception as table_error: logger.warning(f"Table parsing failed: {table_error}") output.append("\n[Table - parsing failed]") if len(output) < MIN_CODE_ELEMENTS_THRESHOLD: fallback_msg = f"Code-focused extraction found minimal content ({len(output)} elements). The page may not contain structured documentation. Try focus='general' for better results." logger.warning(f"Insufficient code elements found at {url}") return fallback_msg else: # GENERAL MODE (Full Context) text = soup.get_text(separator='\n') lines = [line.strip() for line in text.split('\n') if line.strip()] output.append("\n".join(lines)) result = "\n".join(output)[:MAX_CONTENT_LENGTH] logger.info(f"Read successful - Extracted {len(result)} characters from {url}") if len("\n".join(output)) > MAX_CONTENT_LENGTH: result += f"\n\n[Content truncated at {MAX_CONTENT_LENGTH} characters]" logger.debug(f"Content truncated for {url}") return result except httpx.TimeoutException: error_msg = f"Request timed out after {REQUEST_TIMEOUT}s" logger.error(f"Timeout reading {url}") return f"Error: {error_msg}" except httpx.HTTPStatusError as e: error_msg = f"HTTP error {e.response.status_code}: {e.response.reason_phrase}" logger.error(f"HTTP error reading {url}: {error_msg}") return f"Error: {error_msg}" except httpx.RequestError as e: error_msg = f"Network error: {str(e)}" logger.error(f"Network error reading {url}: {error_msg}") return f"Error: {error_msg}" except Exception as e: error_msg = f"Unexpected error reading URL: {str(e)}" logger.exception(f"Unexpected error reading {url}") return f"Error: {error_msg}"
• nexus_server.py:99-111 (schema)

  The docstring providing input/output schema description for the nexus_read tool, used by FastMCP for tool schema generation.
  """ Reads a URL with intelligent parsing logic. Args: url: The URL to visit. focus: 'general' = Returns clean article text (Exa style). 'code' = Returns only headers, code blocks, and tables (Ref style). 'auto' = Detects if it's a doc site and switches to 'code' mode. Returns: Parsed and cleaned content from the URL. """
• nexus_server.py:97-97 (registration)

  The @mcp.tool() decorator that registers the nexus_read function as an MCP tool in the FastMCP server.
  @mcp.tool()