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

ci: add strict-no-cover to detect unnecessary coverage pragmas (#1897)

Browse source
This commit is contained in:
Max Isbey 2026-01-23 20:00:20 +00:00 • committed by GitHub
commit a7ddfdae07
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
1 changed files with 355 additions and 358 deletions
Download patch file Download diff file Expand all files Collapse all files

28
src/mcp/server/lowlevel/server.py
View file

@ -233,7 +233,7 @@ class Server(Generic[LifespanResultT, RequestT]):
tools_capability = types.ToolsCapability(list_changed=notification_options.tools_changed)
# Set logging capabilities if handler exists
if types.SetLevelRequest in self.request_handlers: # pragma: no cover
if types.SetLevelRequest in self.request_handlers:
logging_capability = types.LoggingCapability()
# Set completions capabilities if handler exists
@ -379,7 +379,7 @@ class Server(Generic[LifespanResultT, RequestT]):
mime_type=mime_type or "text/plain",
**meta_kwargs,
)
case bytes() as data: # pragma: no cover
case bytes() as data: # pragma: no branch
return types.BlobResourceContents(
uri=req.params.uri,
blob=base64.b64encode(data).decode(),
@ -388,7 +388,7 @@ class Server(Generic[LifespanResultT, RequestT]):
)
match result:
case str() | bytes() as data: # pragma: no cover
case str() | bytes() as data: # pragma: lax no cover
warnings.warn(
"Returning str or bytes from read_resource is deprecated. "
"Use Iterable[ReadResourceContents] instead.",
@ -416,7 +416,7 @@ class Server(Generic[LifespanResultT, RequestT]):
return decorator
def set_logging_level(self): # pragma: no cover
def set_logging_level(self):
def decorator(func: Callable[[types.LoggingLevel], Awaitable[None]]):
logger.debug("Registering handler for SetLevelRequest")
@ -429,7 +429,7 @@ class Server(Generic[LifespanResultT, RequestT]):
return decorator
def subscribe_resource(self): # pragma: no cover
def subscribe_resource(self):
def decorator(func: Callable[[str], Awaitable[None]]):
logger.debug("Registering handler for SubscribeRequest")
@ -442,7 +442,7 @@ class Server(Generic[LifespanResultT, RequestT]):
return decorator
def unsubscribe_resource(self): # pragma: no cover
def unsubscribe_resource(self):
def decorator(func: Callable[[str], Awaitable[None]]):
logger.debug("Registering handler for UnsubscribeRequest")
@ -468,7 +468,7 @@ class Server(Generic[LifespanResultT, RequestT]):
result = await wrapper(req)
# Handle both old style (list[Tool]) and new style (ListToolsResult)
if isinstance(result, types.ListToolsResult): # pragma: no cover
if isinstance(result, types.ListToolsResult):
# Refresh the tool cache with returned tools
for tool in result.tools:
validate_and_warn_tool_name(tool.name)
@ -571,7 +571,7 @@ class Server(Generic[LifespanResultT, RequestT]):
# tool returned structured content only
maybe_structured_content = cast(StructuredContent, results)
unstructured_content = [types.TextContent(type="text", text=json.dumps(results, indent=2))]
elif hasattr(results, "__iter__"): # pragma: no cover
elif hasattr(results, "__iter__"):
# tool returned unstructured content only
unstructured_content = cast(UnstructuredContent, results)
maybe_structured_content = None
@ -714,7 +714,7 @@ class Server(Generic[LifespanResultT, RequestT]):
await self._handle_request(
message, responder.request, session, lifespan_context, raise_exceptions
)
case Exception(): # pragma: no cover
case Exception():
logger.error(f"Received exception from stream: {message}")
await session.send_log_message(
level="error",
@ -726,7 +726,7 @@ class Server(Generic[LifespanResultT, RequestT]):
case _:
await self._handle_notification(message)
for warning in w: # pragma: no cover
for warning in w: # pragma: lax no cover
logger.info("Warning: %s: %s", warning.category.__name__, warning.message)
async def _handle_request(
@ -781,16 +781,16 @@ class Server(Generic[LifespanResultT, RequestT]):
)
)
response = await handler(req)
except McpError as err: # pragma: no cover
except McpError as err:
response = err.error
except anyio.get_cancelled_exc_class(): # pragma: no cover
except anyio.get_cancelled_exc_class():
logger.info(
"Request %s cancelled - duplicate response suppressed",
message.request_id,
)
return
except Exception as err: # pragma: no cover
if raise_exceptions:
except Exception as err:
if raise_exceptions: # pragma: no cover
raise err
response = types.ErrorData(code=0, message=str(err), data=None)
finally: