Hardening an MCP Fetch Server: Security and Stability Improvements

When building a Model Context Protocol (MCP) server that fetches web content, security and stability aren’t optional—they’re essential. In this post, I’ll walk through the process of analyzing and improving an MCP fetch server, addressing critical vulnerabilities and stability issues that could impact production deployments.

The Challenge

We started with a functional MCP fetch server that could retrieve web pages and convert them to markdown. While it worked, a thorough security audit revealed several concerning issues that needed immediate attention.

Critical Security Vulnerabilities

1. Server-Side Request Forgery (SSRF)

The original code accepted any URL without validation, creating a serious SSRF vulnerability. An attacker could potentially:

• Access internal network resources (127.0.0.1, 192.168.x.x)
• Scan internal ports and services
• Bypass firewall restrictions

Solution: We implemented URL sanitization and IP address validation:

def is_private_ip(hostname: str) -> bool:
    """Check if a hostname resolves to a private IP address."""
    try:
        ip = ipaddress.ip_address(hostname)
        return ip.is_private or ip.is_loopback or ip.is_multicast or ip.is_reserved
    except ValueError:
        return False

def sanitize_url(url: str) -> str:
    """Sanitize URL to prevent potential security issues."""
    parsed = urlparse(url)
    
    if parsed.scheme not in ("http", "https"):
        raise McpError(ErrorData(
            code=INVALID_PARAMS,
            message=f"Unsupported URL scheme: {parsed.scheme}"
        ))
    
    if is_private_ip(parsed.netloc.split(':')[0]):
        raise McpError(ErrorData(
            code=INVALID_PARAMS,
            message="Access to private network addresses is not allowed."
        ))
    
    return url

2. Unlimited Resource Consumption

The original implementation had no safeguards against:

• Downloading extremely large files
• Making unlimited requests to the same domain
• Exhausting connection pools

Solution: We added multiple layers of protection:

# Connection limits
limits = Limits(
    max_connections=100, 
    max_keepalive_connections=20, 
    max_connections_per_host=10
)

# Rate limiting per domain
class RateLimiter:
    def __init__(self, limit_per_minute: int = 10):
        self.limit = limit_per_minute
        self.window_size = 60.0
        self.requests = {}
        
    def can_make_request(self, url: str) -> bool:
        domain = urlparse(url).netloc
        now = time.time()
        
        if domain not in self.requests:
            return True
        
        valid_requests = [t for t in self.requests[domain] 
                         if now - t < self.window_size]
        
        return len(valid_requests) < self.limit

Stability Improvements

1. Robust Error Handling

The original code had minimal error handling, leading to crashes on unexpected inputs. We implemented comprehensive exception handling with fallback mechanisms:

def extract_content_from_html(html: str) -> str:
    try:
        ret = readabilipy.simple_json.simple_json_from_html_string(
            html, use_readability=True
        )
        
        if not ret["content"]:
            # Fallback to body extraction
            body_match = re.search(
                r"<body[^>]*>(.*?)</body>", 
                html, 
                re.DOTALL | re.IGNORECASE
            )
            if body_match:
                content = markdownify.markdownify(
                    body_match.group(1), 
                    heading_style=markdownify.ATX
                )
                return content
            
            # Last resort: convert entire HTML
            content = markdownify.markdownify(
                html, 
                heading_style=markdownify.ATX
            )
            return content
        
        content = markdownify.markdownify(
            ret["content"], 
            heading_style=markdownify.ATX
        )
        return content
    except Exception as e:
        return f"<error>HTML processing failed: {str(e)}</error>\n\nRaw content (first 1000 chars):\n{html[:1000]}"

2. Retry Logic with Exponential Backoff

Network requests can fail for transient reasons. We added intelligent retry logic:

async def fetch_url(url: str, user_agent: str, force_raw: bool, client: AsyncClient):
    retry_count = 0
    last_error = None
    
    while retry_count < MAX_RETRIES:
        try:
            response = await client.get(url, ...)
            # Process successful response
            return content, prefix
            
        except (ReadTimeout, ConnectTimeout) as e:
            retry_count += 1
            last_error = e
            if retry_count < MAX_RETRIES:
                # Exponential backoff: 2, 4, 8 seconds
                await asyncio.sleep(2 ** retry_count)
            continue
        except HTTPError as e:
            # Fatal error, don't retry
            raise McpError(...)
    
    raise McpError(...)

3. Improved Content Type Detection

The original HTML detection was fragile. We created a more robust approach:

def is_likely_html(response: Response) -> bool:
    content_type = response.headers.get("content-type", "").lower()
    
    # Check content type header first
    if "text/html" in content_type or "application/xhtml" in content_type:
        return True
        
    # Fallback to content inspection
    if not content_type or "text/plain" in content_type:
        sample = response.text[:1000].lower()
        return "<html" in sample or "<!doctype html" in sample
        
    return False

Architectural Improvements

Better Resource Management

We refactored the lifespan management to properly handle shared resources:

@dataclass
class AppContext:
    http_client: AsyncClient
    rate_limiter: RateLimiter

@asynccontextmanager
async def app_lifespan(server: FastMCP) -> AsyncIterator[AppContext]:
    limits = Limits(
        max_connections=100,
        max_keepalive_connections=20,
        max_connections_per_host=10
    )
    
    client = AsyncClient(
        timeout=30,
        limits=limits,
        follow_redirects=True,
        verify=True
    )
    
    rate_limiter = RateLimiter()
    
    try:
        yield AppContext(http_client=client, rate_limiter=rate_limiter)
    finally:
        await client.aclose()

Modular Server Creation

We restructured the code to support both CLI and programmatic usage:

def create_server(custom_user_agent=None, ignore_robots_txt=False):
    """Create and configure the MCP server."""
    server = FastMCP("mcp-fetch", lifespan=app_lifespan)
    
    if custom_user_agent:
        server.custom_user_agent = custom_user_agent
    
    server.extra["ignore_robots_txt"] = ignore_robots_txt
    
    # Register routes...
    return server

async def serve(custom_user_agent=None, ignore_robots_txt=False):
    """Start the MCP-fetch server."""
    server = create_server(custom_user_agent, ignore_robots_txt)
    await server.start()

def main():
    """CLI entry point"""
    parser = argparse.ArgumentParser(
        description="give a model the ability to make web requests"
    )
    parser.add_argument("--user-agent", type=str)
    parser.add_argument("--ignore-robots-txt", action="store_true")
    
    args = parser.parse_args()
    asyncio.run(serve(args.user_agent, args.ignore_robots_txt))

Additional Enhancements

• SSL Certificate Verification: Enabled by default to prevent man-in-the-middle attacks
• Reduced Maximum Content Length: From 1M to 100K characters to prevent abuse
• Better Truncation Messaging: Users now see exactly how many characters remain
• Improved robots.txt Handling: More graceful handling of missing or malformed robots.txt files

Testing Recommendations

When deploying these improvements, test for:

1. SSRF Prevention: Try fetching http://127.0.0.1, http://192.168.1.1, http://localhost
2. Rate Limiting: Make rapid requests to the same domain
3. Error Handling: Test with malformed URLs, non-existent domains, timeout scenarios
4. Content Processing: Test with various HTML structures, JavaScript-heavy sites, binary content
5. Resource Limits: Verify connection pooling and memory usage under load

Conclusion

Security and stability in web-facing services require multiple layers of defense. By implementing URL sanitization, rate limiting, robust error handling, and resource management, we transformed a functional but vulnerable MCP server into a production-ready service.

The key takeaways:

• Always validate and sanitize user inputs, especially URLs
• Implement rate limiting to prevent abuse
• Add comprehensive error handling with fallback mechanisms
• Manage resources carefully with proper limits and timeouts
• Design for both direct execution and programmatic usage

These improvements make the MCP fetch server suitable for production use while maintaining its core functionality and ease of use.