test_url_elicitation_error_throw.py

"""Test that UrlElicitationRequiredError is properly propagated as MCP error."""

import pytest
from inline_snapshot import snapshot

from mcp import Client, ErrorData, types
from mcp.server.mcpserver import Context, MCPServer
from mcp.server.session import ServerSession
from mcp.shared.exceptions import MCPError, UrlElicitationRequiredError


@pytest.mark.anyio
async def test_url_elicitation_error_thrown_from_tool():
    """Test that UrlElicitationRequiredError raised from a tool is received as MCPError by client."""
    mcp = MCPServer(name="UrlElicitationErrorServer")

    @mcp.tool(description="A tool that raises UrlElicitationRequiredError")
    async def connect_service(service_name: str, ctx: Context[ServerSession, None]) -> str:
        # This tool cannot proceed without authorization
        raise UrlElicitationRequiredError(
            [
                types.ElicitRequestURLParams(
                    mode="url",
                    message=f"Authorization required to connect to {service_name}",
                    url=f"https://{service_name}.example.com/oauth/authorize",
                    elicitation_id=f"{service_name}-auth-001",
                )
            ]
        )

    async with Client(mcp) as client:
        with pytest.raises(MCPError) as exc_info:
            await client.call_tool("connect_service", {"service_name": "github"})

        assert exc_info.value.error == snapshot(
            ErrorData(
                code=types.URL_ELICITATION_REQUIRED,
                message="URL elicitation required",
                data={
                    "elicitations": [
                        {
                            "mode": "url",
                            "message": "Authorization required to connect to github",
                            "url": "https://github.example.com/oauth/authorize",
                            "elicitationId": "github-auth-001",
                        }
                    ]
                },
            )
        )


@pytest.mark.anyio
async def test_url_elicitation_error_from_error():
    """Test that client can reconstruct UrlElicitationRequiredError from MCPError."""
    mcp = MCPServer(name="UrlElicitationErrorServer")

    @mcp.tool(description="A tool that raises UrlElicitationRequiredError with multiple elicitations")
    async def multi_auth(ctx: Context[ServerSession, None]) -> str:
        raise UrlElicitationRequiredError(
            [
                types.ElicitRequestURLParams(
                    mode="url",
                    message="GitHub authorization required",
                    url="https://github.example.com/oauth",
                    elicitation_id="github-auth",
                ),
                types.ElicitRequestURLParams(
                    mode="url",
                    message="Google Drive authorization required",
                    url="https://drive.google.com/oauth",
                    elicitation_id="gdrive-auth",
                ),
            ]
        )

    async with Client(mcp) as client:
        # Call the tool and catch the error
        with pytest.raises(MCPError) as exc_info:
            await client.call_tool("multi_auth", {})

        # Reconstruct the typed error
        mcp_error = exc_info.value
        assert mcp_error.code == types.URL_ELICITATION_REQUIRED

        url_error = UrlElicitationRequiredError.from_error(mcp_error.error)

        # Verify the reconstructed error has both elicitations
        assert len(url_error.elicitations) == 2
        assert url_error.elicitations[0].elicitation_id == "github-auth"
        assert url_error.elicitations[1].elicitation_id == "gdrive-auth"


@pytest.mark.anyio
async def test_normal_exceptions_still_return_error_result():
    """Test that normal exceptions still return CallToolResult with is_error=True."""
    mcp = MCPServer(name="NormalErrorServer")

    @mcp.tool(description="A tool that raises a normal exception")
    async def failing_tool(ctx: Context[ServerSession, None]) -> str:
        raise ValueError("Something went wrong")

    async with Client(mcp) as client:
        # Normal exceptions should be returned as error results, not MCPError
        result = await client.call_tool("failing_tool", {})
        assert result.is_error is True
        assert len(result.content) == 1
        assert isinstance(result.content[0], types.TextContent)
        assert "An unexpected error occurred executing tool failing_tool" in result.content[0].text
        # Verify the original exception details are NOT leaked to the client
        assert "Something went wrong" not in result.content[0].text