Add basic server primitives

src/fastmcp/server/auth/auth.py

@@ -0,0 +1,38 @@
+from mcp.server.auth.provider import (
+    AccessToken,
+    AuthorizationCode,
+    OAuthAuthorizationServerProvider,
+    RefreshToken,
+)
+from mcp.server.auth.settings import (
+    AuthSettings,
+    ClientRegistrationOptions,
+    RevocationOptions,
+)
+from pydantic import AnyHttpUrl
+
+
+class OAuthProvider(
+    OAuthAuthorizationServerProvider[AuthorizationCode, RefreshToken, AccessToken]
+):
+    def __init__(
+        self,
+        issuer_url: AnyHttpUrl | str,
+        service_documentation_url: AnyHttpUrl | str | None = None,
+        client_registration_options: ClientRegistrationOptions | None = None,
+        revocation_options: RevocationOptions | None = None,
+        required_scopes: list[str] | None = None,
+    ):
+        super().__init__()
+        if isinstance(issuer_url, str):
+            issuer_url = AnyHttpUrl(issuer_url)
+        if isinstance(service_documentation_url, str):
+            service_documentation_url = AnyHttpUrl(service_documentation_url)
+
+        self.settings = AuthSettings(
+            issuer_url=issuer_url,
+            service_documentation_url=service_documentation_url,
+            client_registration_options=client_registration_options,
+            revocation_options=revocation_options,
+            required_scopes=required_scopes,
+        )

src/fastmcp/server/auth/in_memory_provider.py

@@ -0,0 +1,326 @@
+import secrets
+import time
+
+from mcp.server.auth.provider import (
+    AccessToken,
+    AuthorizationCode,
+    AuthorizationParams,
+    AuthorizeError,
+    RefreshToken,
+    TokenError,
+    construct_redirect_uri,
+)
+from mcp.shared.auth import (
+    OAuthClientInformationFull,
+    OAuthToken,
+)
+from pydantic import AnyHttpUrl
+
+from fastmcp.server.auth.auth import (
+    ClientRegistrationOptions,
+    OAuthProvider,
+    RevocationOptions,
+)
+
+# Default expiration times (in seconds)
+DEFAULT_AUTH_CODE_EXPIRY_SECONDS = 5 * 60  # 5 minutes
+DEFAULT_ACCESS_TOKEN_EXPIRY_SECONDS = 60 * 60  # 1 hour
+# Refresh tokens often have longer or no expiry; let's make them non-expiring for simplicity
+DEFAULT_REFRESH_TOKEN_EXPIRY_SECONDS = None
+
+
+class InMemoryOAuthProvider(OAuthProvider):
+    """
+    An in-memory OAuth provider for testing purposes.
+    It simulates the OAuth 2.0 flow locally without external calls.
+    """
+
+    def __init__(
+        self,
+        issuer_url: AnyHttpUrl | str | None = None,
+        service_documentation_url: AnyHttpUrl | str | None = None,
+        client_registration_options: ClientRegistrationOptions | None = None,
+        revocation_options: RevocationOptions | None = None,
+        required_scopes: list[str] | None = None,
+    ):
+        super().__init__(
+            issuer_url or "https://example.com",
+            service_documentation_url=service_documentation_url,
+            client_registration_options=client_registration_options,
+            revocation_options=revocation_options,
+            required_scopes=required_scopes,
+        )
+        self.clients: dict[str, OAuthClientInformationFull] = {}
+        self.auth_codes: dict[str, AuthorizationCode] = {}
+        self.access_tokens: dict[str, AccessToken] = {}
+        self.refresh_tokens: dict[str, RefreshToken] = {}
+
+        # For revoking associated tokens
+        self._access_to_refresh_map: dict[
+            str, str
+        ] = {}  # access_token_str -> refresh_token_str
+        self._refresh_to_access_map: dict[
+            str, str
+        ] = {}  # refresh_token_str -> access_token_str
+
+    async def get_client(self, client_id: str) -> OAuthClientInformationFull | None:
+        return self.clients.get(client_id)
+
+    async def register_client(self, client_info: OAuthClientInformationFull) -> None:
+        if client_info.client_id in self.clients:
+            # As per RFC 7591, if client_id is already known, it's an update.
+            # For this simple provider, we'll treat it as re-registration.
+            # A real provider might handle updates or raise errors for conflicts.
+            pass
+        self.clients[client_info.client_id] = client_info
+
+    async def authorize(
+        self, client: OAuthClientInformationFull, params: AuthorizationParams
+    ) -> str:
+        """
+        Simulates user authorization and generates an authorization code.
+        Returns a redirect URI with the code and state.
+        """
+        if client.client_id not in self.clients:
+            raise AuthorizeError(
+                error="unauthorized_client",
+                error_description=f"Client '{client.client_id}' not registered.",
+            )
+
+        # Validate redirect_uri (already validated by AuthorizationHandler, but good practice)
+        try:
+            # OAuthClientInformationFull should have a method like validate_redirect_uri
+            # For this test provider, we assume it's valid if it matches one in client_info
+            # The AuthorizationHandler already does robust validation using client.validate_redirect_uri
+            if params.redirect_uri not in client.redirect_uris:
+                # This check might be too simplistic if redirect_uris can be patterns
+                # or if params.redirect_uri is None and client has a default.
+                # However, the AuthorizationHandler handles the primary validation.
+                pass  # Let's assume AuthorizationHandler did its job.
+        except Exception:  # Replace with specific validation error if client.validate_redirect_uri existed
+            raise AuthorizeError(
+                error="invalid_request", error_description="Invalid redirect_uri."
+            )
+
+        auth_code_value = f"test_auth_code_{secrets.token_hex(16)}"
+        expires_at = time.time() + DEFAULT_AUTH_CODE_EXPIRY_SECONDS
+
+        # Ensure scopes are a list
+        scopes_list = params.scopes if params.scopes is not None else []
+        if client.scope:  # Filter params.scopes against client's registered scopes
+            client_allowed_scopes = set(client.scope.split())
+            scopes_list = [s for s in scopes_list if s in client_allowed_scopes]
+
+        auth_code = AuthorizationCode(
+            code=auth_code_value,
+            client_id=client.client_id,
+            redirect_uri=params.redirect_uri,
+            redirect_uri_provided_explicitly=params.redirect_uri_provided_explicitly,
+            scopes=scopes_list,
+            expires_at=expires_at,
+            code_challenge=params.code_challenge,
+            # code_challenge_method is assumed S256 by the framework
+        )
+        self.auth_codes[auth_code_value] = auth_code
+
+        return construct_redirect_uri(
+            str(params.redirect_uri), code=auth_code_value, state=params.state
+        )
+
+    async def load_authorization_code(
+        self, client: OAuthClientInformationFull, authorization_code: str
+    ) -> AuthorizationCode | None:
+        auth_code_obj = self.auth_codes.get(authorization_code)
+        if auth_code_obj:
+            if auth_code_obj.client_id != client.client_id:
+                return None  # Belongs to a different client
+            if auth_code_obj.expires_at < time.time():
+                del self.auth_codes[authorization_code]  # Expired
+                return None
+            return auth_code_obj
+        return None
+
+    async def exchange_authorization_code(
+        self, client: OAuthClientInformationFull, authorization_code: AuthorizationCode
+    ) -> OAuthToken:
+        # Authorization code should have been validated (existence, expiry, client_id match)
+        # by the TokenHandler calling load_authorization_code before this.
+        # We might want to re-verify or simply trust it's valid.
+
+        if authorization_code.code not in self.auth_codes:
+            raise TokenError(
+                "invalid_grant", "Authorization code not found or already used."
+            )
+
+        # Consume the auth code
+        del self.auth_codes[authorization_code.code]
+
+        access_token_value = f"test_access_token_{secrets.token_hex(32)}"
+        refresh_token_value = f"test_refresh_token_{secrets.token_hex(32)}"
+
+        access_token_expires_at = int(time.time() + DEFAULT_ACCESS_TOKEN_EXPIRY_SECONDS)
+
+        # Refresh token expiry
+        refresh_token_expires_at = None
+        if DEFAULT_REFRESH_TOKEN_EXPIRY_SECONDS is not None:
+            refresh_token_expires_at = int(
+                time.time() + DEFAULT_REFRESH_TOKEN_EXPIRY_SECONDS
+            )
+
+        self.access_tokens[access_token_value] = AccessToken(
+            token=access_token_value,
+            client_id=client.client_id,
+            scopes=authorization_code.scopes,
+            expires_at=access_token_expires_at,
+        )
+        self.refresh_tokens[refresh_token_value] = RefreshToken(
+            token=refresh_token_value,
+            client_id=client.client_id,
+            scopes=authorization_code.scopes,  # Refresh token inherits scopes
+            expires_at=refresh_token_expires_at,
+        )
+
+        self._access_to_refresh_map[access_token_value] = refresh_token_value
+        self._refresh_to_access_map[refresh_token_value] = access_token_value
+
+        return OAuthToken(
+            access_token=access_token_value,
+            token_type="bearer",
+            expires_in=DEFAULT_ACCESS_TOKEN_EXPIRY_SECONDS,
+            refresh_token=refresh_token_value,
+            scope=" ".join(authorization_code.scopes),
+        )
+
+    async def load_refresh_token(
+        self, client: OAuthClientInformationFull, refresh_token: str
+    ) -> RefreshToken | None:
+        token_obj = self.refresh_tokens.get(refresh_token)
+        if token_obj:
+            if token_obj.client_id != client.client_id:
+                return None  # Belongs to different client
+            if token_obj.expires_at is not None and token_obj.expires_at < time.time():
+                self._revoke_internal(
+                    refresh_token_str=token_obj.token
+                )  # Clean up expired
+                return None
+            return token_obj
+        return None
+
+    async def exchange_refresh_token(
+        self,
+        client: OAuthClientInformationFull,
+        refresh_token: RefreshToken,  # This is the RefreshToken object, already loaded
+        scopes: list[str],  # Requested scopes for the new access token
+    ) -> OAuthToken:
+        # Validate scopes: requested scopes must be a subset of original scopes
+        original_scopes = set(refresh_token.scopes)
+        requested_scopes = set(scopes)
+        if not requested_scopes.issubset(original_scopes):
+            raise TokenError(
+                "invalid_scope",
+                "Requested scopes exceed those authorized by the refresh token.",
+            )
+
+        # Invalidate old refresh token and its associated access token (rotation)
+        self._revoke_internal(refresh_token_str=refresh_token.token)
+
+        # Issue new tokens
+        new_access_token_value = f"test_access_token_{secrets.token_hex(32)}"
+        new_refresh_token_value = f"test_refresh_token_{secrets.token_hex(32)}"
+
+        access_token_expires_at = int(time.time() + DEFAULT_ACCESS_TOKEN_EXPIRY_SECONDS)
+
+        # Refresh token expiry
+        refresh_token_expires_at = None
+        if DEFAULT_REFRESH_TOKEN_EXPIRY_SECONDS is not None:
+            refresh_token_expires_at = int(
+                time.time() + DEFAULT_REFRESH_TOKEN_EXPIRY_SECONDS
+            )
+
+        self.access_tokens[new_access_token_value] = AccessToken(
+            token=new_access_token_value,
+            client_id=client.client_id,
+            scopes=scopes,  # Use newly requested (and validated) scopes
+            expires_at=access_token_expires_at,
+        )
+        self.refresh_tokens[new_refresh_token_value] = RefreshToken(
+            token=new_refresh_token_value,
+            client_id=client.client_id,
+            scopes=scopes,  # New refresh token also gets these scopes
+            expires_at=refresh_token_expires_at,
+        )
+
+        self._access_to_refresh_map[new_access_token_value] = new_refresh_token_value
+        self._refresh_to_access_map[new_refresh_token_value] = new_access_token_value
+
+        return OAuthToken(
+            access_token=new_access_token_value,
+            token_type="bearer",
+            expires_in=DEFAULT_ACCESS_TOKEN_EXPIRY_SECONDS,
+            refresh_token=new_refresh_token_value,
+            scope=" ".join(scopes),
+        )
+
+    async def load_access_token(self, token: str) -> AccessToken | None:
+        token_obj = self.access_tokens.get(token)
+        if token_obj:
+            if token_obj.expires_at is not None and token_obj.expires_at < time.time():
+                self._revoke_internal(
+                    access_token_str=token_obj.token
+                )  # Clean up expired
+                return None
+            return token_obj
+        return None
+
+    def _revoke_internal(
+        self, access_token_str: str | None = None, refresh_token_str: str | None = None
+    ):
+        """Internal helper to remove tokens and their associations."""
+        removed_access_token = None
+        removed_refresh_token = None
+
+        if access_token_str:
+            if access_token_str in self.access_tokens:
+                del self.access_tokens[access_token_str]
+                removed_access_token = access_token_str
+
+            # Get associated refresh token
+            associated_refresh = self._access_to_refresh_map.pop(access_token_str, None)
+            if associated_refresh:
+                if associated_refresh in self.refresh_tokens:
+                    del self.refresh_tokens[associated_refresh]
+                    removed_refresh_token = associated_refresh
+                self._refresh_to_access_map.pop(associated_refresh, None)
+
+        if refresh_token_str:
+            if refresh_token_str in self.refresh_tokens:
+                del self.refresh_tokens[refresh_token_str]
+                removed_refresh_token = refresh_token_str
+
+            # Get associated access token
+            associated_access = self._refresh_to_access_map.pop(refresh_token_str, None)
+            if associated_access:
+                if associated_access in self.access_tokens:
+                    del self.access_tokens[associated_access]
+                    removed_access_token = associated_access
+                self._access_to_refresh_map.pop(associated_access, None)
+
+        # Clean up any dangling references if one part of the pair was already gone
+        if removed_access_token and removed_access_token in self._access_to_refresh_map:
+            del self._access_to_refresh_map[removed_access_token]
+        if (
+            removed_refresh_token
+            and removed_refresh_token in self._refresh_to_access_map
+        ):
+            del self._refresh_to_access_map[removed_refresh_token]
+
+    async def revoke_token(
+        self,
+        token: AccessToken | RefreshToken,
+    ) -> None:
+        """Revokes an access or refresh token and its counterpart."""
+        if isinstance(token, AccessToken):
+            self._revoke_internal(access_token_str=token.token)
+        elif isinstance(token, RefreshToken):
+            self._revoke_internal(refresh_token_str=token.token)
+        # If token is not found or already revoked, _revoke_internal does nothing, which is correct.

src/fastmcp/server/http.py

@@ -10,14 +10,7 @@ from mcp.server.auth.middleware.bearer_auth import (
     BearerAuthBackend,
     RequireAuthMiddleware,
 )
-from mcp.server.auth.provider import (
-    AccessTokenT,
-    AuthorizationCodeT,
-    OAuthAuthorizationServerProvider,
-    RefreshTokenT,
-)
 from mcp.server.auth.routes import create_auth_routes
-from mcp.server.auth.settings import AuthSettings
 from mcp.server.lowlevel.server import LifespanResultT
 from mcp.server.sse import SseServerTransport
 from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
@@ -29,6 +22,7 @@ from starlette.responses import Response
 from starlette.routing import BaseRoute, Mount, Route
 from starlette.types import Lifespan, Receive, Scope, Send
 
+from fastmcp.server.auth.auth import OAuthProvider
 from fastmcp.utilities.logging import get_logger
 
 if TYPE_CHECKING:
@@ -75,17 +69,12 @@ class RequestContextMiddleware:
 
 
 def setup_auth_middleware_and_routes(
-    auth_server_provider: OAuthAuthorizationServerProvider[
-        AuthorizationCodeT, RefreshTokenT, AccessTokenT
-    ]
-    | None,
-    auth_settings: AuthSettings | None,
+    auth: OAuthProvider,
 ) -> tuple[list[Middleware], list[BaseRoute], list[str]]:
     """Set up authentication middleware and routes if auth is enabled.
 
     Args:
-        auth_server_provider: The OAuth authorization server provider
-        auth_settings: The auth settings
+        auth: The OAuthProvider authorization server provider
 
     Returns:
         Tuple of (middleware, auth_routes, required_scopes)
@@ -94,31 +83,25 @@ def setup_auth_middleware_and_routes(
     auth_routes: list[BaseRoute] = []
     required_scopes: list[str] = []
 
-    if auth_server_provider:
-        if not auth_settings:
-            raise ValueError(
-                "auth_settings must be provided when auth_server_provider is specified"
-            )
+    middleware = [
+        Middleware(
+            AuthenticationMiddleware,
+            backend=BearerAuthBackend(provider=auth),
+        ),
+        Middleware(AuthContextMiddleware),
+    ]
 
-        middleware = [
-            Middleware(
-                AuthenticationMiddleware,
-                backend=BearerAuthBackend(provider=auth_server_provider),
-            ),
-            Middleware(AuthContextMiddleware),
-        ]
-
-        required_scopes = auth_settings.required_scopes or []
-
-        auth_routes.extend(
-            create_auth_routes(
-                provider=auth_server_provider,
-                issuer_url=auth_settings.issuer_url,
-                service_documentation_url=auth_settings.service_documentation_url,
-                client_registration_options=auth_settings.client_registration_options,
-                revocation_options=auth_settings.revocation_options,
-            )
+    required_scopes = auth.settings.required_scopes or []
+
+    auth_routes.extend(
+        create_auth_routes(
+            provider=auth,
+            issuer_url=auth.settings.issuer_url,
+            service_documentation_url=auth.settings.service_documentation_url,
+            client_registration_options=auth.settings.client_registration_options,
+            revocation_options=auth.settings.revocation_options,
         )
+    )
 
     return middleware, auth_routes, required_scopes
 
@@ -155,11 +138,7 @@ def create_sse_app(
     server: FastMCP[LifespanResultT],
     message_path: str,
     sse_path: str,
-    auth_server_provider: OAuthAuthorizationServerProvider[
-        AuthorizationCodeT, RefreshTokenT, AccessTokenT
-    ]
-    | None = None,
-    auth_settings: AuthSettings | None = None,
+    auth: OAuthProvider | None = None,
     debug: bool = False,
     routes: list[BaseRoute] | None = None,
     middleware: list[Middleware] | None = None,
@@ -170,8 +149,7 @@ def create_sse_app(
         server: The FastMCP server instance
         message_path: Path for SSE messages
         sse_path: Path for SSE connections
-        auth_server_provider: Optional auth provider
-        auth_settings: Optional auth settings
+        auth: Optional auth provider
         debug: Whether to enable debug mode
         routes: Optional list of custom routes
         middleware: Optional list of middleware
@@ -196,15 +174,15 @@ def create_sse_app(
         return Response()
 
     # Get auth middleware and routes
-    auth_middleware, auth_routes, required_scopes = setup_auth_middleware_and_routes(
-        auth_server_provider, auth_settings
-    )
-
-    server_routes.extend(auth_routes)
-    server_middleware.extend(auth_middleware)
 
     # Add SSE routes with or without auth
-    if auth_server_provider:
+    if auth:
+        auth_middleware, auth_routes, required_scopes = (
+            setup_auth_middleware_and_routes(auth)
+        )
+
+        server_routes.extend(auth_routes)
+        server_middleware.extend(auth_middleware)
         # Auth is enabled, wrap endpoints with RequireAuthMiddleware
         server_routes.append(
             Route(
@@ -264,11 +242,7 @@ def create_streamable_http_app(
     server: FastMCP[LifespanResultT],
     streamable_http_path: str,
     event_store: None = None,
-    auth_server_provider: OAuthAuthorizationServerProvider[
-        AuthorizationCodeT, RefreshTokenT, AccessTokenT
-    ]
-    | None = None,
-    auth_settings: AuthSettings | None = None,
+    auth: OAuthProvider | None = None,
     json_response: bool = False,
     stateless_http: bool = False,
     debug: bool = False,
@@ -281,8 +255,7 @@ def create_streamable_http_app(
         server: The FastMCP server instance
         streamable_http_path: Path for StreamableHTTP connections
         event_store: Optional event store for session management
-        auth_server_provider: Optional auth provider
-        auth_settings: Optional auth settings
+        auth: Optional auth provider
         json_response: Whether to use JSON response format
         stateless_http: Whether to use stateless mode (new transport per request)
         debug: Whether to enable debug mode
@@ -331,16 +304,15 @@ def create_streamable_http_app(
                 # Re-raise other RuntimeErrors if they don't match the specific message
                 raise
 
-    # Get auth middleware and routes
-    auth_middleware, auth_routes, required_scopes = setup_auth_middleware_and_routes(
-        auth_server_provider, auth_settings
-    )
+    # Add StreamableHTTP routes with or without auth
+    if auth:
+        auth_middleware, auth_routes, required_scopes = (
+            setup_auth_middleware_and_routes(auth)
+        )
 
-    server_routes.extend(auth_routes)
-    server_middleware.extend(auth_middleware)
+        server_routes.extend(auth_routes)
+        server_middleware.extend(auth_middleware)
 
-    # Add StreamableHTTP routes with or without auth
-    if auth_server_provider:
         # Auth is enabled, wrap endpoint with RequireAuthMiddleware
         server_routes.append(
             Mount(

src/fastmcp/server/server.py

@@ -18,7 +18,6 @@ from typing import TYPE_CHECKING, Any, Generic, Literal
 import anyio
 import httpx
 import uvicorn
-from mcp.server.auth.provider import OAuthAuthorizationServerProvider
 from mcp.server.lowlevel.helper_types import ReadResourceContents
 from mcp.server.lowlevel.server import LifespanResultT, NotificationOptions
 from mcp.server.lowlevel.server import Server as MCPServer
@@ -48,6 +47,7 @@ from fastmcp.prompts import Prompt, PromptManager
 from fastmcp.prompts.prompt import PromptResult
 from fastmcp.resources import Resource, ResourceManager
 from fastmcp.resources.template import ResourceTemplate
+from fastmcp.server.auth.auth import OAuthProvider
 from fastmcp.server.http import (
     StarletteWithLifespan,
     create_sse_app,
@@ -110,8 +110,7 @@ class FastMCP(Generic[LifespanResultT]):
         self,
         name: str | None = None,
         instructions: str | None = None,
-        auth_server_provider: OAuthAuthorizationServerProvider[Any, Any, Any]
-        | None = None,
+        auth: OAuthProvider | None = None,
         lifespan: (
             Callable[
                 [FastMCP[LifespanResultT]],
@@ -186,13 +185,7 @@ class FastMCP(Generic[LifespanResultT]):
             lifespan=_lifespan_wrapper(self, lifespan),
         )
 
-        if (self.settings.auth is not None) != (auth_server_provider is not None):
-            # TODO: after we support separate authorization servers (see
-            raise ValueError(
-                "settings.auth must be specified if and only if auth_server_provider "
-                "is specified"
-            )
-        self._auth_server_provider = auth_server_provider
+        self.auth = auth
 
         # Set up MCP protocol handlers
         self._setup_handlers()
@@ -903,8 +896,7 @@ class FastMCP(Generic[LifespanResultT]):
             server=self,
             message_path=message_path or self.settings.message_path,
             sse_path=path or self.settings.sse_path,
-            auth_server_provider=self._auth_server_provider,
-            auth_settings=self.settings.auth,
+            auth=self.auth,
             debug=self.settings.debug,
             middleware=middleware,
         )
@@ -951,8 +943,7 @@ class FastMCP(Generic[LifespanResultT]):
                 server=self,
                 streamable_http_path=path or self.settings.streamable_http_path,
                 event_store=None,
-                auth_server_provider=self._auth_server_provider,
-                auth_settings=self.settings.auth,
+                auth=self.auth,
                 json_response=self.settings.json_response,
                 stateless_http=self.settings.stateless_http,
                 debug=self.settings.debug,
@@ -963,8 +954,7 @@ class FastMCP(Generic[LifespanResultT]):
                 server=self,
                 message_path=self.settings.message_path,
                 sse_path=path or self.settings.sse_path,
-                auth_server_provider=self._auth_server_provider,
-                auth_settings=self.settings.auth,
+                auth=self.auth,
                 debug=self.settings.debug,
                 middleware=middleware,
             )

src/fastmcp/settings.py

@@ -4,7 +4,6 @@ import inspect
 from pathlib import Path
 from typing import Annotated, Literal
 
-from mcp.server.auth.settings import AuthSettings
 from pydantic import Field, model_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
 from typing_extensions import Self
@@ -171,8 +170,6 @@ class ServerSettings(BaseSettings):
     # cache settings (for checking mounted servers)
     cache_expiration_seconds: float = 0
 
-    auth: AuthSettings | None = None
-
     # StreamableHTTP settings
     json_response: bool = False
     stateless_http: bool = (

src/fastmcp/utilities/tests.py

@@ -94,7 +94,7 @@ def run_server_in_process(
     proc.start()
 
     # Wait for server to be running
-    max_attempts = 100
+    max_attempts = 10
     attempt = 0
     while attempt < max_attempts and proc.is_alive():
         try:
@@ -102,7 +102,10 @@ def run_server_in_process(
                 s.connect((host, port))
                 break
         except ConnectionRefusedError:
-            time.sleep(0.01)
+            if attempt < 3:
+                time.sleep(0.01)
+            else:
+                time.sleep(0.1)
             attempt += 1
     else:
         raise RuntimeError(f"Server failed to start after {max_attempts} attempts")