refactor: move inline imports to module level

src/mcp/client/__main__.py

@@ -1,6 +1,7 @@
 import argparse
 import logging
 import sys
+import warnings
 from functools import partial
 from urllib.parse import urlparse
 
@@ -15,8 +16,6 @@
 from mcp.shared.session import RequestResponder
 
 if not sys.warnoptions:
-    import warnings
-
     warnings.simplefilter("ignore")
 
 logging.basicConfig(level=logging.INFO)

src/mcp/client/auth/extensions/client_credentials.py

@@ -9,6 +9,7 @@
 """
 
 import time
+import warnings
 from collections.abc import Awaitable, Callable
 from typing import Any, Literal
 from uuid import uuid4
@@ -409,8 +410,6 @@ def __init__(
         timeout: float = 300.0,
         jwt_parameters: JWTParameters | None = None,
     ) -> None:
-        import warnings
-
         warnings.warn(
             "RFC7523OAuthClientProvider is deprecated. Use ClientCredentialsOAuthProvider "
             "or PrivateKeyJWTOAuthProvider instead.",

src/mcp/server/__main__.py

@@ -1,6 +1,7 @@
 import importlib.metadata
 import logging
 import sys
+import warnings
 
 import anyio
 
@@ -10,8 +11,6 @@
 from mcp.types import ServerCapabilities
 
 if not sys.warnoptions:
-    import warnings
-
     warnings.simplefilter("ignore")
 
 logging.basicConfig(level=logging.INFO)

src/mcp/server/lowlevel/server.py

@@ -74,6 +74,7 @@ async def main():
 import warnings
 from collections.abc import AsyncIterator, Awaitable, Callable, Iterable
 from contextlib import AbstractAsyncContextManager, AsyncExitStack, asynccontextmanager
+from importlib.metadata import version as importlib_version
 from typing import Any, Generic, TypeAlias, cast
 
 import anyio
@@ -173,9 +174,7 @@ def create_initialization_options(
 
         def pkg_version(package: str) -> str:
             try:
-                from importlib.metadata import version
-
-                return version(package)
+                return importlib_version(package)
             except Exception:  # pragma: no cover
                 pass
 

tests/client/auth/extensions/test_client_credentials.py

@@ -1,4 +1,5 @@
 import urllib.parse
+import warnings
 
 import jwt
 import pytest
@@ -60,8 +61,6 @@ async def callback_handler() -> tuple[str, str | None]:  # pragma: no cover
         """Mock callback handler."""
         return "test_auth_code", "test_state"
 
-    import warnings
-
     with warnings.catch_warnings():
         warnings.simplefilter("ignore", DeprecationWarning)
         return RFC7523OAuthClientProvider(

tests/client/test_auth.py

@@ -5,7 +5,7 @@
 import base64
 import time
 from unittest import mock
-from urllib.parse import unquote
+from urllib.parse import parse_qs, quote, unquote, urlparse
 
 import httpx
 import pytest
@@ -27,6 +27,8 @@
     is_valid_client_metadata_url,
     should_use_client_metadata_url,
 )
+from mcp.server.auth.routes import build_metadata
+from mcp.server.auth.settings import ClientRegistrationOptions, RevocationOptions
 from mcp.shared.auth import (
     OAuthClientInformationFull,
     OAuthClientMetadata,
@@ -758,8 +760,6 @@ async def test_resource_param_included_with_recent_protocol_version(self, oauth_
         content = request.content.decode()
         assert "resource=" in content
         # Check URL-encoded resource parameter
-        from urllib.parse import quote
-
         expected_resource = quote(oauth_provider.context.get_resource_url(), safe="")
         assert f"resource={expected_resource}" in content
 
@@ -1226,8 +1226,6 @@ async def capture_redirect(url: str) -> None:
                 "%3A", ":"
             ).replace("+", " ")
             # Extract state from redirect URL
-            from urllib.parse import parse_qs, urlparse
-
             parsed = urlparse(url)
             params = parse_qs(parsed.query)
             captured_state = params.get("state", [None])[0]
@@ -1336,9 +1334,6 @@ def test_build_metadata(
     registration_endpoint: str,
     revocation_endpoint: str,
 ):
-    from mcp.server.auth.routes import build_metadata
-    from mcp.server.auth.settings import ClientRegistrationOptions, RevocationOptions
-
     metadata = build_metadata(
         issuer_url=AnyHttpUrl(issuer_url),
         service_documentation_url=AnyHttpUrl(service_documentation_url),

tests/client/test_http_unicode.py

@@ -7,12 +7,20 @@
 
 import multiprocessing
 import socket
-from collections.abc import Generator
+from collections.abc import AsyncGenerator, Generator
+from contextlib import asynccontextmanager
+from typing import Any
 
 import pytest
+from starlette.applications import Starlette
+from starlette.routing import Mount
 
+import mcp.types as types
 from mcp.client.session import ClientSession
 from mcp.client.streamable_http import streamable_http_client
+from mcp.server import Server
+from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
+from mcp.types import TextContent, Tool
 from tests.test_helpers import wait_for_server
 
 # Test constants with various Unicode characters
@@ -37,19 +45,7 @@
 
 def run_unicode_server(port: int) -> None:  # pragma: no cover
     """Run the Unicode test server in a separate process."""
-    # Import inside the function since this runs in a separate process
-    from collections.abc import AsyncGenerator
-    from contextlib import asynccontextmanager
-    from typing import Any
-
     import uvicorn
-    from starlette.applications import Starlette
-    from starlette.routing import Mount
-
-    import mcp.types as types
-    from mcp.server import Server
-    from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
-    from mcp.types import TextContent, Tool
 
     # Need to recreate the server setup in this process
     server = Server(name="unicode_test_server")

tests/client/test_list_roots_callback.py

@@ -2,6 +2,7 @@
 from pydantic import FileUrl
 
 from mcp.client.session import ClientSession
+from mcp.server.fastmcp import FastMCP
 from mcp.server.fastmcp.server import Context
 from mcp.server.session import ServerSession
 from mcp.shared.context import RequestContext
@@ -13,8 +14,6 @@
 
 @pytest.mark.anyio
 async def test_list_roots_callback():
-    from mcp.server.fastmcp import FastMCP
-
     server = FastMCP("test")
 
     callback_return = ListRootsResult(

tests/client/test_logging_callback.py

@@ -3,6 +3,7 @@
 import pytest
 
 import mcp.types as types
+from mcp.server.fastmcp import FastMCP
 from mcp.shared.memory import (
     create_connected_server_and_client_session as create_session,
 )
@@ -23,8 +24,6 @@ async def __call__(self, params: LoggingMessageNotificationParams) -> None:
 
 @pytest.mark.anyio
 async def test_logging_callback():
-    from mcp.server.fastmcp import FastMCP
-
     server = FastMCP("test")
     logging_collector = LoggingCollector()
 

tests/client/test_output_schema_validation.py

@@ -1,8 +1,10 @@
+import inspect
 import logging
 from contextlib import contextmanager
 from typing import Any
 from unittest.mock import patch
 
+import jsonschema
 import pytest
 
 from mcp.server.lowlevel import Server
@@ -19,15 +21,11 @@ def bypass_server_output_validation():
     This simulates a malicious or non-compliant server that doesn't validate
     its outputs, allowing us to test client-side validation.
     """
-    import jsonschema
-
     # Save the original validate function
     original_validate = jsonschema.validate
 
     # Create a mock that tracks which module is calling it
     def selective_mock(instance: Any = None, schema: Any = None, *args: Any, **kwargs: Any) -> None:
-        import inspect
-
         # Check the call stack to see where this is being called from
         for frame_info in inspect.stack():
             # If called from the server module, skip validation

tests/client/test_sampling_callback.py

@@ -1,6 +1,7 @@
 import pytest
 
 from mcp.client.session import ClientSession
+from mcp.server.fastmcp import FastMCP
 from mcp.shared.context import RequestContext
 from mcp.shared.memory import (
     create_connected_server_and_client_session as create_session,
@@ -17,8 +18,6 @@
 
 @pytest.mark.anyio
 async def test_sampling_callback():
-    from mcp.server.fastmcp import FastMCP
-
     server = FastMCP("test")
 
     callback_return = CreateMessageResult(
@@ -63,8 +62,6 @@ async def test_sampling_tool(message: str):
 @pytest.mark.anyio
 async def test_create_message_backwards_compat_single_content():
     """Test backwards compatibility: create_message without tools returns single content."""
-    from mcp.server.fastmcp import FastMCP
-
     server = FastMCP("test")
 
     # Callback returns single content (text)

tests/client/test_session_group.py

@@ -1,6 +1,7 @@
 import contextlib
 from unittest import mock
 
+import httpx
 import pytest
 
 import mcp
@@ -356,8 +357,6 @@ async def test_establish_session_parameterized(
                     assert isinstance(server_params_instance, StreamableHttpParameters)
                     # Verify streamable_http_client was called with url, httpx_client, and terminate_on_close
                     # The http_client is created by the real create_mcp_http_client
-                    import httpx
-
                     call_args = mock_specific_client_func.call_args
                     assert call_args.kwargs["url"] == server_params_instance.url
                     assert call_args.kwargs["terminate_on_close"] == server_params_instance.terminate_on_close

tests/client/test_stdio.py

@@ -10,7 +10,12 @@
 import pytest
 
 from mcp.client.session import ClientSession
-from mcp.client.stdio import StdioServerParameters, _create_platform_compatible_process, stdio_client
+from mcp.client.stdio import (
+    StdioServerParameters,
+    _create_platform_compatible_process,
+    _terminate_process_tree,
+    stdio_client,
+)
 from mcp.shared.exceptions import McpError
 from mcp.shared.message import SessionMessage
 from mcp.types import CONNECTION_CLOSED, JSONRPCMessage, JSONRPCRequest, JSONRPCResponse
@@ -312,8 +317,6 @@ async def test_basic_child_process_cleanup(self):
 
             # Terminate using our function
             print("Terminating process and children...")
-            from mcp.client.stdio import _terminate_process_tree
-
             await _terminate_process_tree(proc)
 
             # Verify processes stopped
@@ -413,8 +416,6 @@ async def test_nested_process_tree(self):
                     assert new_size > initial_size, f"{name} process should be writing"
 
             # Terminate the whole tree
-            from mcp.client.stdio import _terminate_process_tree
-
             await _terminate_process_tree(proc)
 
             # Verify all stopped
@@ -494,8 +495,6 @@ def handle_term(sig, frame):
                 assert size2 > size1, "Child should be writing"
 
             # Terminate - this will kill the process group even if parent exits first
-            from mcp.client.stdio import _terminate_process_tree
-
             await _terminate_process_tree(proc)
 
             # Verify child stopped

tests/server/auth/test_error_handling.py

@@ -2,6 +2,9 @@
 Tests for OAuth error handling in the auth handlers.
 """
 
+import base64
+import hashlib
+import secrets
 import unittest.mock
 from typing import Any
 from urllib.parse import parse_qs, urlparse
@@ -14,6 +17,7 @@
 
 from mcp.server.auth.provider import AuthorizeError, RegistrationError, TokenError
 from mcp.server.auth.routes import create_auth_routes
+from mcp.server.auth.settings import ClientRegistrationOptions, RevocationOptions
 from tests.server.fastmcp.auth.test_auth_integration import MockOAuthProvider
 
 
@@ -25,8 +29,6 @@ def oauth_provider():
 
 @pytest.fixture
 def app(oauth_provider: MockOAuthProvider):
-    from mcp.server.auth.settings import ClientRegistrationOptions, RevocationOptions
-
     # Enable client registration
     client_registration_options = ClientRegistrationOptions(enabled=True)
     revocation_options = RevocationOptions(enabled=True)
@@ -53,10 +55,6 @@ def client(app: Starlette):
 @pytest.fixture
 def pkce_challenge():
     """Create a PKCE challenge with code_verifier and code_challenge."""
-    import base64
-    import hashlib
-    import secrets
-
     # Generate a code verifier
     code_verifier = secrets.token_urlsafe(64)[:128]
 

tests/server/auth/test_protected_resource.py

@@ -2,6 +2,8 @@
 Integration tests for MCP Oauth Protected Resource.
 """
 
+from urllib.parse import urlparse
+
 import httpx
 import pytest
 from inline_snapshot import snapshot
@@ -159,8 +161,6 @@ def test_route_path_matches_metadata_url(self):
         )
 
         # Extract path from metadata URL
-        from urllib.parse import urlparse
-
         metadata_path = urlparse(str(metadata_url)).path
 
         # Verify consistency
@@ -181,8 +181,6 @@ def test_consistent_paths_for_various_resources(self, resource_url: str, expecte
 
         # Test URL generation
         metadata_url = build_resource_metadata_url(resource_url_obj)
-        from urllib.parse import urlparse
-
         url_path = urlparse(str(metadata_url)).path
 
         # Test route creation