coverage: run ruff on long lines

Author: maxisbey

src/mcp/server/auth/provider.py

@@ -173,6 +173,7 @@ async def authorize(self, client: OAuthClientInformationFull, params: Authorizat
         Raises:
             AuthorizeError: If the authorization request is invalid.
         """
+        ...  # pragma: no cover
 
     async def load_authorization_code(
         self, client: OAuthClientInformationFull, authorization_code: str
@@ -187,6 +188,7 @@ async def load_authorization_code(
         Returns:
             The AuthorizationCode, or None if not found
         """
+        ...  # pragma: no cover
 
     async def exchange_authorization_code(
         self, client: OAuthClientInformationFull, authorization_code: AuthorizationCodeT
@@ -204,6 +206,7 @@ async def exchange_authorization_code(
         Raises:
             TokenError: If the request is invalid
         """
+        ...  # pragma: no cover
 
     async def load_refresh_token(self, client: OAuthClientInformationFull, refresh_token: str) -> RefreshTokenT | None:
         """
@@ -216,6 +219,7 @@ async def load_refresh_token(self, client: OAuthClientInformationFull, refresh_t
         Returns:
             The RefreshToken object if found, or None if not found.
         """
+        ...  # pragma: no cover
 
     async def exchange_refresh_token(
         self,
@@ -239,6 +243,7 @@ async def exchange_refresh_token(
         Raises:
             TokenError: If the request is invalid
         """
+        ...
 
     async def load_access_token(self, token: str) -> AccessTokenT | None:
         """

src/mcp/server/fastmcp/resources/types.py

@@ -171,7 +171,7 @@ class DirectoryResource(Resource):
 
     @pydantic.field_validator("path")
     @classmethod
-    def validate_absolute_path(cls, path: Path) -> Path: # pragma: no cover
+    def validate_absolute_path(cls, path: Path) -> Path:  # pragma: no cover
         """Ensure path is absolute."""
         if not path.is_absolute():
             raise ValueError("Path must be absolute")

src/mcp/server/fastmcp/server.py

@@ -1225,7 +1225,9 @@ async def log(
     @property
     def client_id(self) -> str | None:
         """Get the client ID if available."""
-        return getattr(self.request_context.meta, "client_id", None) if self.request_context.meta else None  # pragma: no cover
+        return (
+            getattr(self.request_context.meta, "client_id", None) if self.request_context.meta else None
+        )  # pragma: no cover
 
     @property
     def request_id(self) -> str:

src/mcp/server/fastmcp/utilities/func_metadata.py

@@ -477,7 +477,9 @@ class DictModel(RootModel[dict_annotation]):
 
 
 def _get_typed_annotation(annotation: Any, globalns: dict[str, Any]) -> Any:
-    def try_eval_type(value: Any, globalns: dict[str, Any], localns: dict[str, Any]) -> tuple[Any, bool]:  # pragma: no cover
+    def try_eval_type(
+        value: Any, globalns: dict[str, Any], localns: dict[str, Any]
+    ) -> tuple[Any, bool]:  # pragma: no cover
         try:
             return eval_type_backport(value, globalns, localns), True
         except NameError:

src/mcp/server/lowlevel/server.py

@@ -679,7 +679,9 @@ async def _handle_request(
             try:
                 # Extract request context from message metadata
                 request_data = None
-                if message.message_metadata is not None and isinstance(message.message_metadata, ServerMessageMetadata):  # pragma: no cover
+                if message.message_metadata is not None and isinstance(
+                    message.message_metadata, ServerMessageMetadata
+                ):  # pragma: no cover
                     request_data = message.message_metadata.request_context
 
                 # Set our global state that can be retrieved via

src/mcp/server/streamable_http.py

@@ -369,9 +369,11 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re
                 return
 
             # Check if this is an initialization request
-            is_initialization_request = isinstance(message.root, JSONRPCRequest) and message.root.method == "initialize" # pragma: no cover
+            is_initialization_request = (
+                isinstance(message.root, JSONRPCRequest) and message.root.method == "initialize"
+            )  # pragma: no cover
 
-            if is_initialization_request: # pragma: no cover
+            if is_initialization_request:  # pragma: no cover
                 # Check if the server already has an established session
                 if self.mcp_session_id:
                     # Check if request has a session ID
@@ -385,11 +387,11 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re
                         )
                         await response(scope, receive, send)
                         return
-            elif not await self._validate_request_headers(request, send): # pragma: no cover
+            elif not await self._validate_request_headers(request, send):  # pragma: no cover
                 return
 
             # For notifications and responses only, return 202 Accepted
-            if not isinstance(message.root, JSONRPCRequest): # pragma: no cover
+            if not isinstance(message.root, JSONRPCRequest):  # pragma: no cover
                 # Create response object and send it
                 response = self._create_json_response(
                     None,
@@ -405,10 +407,10 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re
                 return
 
             # Extract the request ID outside the try block for proper scope
-            request_id = str(message.root.id) # pragma: no cover
+            request_id = str(message.root.id)  # pragma: no cover
             # Register this stream for the request ID
-            self._request_streams[request_id] = anyio.create_memory_object_stream[EventMessage](0) # pragma: no cover
-            request_stream_reader = self._request_streams[request_id][1] # pragma: no cover
+            self._request_streams[request_id] = anyio.create_memory_object_stream[EventMessage](0)  # pragma: no cover
+            request_stream_reader = self._request_streams[request_id][1]  # pragma: no cover
 
             if self.is_json_response_enabled:  # pragma: no cover
                 # Process the message
@@ -453,7 +455,7 @@ async def _handle_post_request(self, scope: Scope, request: Request, receive: Re
                     await response(scope, receive, send)
                 finally:
                     await self._clean_up_memory_streams(request_id)
-            else: # pragma: no cover
+            else:  # pragma: no cover
                 # Create SSE stream
                 sse_stream_writer, sse_stream_reader = anyio.create_memory_object_stream[dict[str, str]](0)
 
@@ -509,7 +511,7 @@ async def sse_writer():
                     await sse_stream_reader.aclose()
                     await self._clean_up_memory_streams(request_id)
 
-        except Exception as err: # pragma: no cover
+        except Exception as err:  # pragma: no cover
             logger.exception("Error handling POST request")
             response = self._create_error_response(
                 f"Error handling POST request: {err}",

src/mcp/shared/session.py

@@ -46,7 +46,9 @@
 class ProgressFnT(Protocol):
     """Protocol for progress notification callbacks."""
 
-    async def __call__(self, progress: float, total: float | None, message: str | None) -> None: ...  # pragma: no branch
+    async def __call__(
+        self, progress: float, total: float | None, message: str | None
+    ) -> None: ...  # pragma: no branch
 
 
 class RequestResponder(Generic[ReceiveRequestT, SendResultT]):

tests/client/test_resource_cleanup.py

@@ -19,7 +19,9 @@ async def test_send_request_stream_cleanup():
 
     # Create a mock session with the minimal required functionality
     class TestSession(BaseSession[ClientRequest, ClientNotification, ClientResult, Any, Any]):
-        async def _send_response(self, request_id: RequestId, response: SendResultT | ErrorData) -> None:  # pragma: no cover
+        async def _send_response(
+            self, request_id: RequestId, response: SendResultT | ErrorData
+        ) -> None:  # pragma: no cover
             pass
 
     # Create streams

tests/server/fastmcp/test_elicitation.py

@@ -71,7 +71,9 @@ async def test_stdio_elicitation():
     create_ask_user_tool(mcp)
 
     # Create a custom handler for elicitation requests
-    async def elicitation_callback(context: RequestContext[ClientSession, None], params: ElicitRequestParams):  # pragma: no cover
+    async def elicitation_callback(
+        context: RequestContext[ClientSession, None], params: ElicitRequestParams
+    ):  # pragma: no cover
         if params.message == "Tool wants to ask: What is your name?":
             return ElicitResult(action="accept", content={"answer": "Test User"})
         else:
@@ -126,7 +128,9 @@ class InvalidNestedSchema(BaseModel):
     create_validation_tool("nested_model", InvalidNestedSchema)
 
     # Dummy callback (won't be called due to validation failure)
-    async def elicitation_callback(context: RequestContext[ClientSession, None], params: ElicitRequestParams):  # pragma: no cover
+    async def elicitation_callback(
+        context: RequestContext[ClientSession, None], params: ElicitRequestParams
+    ):  # pragma: no cover
         return ElicitResult(action="accept", content={})
 
     async with create_connected_server_and_client_session(
@@ -203,7 +207,9 @@ async def invalid_optional_tool(ctx: Context[ServerSession, None]) -> str:  # pr
         except TypeError as e:
             return f"Validation failed: {str(e)}"
 
-    async def elicitation_callback(context: RequestContext[ClientSession, None], params: ElicitRequestParams):  # pragma: no cover
+    async def elicitation_callback(
+        context: RequestContext[ClientSession, None], params: ElicitRequestParams
+    ):  # pragma: no cover
         return ElicitResult(action="accept", content={})
 
     await call_tool_and_assert(

tests/server/fastmcp/test_tool_manager.py

@@ -347,7 +347,9 @@ def tool_without_context(x: int) -> str:  # pragma: no cover
         tool = manager.add_tool(tool_without_context)
         assert tool.context_kwarg is None
 
-        def tool_with_parametrized_context(x: int, ctx: Context[ServerSessionT, LifespanContextT, RequestT]) -> str:  # pragma: no cover
+        def tool_with_parametrized_context(
+            x: int, ctx: Context[ServerSessionT, LifespanContextT, RequestT]
+        ) -> str:  # pragma: no cover
             return str(x)
 
         tool = manager.add_tool(tool_with_parametrized_context)