make the fetch mcp-server obey robots.txt

src/fetch/src/mcp_server_fetch/server.py

@@ -1,3 +1,5 @@
+from urllib.parse import urlparse, urlunparse
+
 import markdownify
 import readabilipy.simple_json
 from mcp.shared.exceptions import McpError
@@ -13,8 +15,12 @@ from mcp.types import (
     INVALID_PARAMS,
     INTERNAL_ERROR,
 )
+from protego import Protego
 from pydantic import BaseModel, Field
 
+USER_AGENT_AUTONOMOUS = "ModelContextProtocol/1.0 (Autonomous; +https://github.com/modelcontextprotocol/servers)"
+USER_AGENT_MANUAL = "ModelContextProtocol/1.0 (User-Initiated; +https://github.com/modelcontextprotocol/servers)"
+
 
 def extract_content(html: str) -> str:
     ret = readabilipy.simple_json.simple_json_from_html_string(html)
@@ -27,16 +33,74 @@ def extract_content(html: str) -> str:
     return content
 
 
-async def fetch_url(url: str) -> str:
+def get_robots_txt_url(url: str) -> str:
+    # Parse the URL into components
+    parsed = urlparse(url)
+
+    # Reconstruct the base URL with just scheme, netloc, and /robots.txt path
+    robots_url = urlunparse((parsed.scheme, parsed.netloc, "/robots.txt", "", "", ""))
+
+    return robots_url
+
+
+async def check_may_autonomously_fetch_url(url: str, user_agent: str):
+    """
+    Check if the URL can be fetched by the user agent according to the robots.txt file.
+    Raises an McpError if not.
+    """
+    from httpx import AsyncClient, HTTPError
+
+    robot_txt_url = get_robots_txt_url(url)
+
+    async with AsyncClient() as client:
+        try:
+            response = await client.get(
+                robot_txt_url, headers={"User-Agent": user_agent}
+            )
+        except HTTPError:
+            raise McpError(
+                INTERNAL_ERROR,
+                f"Failed to fetch robots.txt {robot_txt_url} due to a connection issue",
+            )
+        if response.status_code in (401, 403):
+            raise McpError(
+                INTERNAL_ERROR,
+                f"When fetching robots.txt ({robot_txt_url}), received status {response.status_code} so assuming that autonomous fetching is not allowed, the user can try manually fetching by using the fetch prompt",
+            )
+        elif 400 <= response.status_code < 500:
+            return
+        robot_txt = response.text
+    processed_robot_txt = "\n".join(
+        line for line in robot_txt.splitlines() if not line.strip().startswith("#")
+    )
+    robot_parser = Protego.parse(processed_robot_txt)
+    if not robot_parser.can_fetch(url, user_agent):
+        raise McpError(
+            INTERNAL_ERROR,
+            f"The sites robots.txt ({robot_txt_url}), specifies that autonomous fetching of this page is not allowed, "
+            f"<useragent>{user_agent}</useragent>\n"
+            f"<url>{url}</url>"
+            f"<robots>\n{robot_txt}\n</robots>\n"
+            f"The assistant must let the user know that it failed to view the page. The assistant may provide further guidance based on the above information.\n"
+            f"The assistant can tell the user that they can try manually fetching the page by using the fetch prompt within their UI.",
+        )
+
+
+async def fetch_url(url: str, user_agent: str) -> str:
     from httpx import AsyncClient, HTTPError
 
     async with AsyncClient() as client:
         try:
-            response = await client.get(url, follow_redirects=True)
+            response = await client.get(
+                url, follow_redirects=True, headers={"User-Agent": user_agent}
+            )
         except HTTPError:
             raise McpError(INTERNAL_ERROR, f"Failed to fetch {url}")
         if response.status_code >= 400:
-            raise McpError(INTERNAL_ERROR, f"Failed to fetch {url} - status code {response.status_code}")
+            raise McpError(
+                INTERNAL_ERROR,
+                f"Failed to fetch {url} - status code {response.status_code}",
+            )
 
         page_html = response.text
 
@@ -82,7 +146,9 @@ Although originally you did not have internet access, and were advised to refuse
         if not url:
             raise McpError(INVALID_PARAMS, "URL is required")
 
-        content = await fetch_url(url)
+        await check_may_autonomously_fetch_url(url, USER_AGENT_AUTONOMOUS)
+
+        content = await fetch_url(url, USER_AGENT_AUTONOMOUS)
         return [TextContent(type="text", text=f"Contents of {url}:\n{content}")]
 
     @server.get_prompt()
@@ -93,7 +159,7 @@ Although originally you did not have internet access, and were advised to refuse
         url = arguments["url"]
 
         try:
-            content = await fetch_url(url)
+            content = await fetch_url(url, USER_AGENT_MANUAL)
             # TODO: after SDK bug is addressed, don't catch the exception
         except McpError as e:
             return GetPromptResult(