Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Update error handling to use ErrorData consistently #507

Merged
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion src/fetch/pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ classifiers = [
]
dependencies = [
"markdownify>=0.13.1",
"mcp>=1.0.0",
"mcp>=1.1.3",
"protego>=0.3.1",
"pydantic>=2.0.0",
"readabilipy>=0.2.0",
Expand Down
25 changes: 13 additions & 12 deletions src/fetch/src/mcp_server_fetch/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import (
ErrorData,
GetPromptResult,
Prompt,
PromptArgument,
Expand Down Expand Up @@ -79,15 +80,15 @@ async def check_may_autonomously_fetch_url(url: str, user_agent: str) -> None:
headers={"User-Agent": user_agent},
)
except HTTPError:
raise McpError(
raise McpError(ErrorData(
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(
raise McpError(ErrorData(
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
Expand All @@ -96,15 +97,15 @@ async def check_may_autonomously_fetch_url(url: str, user_agent: str) -> None:
)
robot_parser = Protego.parse(processed_robot_txt)
if not robot_parser.can_fetch(str(url), user_agent):
raise McpError(
raise McpError(ErrorData(
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(
Expand All @@ -124,12 +125,12 @@ async def fetch_url(
timeout=30,
)
except HTTPError as e:
raise McpError(INTERNAL_ERROR, f"Failed to fetch {url}: {e!r}")
raise McpError(ErrorData(INTERNAL_ERROR, f"Failed to fetch {url}: {e!r}"))
if response.status_code >= 400:
raise McpError(
raise McpError(ErrorData(
INTERNAL_ERROR,
f"Failed to fetch {url} - status code {response.status_code}",
)
))

page_raw = response.text

Expand Down Expand Up @@ -221,11 +222,11 @@ async def call_tool(name, arguments: dict) -> list[TextContent]:
try:
args = Fetch(**arguments)
except ValueError as e:
raise McpError(INVALID_PARAMS, str(e))
raise McpError(ErrorData(INVALID_PARAMS, str(e)))

url = str(args.url)
if not url:
raise McpError(INVALID_PARAMS, "URL is required")
raise McpError(ErrorData(INVALID_PARAMS, "URL is required"))

if not ignore_robots_txt:
await check_may_autonomously_fetch_url(url, user_agent_autonomous)
Expand Down Expand Up @@ -253,7 +254,7 @@ async def call_tool(name, arguments: dict) -> list[TextContent]:
@server.get_prompt()
async def get_prompt(name: str, arguments: dict | None) -> GetPromptResult:
if not arguments or "url" not in arguments:
raise McpError(INVALID_PARAMS, "URL is required")
raise McpError(ErrorData(INVALID_PARAMS, "URL is required"))

url = arguments["url"]

Expand Down