Skip to content
Closed
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
5 changes: 5 additions & 0 deletions src/mcp/shared/exceptions.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
from __future__ import annotations

from collections.abc import Callable
from typing import Any, cast

from mcp_types import INVALID_REQUEST, URL_ELICITATION_REQUIRED, ElicitRequestURLParams, ErrorData, JSONRPCError
Expand Down Expand Up @@ -117,3 +118,7 @@ def from_error(cls, error: ErrorData) -> UrlElicitationRequiredError:
raw_elicitations = cast(list[dict[str, Any]], data.get("elicitations", []))
elicitations = [ElicitRequestURLParams.model_validate(e) for e in raw_elicitations]
return cls(elicitations, error.message)

def __reduce__(self) -> tuple[Callable[[ErrorData], UrlElicitationRequiredError], tuple[ErrorData]]:
"""Reconstruct the exception from its wire-compatible error payload."""
return (self.from_error, (self.error,))
23 changes: 23 additions & 0 deletions tests/shared/test_exceptions.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,7 @@
"""Tests for MCP exception classes."""

import pickle

import pytest
from mcp_types import URL_ELICITATION_REQUIRED, ElicitRequestURLParams, ErrorData, JSONRPCError

Expand Down Expand Up @@ -117,6 +119,27 @@ def test_url_elicitation_required_error_serialization_roundtrip() -> None:
assert reconstructed.elicitations[0].message == original.elicitations[0].message


def test_url_elicitation_required_error_pickle_roundtrip() -> None:
"""Pickle reconstruction preserves the specialized exception payload."""
original = UrlElicitationRequiredError(
[
ElicitRequestURLParams(
mode="url",
message="Auth required",
url="https://example.com/auth",
elicitation_id="test-123",
)
],
message="Please authenticate",
)

restored = pickle.loads(pickle.dumps(original))

assert isinstance(restored, UrlElicitationRequiredError)
assert restored.error == original.error
assert restored.elicitations == original.elicitations


def test_url_elicitation_required_error_data_contains_elicitations() -> None:
"""Test that error data contains properly serialized elicitations."""
elicitation = ElicitRequestURLParams(
Expand Down
Loading