modelcontextprotocol/python-sdk
Watch
1
0
Fork
You've already forked python-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor: McpError renamed to MCPError and flatten parameters (#1956)

Browse source 
Co-authored-by: Max Isbey <224885523+maxisbey@users.noreply.github.com>
This commit is contained in:
Marcelo Trylesinski 2026-01-26 14:37:44 +01:00 • committed by GitHub
commit acba5478a9
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
4 changed files with 253 additions and 326 deletions
Download patch file Download diff file Expand all files Collapse all files

46
src/mcp/shared/exceptions.py
View file

@ -2,18 +2,40 @@ from __future__ import annotations
from typing import Any, cast
from mcp.types import URL_ELICITATION_REQUIRED, ElicitRequestURLParams, ErrorData
from mcp.types import URL_ELICITATION_REQUIRED, ElicitRequestURLParams, ErrorData, JSONRPCError
class McpError(Exception):
class MCPError(Exception):
"""Exception type raised when an error arrives over an MCP connection."""
error: ErrorData
def __init__(self, error: ErrorData):
"""Initialize McpError."""
super().__init__(error.message)
self.error = error
def __init__(self, code: int, message: str, data: Any = None):
super().__init__(code, message, data)
self.error = ErrorData(code=code, message=message, data=data)
@property
def code(self) -> int:
return self.error.code
@property
def message(self) -> str:
return self.error.message
@property
def data(self) -> Any:
return self.error.data # pragma: no cover
@classmethod
def from_jsonrpc_error(cls, error: JSONRPCError) -> MCPError:
return cls.from_error_data(error.error)
@classmethod
def from_error_data(cls, error: ErrorData) -> MCPError:
return cls(code=error.code, message=error.message, data=error.data)
def __str__(self) -> str:
return self.message
class StatelessModeNotSupported(RuntimeError):
@ -33,7 +55,7 @@ class StatelessModeNotSupported(RuntimeError):
self.method = method
class UrlElicitationRequiredError(McpError):
class UrlElicitationRequiredError(MCPError):
"""Specialized error for when a tool requires URL mode elicitation(s) before proceeding.
Servers can raise this error from tool handlers to indicate that the client
@ -42,7 +64,6 @@ class UrlElicitationRequiredError(McpError):
Example:
raise UrlElicitationRequiredError([
ElicitRequestURLParams(
mode="url",
message="Authorization required for your files",
url="https://example.com/oauth/authorize",
elicitation_id="auth-001"
@ -50,23 +71,18 @@ class UrlElicitationRequiredError(McpError):
])
"""
def __init__(
self,
elicitations: list[ElicitRequestURLParams],
message: str | None = None,
):
def __init__(self, elicitations: list[ElicitRequestURLParams], message: str | None = None):
"""Initialize UrlElicitationRequiredError."""
if message is None:
message = f"URL elicitation{'s' if len(elicitations) > 1 else ''} required"
self._elicitations = elicitations
error = ErrorData(
super().__init__(
code=URL_ELICITATION_REQUIRED,
message=message,
data={"elicitations": [e.model_dump(by_alias=True, exclude_none=True) for e in elicitations]},
)
super().__init__(error)
@property
def elicitations(self) -> list[ElicitRequestURLParams]:

31
src/mcp/shared/experimental/tasks/capabilities.py
View file

@ -7,13 +7,8 @@ the main session code clean.
WARNING: These APIs are experimental and may change without notice.
"""
from mcp.shared.exceptions import McpError
from mcp.types import (
INVALID_REQUEST,
ClientCapabilities,
ClientTasksCapability,
ErrorData,
)
from mcp.shared.exceptions import MCPError
from mcp.types import INVALID_REQUEST, ClientCapabilities, ClientTasksCapability
def check_tasks_capability(
@ -76,36 +71,26 @@ def has_task_augmented_sampling(caps: ClientCapabilities) -> bool:
def require_task_augmented_elicitation(client_caps: ClientCapabilities | None) -> None:
"""Raise McpError if client doesn't support task-augmented elicitation.
"""Raise MCPError if client doesn't support task-augmented elicitation.
Args:
client_caps: The client's declared capabilities, or None if not initialized
Raises:
McpError: If client doesn't support task-augmented elicitation
MCPError: If client doesn't support task-augmented elicitation
"""
if client_caps is None or not has_task_augmented_elicitation(client_caps):
raise McpError(
ErrorData(
code=INVALID_REQUEST,
message="Client does not support task-augmented elicitation",
)
)
raise MCPError(code=INVALID_REQUEST, message="Client does not support task-augmented elicitation")
def require_task_augmented_sampling(client_caps: ClientCapabilities | None) -> None:
"""Raise McpError if client doesn't support task-augmented sampling.
"""Raise MCPError if client doesn't support task-augmented sampling.
Args:
client_caps: The client's declared capabilities, or None if not initialized
Raises:
McpError: If client doesn't support task-augmented sampling
MCPError: If client doesn't support task-augmented sampling
"""
if client_caps is None or not has_task_augmented_sampling(client_caps):
raise McpError(
ErrorData(
code=INVALID_REQUEST,
message="Client does not support task-augmented sampling",
)
)
raise MCPError(code=INVALID_REQUEST, message="Client does not support task-augmented sampling")

19
src/mcp/shared/experimental/tasks/helpers.py
View file

@ -9,7 +9,7 @@ from contextlib import asynccontextmanager
from datetime import datetime, timezone
from uuid import uuid4
from mcp.shared.exceptions import McpError
from mcp.shared.exceptions import MCPError
from mcp.shared.experimental.tasks.context import TaskContext
from mcp.shared.experimental.tasks.store import TaskStore
from mcp.types import (
@ -19,7 +19,6 @@ from mcp.types import (
TASK_STATUS_FAILED,
TASK_STATUS_WORKING,
CancelTaskResult,
ErrorData,
Task,
TaskMetadata,
TaskStatus,
@ -68,7 +67,7 @@ async def cancel_task(
CancelTaskResult with the cancelled task state
Raises:
McpError: With INVALID_PARAMS (-32602) if:
MCPError: With INVALID_PARAMS (-32602) if:
- Task does not exist
- Task is already in a terminal state (completed, failed, cancelled)
@ -79,20 +78,10 @@ async def cancel_task(
"""
task = await store.get_task(task_id)
if task is None:
raise McpError(
ErrorData(
code=INVALID_PARAMS,
message=f"Task not found: {task_id}",
)
)
raise MCPError(code=INVALID_PARAMS, message=f"Task not found: {task_id}")
if is_terminal(task.status):
raise McpError(
ErrorData(
code=INVALID_PARAMS,
message=f"Cannot cancel task in terminal state '{task.status}'",
)
)
raise MCPError(code=INVALID_PARAMS, message=f"Cannot cancel task in terminal state '{task.status}'")
# Update task to cancelled status
cancelled_task = await store.update_task(task_id, status=TASK_STATUS_CANCELLED)

18
src/mcp/shared/session.py
View file

@ -11,7 +11,7 @@ from anyio.streams.memory import MemoryObjectReceiveStream, MemoryObjectSendStre
from pydantic import BaseModel, TypeAdapter
from typing_extensions import Self
from mcp.shared.exceptions import McpError
from mcp.shared.exceptions import MCPError
from mcp.shared.message import MessageMetadata, ServerMessageMetadata, SessionMessage
from mcp.shared.response_router import ResponseRouter
from mcp.types import (
@ -237,7 +237,7 @@ class BaseSession(
) -> ReceiveResultT:
"""Sends a request and wait for a response.
Raises an McpError if the response contains an error. If a request read timeout is provided, it will take
Raises an MCPError if the response contains an error. If a request read timeout is provided, it will take
precedence over the session read timeout.
Do not use this method to emit notifications! Use send_notification() instead.
@ -271,18 +271,12 @@ class BaseSession(
with anyio.fail_after(timeout):
response_or_error = await response_stream_reader.receive()
except TimeoutError:
raise McpError(
ErrorData(
code=REQUEST_TIMEOUT,
message=(
f"Timed out while waiting for response to {request.__class__.__name__}. "
f"Waited {timeout} seconds."
),
)
)
class_name = request.__class__.__name__
message = f"Timed out while waiting for response to {class_name}. Waited {timeout} seconds."
raise MCPError(code=REQUEST_TIMEOUT, message=message)
if isinstance(response_or_error, JSONRPCError):
raise McpError(response_or_error.error)
raise MCPError.from_jsonrpc_error(response_or_error)
else:
return result_type.model_validate(response_or_error.result, by_name=False)