Merge branch 'main' into responsecachingmiddleware

tests/server/auth/providers/test_aws.py

@@ -0,0 +1,245 @@
+"""Unit tests for AWS Cognito OAuth provider."""
+
+import os
+from contextlib import contextmanager
+from unittest.mock import patch
+
+import pytest
+
+from fastmcp.server.auth.providers.aws import (
+    AWSCognitoProvider,
+    AWSCognitoProviderSettings,
+)
+
+
+@contextmanager
+def mock_cognito_oidc_discovery():
+    """Context manager to mock AWS Cognito OIDC discovery endpoint."""
+    mock_oidc_config = {
+        "issuer": "https://cognito-idp.us-east-1.amazonaws.com/us-east-1_XXXXXXXXX",
+        "authorization_endpoint": "https://test.auth.us-east-1.amazoncognito.com/oauth2/authorize",
+        "token_endpoint": "https://test.auth.us-east-1.amazoncognito.com/oauth2/token",
+        "jwks_uri": "https://cognito-idp.us-east-1.amazonaws.com/us-east-1_XXXXXXXXX/.well-known/jwks.json",
+        "userinfo_endpoint": "https://test.auth.us-east-1.amazoncognito.com/oauth2/userInfo",
+        "response_types_supported": ["code", "token"],
+        "subject_types_supported": ["public"],
+        "id_token_signing_alg_values_supported": ["RS256"],
+        "scopes_supported": ["openid", "email", "phone", "profile"],
+        "token_endpoint_auth_methods_supported": [
+            "client_secret_basic",
+            "client_secret_post",
+        ],
+    }
+
+    with patch("httpx.get") as mock_get:
+        mock_response = mock_get.return_value
+        mock_response.raise_for_status.return_value = None
+        mock_response.json.return_value = mock_oidc_config
+        yield
+
+
+class TestAWSCognitoProviderSettings:
+    """Test settings for AWS Cognito OAuth provider."""
+
+    def test_settings_from_env_vars(self):
+        """Test that settings can be loaded from environment variables."""
+        with patch.dict(
+            os.environ,
+            {
+                "FASTMCP_SERVER_AUTH_AWS_COGNITO_USER_POOL_ID": "us-east-1_XXXXXXXXX",
+                "FASTMCP_SERVER_AUTH_AWS_COGNITO_AWS_REGION": "us-east-1",
+                "FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_ID": "env_client_id",
+                "FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_SECRET": "env_secret",
+                "FASTMCP_SERVER_AUTH_AWS_COGNITO_BASE_URL": "https://example.com",
+                "FASTMCP_SERVER_AUTH_AWS_COGNITO_REDIRECT_PATH": "/custom/callback",
+            },
+        ):
+            settings = AWSCognitoProviderSettings()
+
+            assert settings.user_pool_id == "us-east-1_XXXXXXXXX"
+            assert settings.aws_region == "us-east-1"
+            assert settings.client_id == "env_client_id"
+            assert (
+                settings.client_secret
+                and settings.client_secret.get_secret_value() == "env_secret"
+            )
+            assert settings.base_url == "https://example.com"
+            assert settings.redirect_path == "/custom/callback"
+
+    def test_settings_explicit_override_env(self):
+        """Test that explicit settings override environment variables."""
+        with patch.dict(
+            os.environ,
+            {
+                "FASTMCP_SERVER_AUTH_AWS_COGNITO_USER_POOL_ID": "env_pool_id",
+                "FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_ID": "env_client_id",
+                "FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_SECRET": "env_secret",
+            },
+        ):
+            settings = AWSCognitoProviderSettings.model_validate(
+                {
+                    "user_pool_id": "explicit_pool_id",
+                    "client_id": "explicit_client_id",
+                    "client_secret": "explicit_secret",
+                }
+            )
+
+            assert settings.user_pool_id == "explicit_pool_id"
+            assert settings.client_id == "explicit_client_id"
+            assert (
+                settings.client_secret
+                and settings.client_secret.get_secret_value() == "explicit_secret"
+            )
+
+
+class TestAWSCognitoProvider:
+    """Test AWSCognitoProvider initialization."""
+
+    def test_init_with_explicit_params(self):
+        """Test initialization with explicit parameters."""
+        with mock_cognito_oidc_discovery():
+            provider = AWSCognitoProvider(
+                user_pool_id="us-east-1_XXXXXXXXX",
+                aws_region="us-east-1",
+                client_id="test_client",
+                client_secret="test_secret",
+                base_url="https://example.com",
+                redirect_path="/custom/callback",
+                required_scopes=["openid", "email"],
+            )
+
+            # Check that the provider was initialized correctly
+            assert provider._upstream_client_id == "test_client"
+            assert provider._upstream_client_secret.get_secret_value() == "test_secret"
+            assert (
+                str(provider.base_url) == "https://example.com/"
+            )  # URLs get normalized with trailing slash
+            assert provider._redirect_path == "/custom/callback"
+            # OIDC provider should have discovered the endpoints automatically
+            assert (
+                provider._upstream_authorization_endpoint
+                == "https://test.auth.us-east-1.amazoncognito.com/oauth2/authorize"
+            )
+            assert (
+                provider._upstream_token_endpoint
+                == "https://test.auth.us-east-1.amazoncognito.com/oauth2/token"
+            )
+
+    @pytest.mark.parametrize(
+        "scopes_env",
+        [
+            "openid,email",
+            '["openid", "email"]',
+        ],
+    )
+    def test_init_with_env_vars(self, scopes_env):
+        """Test initialization with environment variables."""
+        with patch.dict(
+            os.environ,
+            {
+                "FASTMCP_SERVER_AUTH_AWS_COGNITO_USER_POOL_ID": "us-east-1_XXXXXXXXX",
+                "FASTMCP_SERVER_AUTH_AWS_COGNITO_AWS_REGION": "us-east-1",
+                "FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_ID": "env_client_id",
+                "FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_SECRET": "env_secret",
+                "FASTMCP_SERVER_AUTH_AWS_COGNITO_BASE_URL": "https://env-example.com",
+                "FASTMCP_SERVER_AUTH_AWS_COGNITO_REQUIRED_SCOPES": scopes_env,
+            },
+        ):
+            with mock_cognito_oidc_discovery():
+                provider = AWSCognitoProvider()
+
+                assert provider._upstream_client_id == "env_client_id"
+                assert (
+                    provider._upstream_client_secret.get_secret_value() == "env_secret"
+                )
+                assert str(provider.base_url) == "https://env-example.com/"
+                assert provider._token_validator.required_scopes == ["openid", "email"]
+
+    def test_init_explicit_overrides_env(self):
+        """Test that explicit parameters override environment variables."""
+        with patch.dict(
+            os.environ,
+            {
+                "FASTMCP_SERVER_AUTH_AWS_COGNITO_USER_POOL_ID": "env_pool_id",
+                "FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_ID": "env_client_id",
+                "FASTMCP_SERVER_AUTH_AWS_COGNITO_CLIENT_SECRET": "env_secret",
+            },
+        ):
+            with mock_cognito_oidc_discovery():
+                provider = AWSCognitoProvider(
+                    user_pool_id="explicit_pool_id",
+                    client_id="explicit_client",
+                    client_secret="explicit_secret",
+                    base_url="https://example.com",
+                )
+
+                assert provider._upstream_client_id == "explicit_client"
+                assert (
+                    provider._upstream_client_secret.get_secret_value()
+                    == "explicit_secret"
+                )
+                # OIDC discovery should have configured the endpoints automatically
+                assert provider._upstream_authorization_endpoint is not None
+
+    def test_init_missing_user_pool_id_raises_error(self):
+        """Test that missing user_pool_id raises ValueError."""
+        with patch.dict(os.environ, {}, clear=True):
+            with pytest.raises(ValueError, match="user_pool_id is required"):
+                AWSCognitoProvider(
+                    client_id="test_client",
+                    client_secret="test_secret",
+                )
+
+    def test_init_missing_client_id_raises_error(self):
+        """Test that missing client_id raises ValueError."""
+        with patch.dict(os.environ, {}, clear=True):
+            with pytest.raises(ValueError, match="client_id is required"):
+                AWSCognitoProvider(
+                    user_pool_id="us-east-1_XXXXXXXXX",
+                    client_secret="test_secret",
+                )
+
+    def test_init_missing_client_secret_raises_error(self):
+        """Test that missing client_secret raises ValueError."""
+        with patch.dict(os.environ, {}, clear=True):
+            with pytest.raises(ValueError, match="client_secret is required"):
+                AWSCognitoProvider(
+                    user_pool_id="us-east-1_XXXXXXXXX",
+                    client_id="test_client",
+                )
+
+    def test_init_defaults(self):
+        """Test that default values are applied correctly."""
+        with mock_cognito_oidc_discovery():
+            provider = AWSCognitoProvider(
+                user_pool_id="us-east-1_XXXXXXXXX",
+                client_id="test_client",
+                client_secret="test_secret",
+                base_url="https://example.com",
+            )
+
+            # Check defaults
+            assert str(provider.base_url) == "https://example.com/"
+            assert provider._redirect_path == "/auth/callback"
+            assert provider._token_validator.required_scopes == ["openid"]
+            assert provider.aws_region == "eu-central-1"
+
+    def test_oidc_discovery_integration(self):
+        """Test that OIDC discovery endpoints are used correctly."""
+        with mock_cognito_oidc_discovery():
+            provider = AWSCognitoProvider(
+                user_pool_id="us-west-2_YYYYYYYY",
+                aws_region="us-west-2",
+                client_id="test_client",
+                client_secret="test_secret",
+                base_url="https://example.com",
+            )
+
+            # OIDC discovery should have configured the endpoints automatically
+            assert provider._upstream_authorization_endpoint is not None
+            assert provider._upstream_token_endpoint is not None
+            assert "amazoncognito.com" in provider._upstream_authorization_endpoint
+
+
+# Token verification functionality is now tested as part of the OIDC provider integration
+# The CognitoTokenVerifier class is an internal implementation detail

tests/server/auth/providers/test_azure.py

@@ -2,11 +2,15 @@
 
 import os
 from unittest.mock import patch
-from urllib.parse import urlparse
+from urllib.parse import parse_qs, urlparse
 
 import pytest
+from mcp.server.auth.provider import AuthorizationParams
+from mcp.shared.auth import OAuthClientInformationFull
+from pydantic import AnyUrl
 
 from fastmcp.server.auth.providers.azure import AzureProvider
+from fastmcp.server.auth.providers.jwt import JWTVerifier
 
 
 class TestAzureProvider:
@@ -95,6 +99,7 @@ class TestAzureProvider:
             client_id="test_client",
             client_secret="test_secret",
             tenant_id="test-tenant",
+            required_scopes=["User.Read"],
         )
 
         # Check defaults
@@ -109,6 +114,7 @@ class TestAzureProvider:
             client_secret="test_secret",
             tenant_id="my-tenant-id",
             base_url="https://myserver.com",
+            required_scopes=["User.Read"],
         )
 
         # Check that endpoints use the correct Azure OAuth2 v2.0 endpoints with tenant
@@ -131,6 +137,7 @@ class TestAzureProvider:
             client_id="test_client",
             client_secret="test_secret",
             tenant_id="organizations",
+            required_scopes=["User.Read"],
         )
         parsed = urlparse(provider1._upstream_authorization_endpoint)
         assert "/organizations/" in parsed.path
@@ -140,6 +147,7 @@ class TestAzureProvider:
             client_id="test_client",
             client_secret="test_secret",
             tenant_id="consumers",
+            required_scopes=["User.Read"],
         )
         parsed = urlparse(provider2._upstream_authorization_endpoint)
         assert "/consumers/" in parsed.path
@@ -162,3 +170,107 @@ class TestAzureProvider:
 
         # Provider should initialize successfully with these scopes
         assert provider is not None
+
+    def test_init_does_not_require_api_client_id_anymore(self):
+        """API client ID is no longer required; audience is client_id."""
+        provider = AzureProvider(
+            client_id="test_client",
+            client_secret="test_secret",
+            tenant_id="test-tenant",
+            required_scopes=["User.Read"],
+        )
+        assert provider is not None
+
+    def test_init_with_custom_audience_uses_jwt_verifier(self):
+        """When audience is provided, JWTVerifier is configured with JWKS and issuer."""
+        provider = AzureProvider(
+            client_id="test_client",
+            client_secret="test_secret",
+            tenant_id="my-tenant",
+            identifier_uri="api://my-api",
+            required_scopes=[".default"],
+        )
+
+        assert provider._token_validator is not None
+        assert isinstance(provider._token_validator, JWTVerifier)
+        verifier = provider._token_validator
+        assert verifier.jwks_uri is not None
+        assert verifier.jwks_uri.startswith(
+            "https://login.microsoftonline.com/my-tenant/discovery/v2.0/keys"
+        )
+        assert verifier.issuer == "https://login.microsoftonline.com/my-tenant/v2.0"
+        assert verifier.audience == "test_client"
+
+    @pytest.mark.asyncio
+    async def test_authorize_filters_resource_and_prefixes_scopes_with_audience(self):
+        """authorize() should drop resource and prefix non-openid scopes with audience."""
+        provider = AzureProvider(
+            client_id="test_client",
+            client_secret="test_secret",
+            tenant_id="common",
+            identifier_uri="api://my-api",
+            required_scopes=["read", "write"],
+            base_url="https://srv.example",
+        )
+
+        client = OAuthClientInformationFull(
+            client_id="dummy",
+            client_secret="secret",
+            redirect_uris=[AnyUrl("http://localhost:12345/callback")],
+        )
+
+        params = AuthorizationParams(
+            redirect_uri=AnyUrl("http://localhost:12345/callback"),
+            redirect_uri_provided_explicitly=True,
+            scopes=["read", "profile"],
+            state="abc",
+            code_challenge="xyz",
+            resource="https://should.be.ignored",
+        )
+
+        url = await provider.authorize(client, params)
+
+        parsed = urlparse(url)
+        qs = parse_qs(parsed.query)
+        assert "resource" not in qs
+        scope_value = qs.get("scope", [""])[0]
+        scope_parts = scope_value.split(" ") if scope_value else []
+        assert "api://my-api/read" in scope_parts
+        assert "api://my-api/profile" in scope_parts
+
+    @pytest.mark.asyncio
+    async def test_authorize_appends_unprefixed_additional_scopes(self):
+        """authorize() should append additional_authorize_scopes without prefixing them."""
+        provider = AzureProvider(
+            client_id="test_client",
+            client_secret="test_secret",
+            tenant_id="common",
+            identifier_uri="api://my-api",
+            required_scopes=["read"],
+            base_url="https://srv.example",
+            additional_authorize_scopes=["Mail.Read", "User.Read"],
+        )
+
+        client = OAuthClientInformationFull(
+            client_id="dummy",
+            client_secret="secret",
+            redirect_uris=[AnyUrl("http://localhost:12345/callback")],
+        )
+
+        params = AuthorizationParams(
+            redirect_uri=AnyUrl("http://localhost:12345/callback"),
+            redirect_uri_provided_explicitly=True,
+            scopes=["read"],
+            state="abc",
+            code_challenge="xyz",
+        )
+
+        url = await provider.authorize(client, params)
+
+        parsed = urlparse(url)
+        qs = parse_qs(parsed.query)
+        scope_value = qs.get("scope", [""])[0]
+        scope_parts = scope_value.split(" ") if scope_value else []
+        assert "api://my-api/read" in scope_parts
+        assert "Mail.Read" in scope_parts
+        assert "User.Read" in scope_parts

tests/server/auth/providers/test_descope.py

@@ -0,0 +1,170 @@
+"""Tests for Descope OAuth provider."""
+
+import os
+from collections.abc import Generator
+from unittest.mock import patch
+
+import httpx
+import pytest
+
+from fastmcp import Client, FastMCP
+from fastmcp.client.transports import StreamableHttpTransport
+from fastmcp.server.auth.providers.descope import DescopeProvider
+from fastmcp.utilities.tests import HeadlessOAuth, run_server_in_process
+
+
+class TestDescopeProvider:
+    """Test Descope OAuth provider functionality."""
+
+    def test_init_with_explicit_params(self):
+        """Test DescopeProvider initialization with explicit parameters."""
+        provider = DescopeProvider(
+            project_id="P2abc123",
+            base_url="https://myserver.com",
+            descope_base_url="https://api.descope.com",
+        )
+
+        assert provider.project_id == "P2abc123"
+        assert str(provider.base_url) == "https://myserver.com/"
+        assert str(provider.descope_base_url) == "https://api.descope.com"
+
+    @pytest.mark.parametrize(
+        "scopes_env",
+        [
+            "openid,email",
+            '["openid", "email"]',
+        ],
+    )
+    def test_init_with_env_vars(self, scopes_env):
+        """Test DescopeProvider initialization from environment variables."""
+        with patch.dict(
+            os.environ,
+            {
+                "FASTMCP_SERVER_AUTH_DESCOPEPROVIDER_PROJECT_ID": "P2env123",
+                "FASTMCP_SERVER_AUTH_DESCOPEPROVIDER_BASE_URL": "https://envserver.com",
+                "FASTMCP_SERVER_AUTH_DESCOPEPROVIDER_DESCOPE_BASE_URL": "https://api.descope.com",
+            },
+        ):
+            provider = DescopeProvider()
+
+            assert provider.project_id == "P2env123"
+            assert str(provider.base_url) == "https://envserver.com/"
+            assert str(provider.descope_base_url) == "https://api.descope.com"
+
+    def test_environment_variable_loading(self):
+        """Test that environment variables are loaded correctly."""
+        # This test verifies that the provider can be created with environment variables
+        provider = DescopeProvider(
+            project_id="P2env123", base_url="http://env-server.com"
+        )
+
+        # Should have loaded from environment
+        assert provider.project_id == "P2env123"
+        assert str(provider.base_url) == "http://env-server.com/"
+        assert str(provider.descope_base_url) == "https://api.descope.com"
+
+    def test_descope_base_url_https_prefix_handling(self):
+        """Test that descope_base_url handles missing https:// prefix."""
+        # Without https:// - should add it
+        provider1 = DescopeProvider(
+            project_id="P2abc123",
+            base_url="https://myserver.com",
+            descope_base_url="https://api.descope.com",
+        )
+        assert str(provider1.descope_base_url) == "https://api.descope.com"
+
+        # With https:// - should keep it
+        provider2 = DescopeProvider(
+            project_id="P2abc123",
+            base_url="https://myserver.com",
+            descope_base_url="https://api.descope.com",
+        )
+        assert str(provider2.descope_base_url) == "https://api.descope.com"
+
+        # With http:// - should be preserved
+        provider3 = DescopeProvider(
+            project_id="P2abc123",
+            base_url="https://myserver.com",
+            descope_base_url="http://localhost:8080",
+        )
+        assert str(provider3.descope_base_url) == "http://localhost:8080"
+
+    def test_init_defaults(self):
+        """Test that default values are applied correctly."""
+        provider = DescopeProvider(
+            project_id="P2abc123",
+            base_url="https://myserver.com",
+        )
+
+        # Check defaults
+        assert str(provider.descope_base_url) == "https://api.descope.com"
+
+    def test_jwt_verifier_configured_correctly(self):
+        """Test that JWT verifier is configured correctly."""
+        provider = DescopeProvider(
+            project_id="P2abc123",
+            base_url="https://myserver.com",
+            descope_base_url="https://api.descope.com",
+        )
+
+        # Check that JWT verifier uses the correct endpoints
+        assert (
+            provider.token_verifier.jwks_uri  # type: ignore[attr-defined]
+            == "https://api.descope.com/P2abc123/.well-known/jwks.json"
+        )
+        assert (
+            provider.token_verifier.issuer == "https://api.descope.com/v1/apps/P2abc123"  # type: ignore[attr-defined]
+        )
+        assert provider.token_verifier.audience == "P2abc123"  # type: ignore[attr-defined]
+
+
+def run_mcp_server(host: str, port: int) -> None:
+    mcp = FastMCP(
+        auth=DescopeProvider(
+            project_id="P2test123",
+            base_url="http://localhost:4321",
+            descope_base_url="https://api.descope.com",
+        )
+    )
+
+    @mcp.tool
+    def add(a: int, b: int) -> int:
+        return a + b
+
+    mcp.run(host=host, port=port, transport="http")
+
+
+@pytest.fixture
+def mcp_server_url() -> Generator[str]:
+    with run_server_in_process(run_mcp_server) as url:
+        yield f"{url}/mcp"
+
+
+@pytest.fixture()
+def client_with_headless_oauth(
+    mcp_server_url: str,
+) -> Generator[Client, None, None]:
+    """Client with headless OAuth that bypasses browser interaction."""
+    client = Client(
+        transport=StreamableHttpTransport(mcp_server_url),
+        auth=HeadlessOAuth(mcp_url=mcp_server_url),
+    )
+    yield client
+
+
+class TestDescopeProviderIntegration:
+    async def test_unauthorized_access(self, mcp_server_url: str):
+        with pytest.raises(httpx.HTTPStatusError) as exc_info:
+            async with Client(mcp_server_url) as client:
+                tools = await client.list_tools()  # noqa: F841
+
+        assert isinstance(exc_info.value, httpx.HTTPStatusError)
+        assert exc_info.value.response.status_code == 401
+        assert "tools" not in locals()
+
+    # async def test_authorized_access(self, client_with_headless_oauth: Client):
+    #     async with client_with_headless_oauth:
+    #         tools = await client_with_headless_oauth.list_tools()
+    #     assert tools is not None
+    #     assert len(tools) > 0
+    #     assert "add" in tools

tests/server/auth/providers/test_scalekit.py

@@ -0,0 +1,162 @@
+"""Tests for Scalekit OAuth provider."""
+
+import os
+from collections.abc import Generator
+from unittest.mock import patch
+
+import httpx
+import pytest
+
+from fastmcp import Client, FastMCP
+from fastmcp.client.transports import StreamableHttpTransport
+from fastmcp.server.auth.providers.scalekit import ScalekitProvider
+from fastmcp.utilities.tests import HeadlessOAuth, run_server_in_process
+
+
+class TestScalekitProvider:
+    """Test Scalekit OAuth provider functionality."""
+
+    def test_init_with_explicit_params(self):
+        """Test ScalekitProvider initialization with explicit parameters."""
+        provider = ScalekitProvider(
+            environment_url="https://my-env.scalekit.com",
+            client_id="sk_client_123",
+            resource_id="sk_resource_456",
+            mcp_url="https://myserver.com/",
+        )
+
+        assert provider.environment_url == "https://my-env.scalekit.com"
+        assert provider.client_id == "sk_client_123"
+        assert provider.resource_id == "sk_resource_456"
+        assert str(provider.mcp_url) == "https://myserver.com/"
+
+    def test_init_with_env_vars(self):
+        """Test ScalekitProvider initialization from environment variables."""
+        with patch.dict(
+            os.environ,
+            {
+                "FASTMCP_SERVER_AUTH_SCALEKITPROVIDER_ENVIRONMENT_URL": "https://env-scalekit.com",
+                "FASTMCP_SERVER_AUTH_SCALEKITPROVIDER_CLIENT_ID": "skc_123",
+                "FASTMCP_SERVER_AUTH_SCALEKITPROVIDER_RESOURCE_ID": "res_456",
+                "FASTMCP_SERVER_AUTH_SCALEKITPROVIDER_MCP_URL": "https://envserver.com/mcp",
+            },
+        ):
+            provider = ScalekitProvider()
+
+            assert provider.environment_url == "https://env-scalekit.com"
+            assert provider.client_id == "skc_123"
+            assert provider.resource_id == "res_456"
+            assert str(provider.mcp_url) == "https://envserver.com/mcp"
+
+    def test_environment_variable_loading(self):
+        """Test that environment variables are loaded correctly."""
+        provider = ScalekitProvider(
+            environment_url="https://test-env.scalekit.com",
+            client_id="sk_client_test_123",
+            resource_id="sk_resource_test_456",
+            mcp_url="http://test-server.com",
+        )
+
+        assert provider.environment_url == "https://test-env.scalekit.com"
+        assert provider.client_id == "sk_client_test_123"
+        assert provider.resource_id == "sk_resource_test_456"
+        assert str(provider.mcp_url) == "http://test-server.com/"
+
+    def test_url_trailing_slash_handling(self):
+        """Test that URLs handle trailing slashes correctly."""
+        provider = ScalekitProvider(
+            environment_url="https://my-env.scalekit.com/",
+            client_id="sk_client_123",
+            resource_id="sk_resource_456",
+            mcp_url="https://myserver.com/",
+        )
+
+        assert provider.environment_url == "https://my-env.scalekit.com"
+        assert str(provider.mcp_url) == "https://myserver.com/"
+
+    def test_jwt_verifier_configured_correctly(self):
+        """Test that JWT verifier is configured correctly."""
+        provider = ScalekitProvider(
+            environment_url="https://my-env.scalekit.com",
+            client_id="sk_client_123",
+            resource_id="sk_resource_456",
+            mcp_url="https://myserver.com/",
+        )
+
+        # Check that JWT verifier uses the correct endpoints
+        assert (
+            provider.token_verifier.jwks_uri  # type: ignore[attr-defined]
+            == "https://my-env.scalekit.com/keys"
+        )
+        assert (
+            provider.token_verifier.issuer == "https://my-env.scalekit.com"  # type: ignore[attr-defined]
+        )
+        assert provider.token_verifier.audience == "https://myserver.com/"  # type: ignore[attr-defined]
+
+    def test_authorization_servers_configuration(self):
+        """Test that authorization servers are configured correctly."""
+        provider = ScalekitProvider(
+            environment_url="https://my-env.scalekit.com",
+            client_id="sk_client_123",
+            resource_id="sk_resource_456",
+            mcp_url="https://myserver.com/",
+        )
+
+        assert len(provider.authorization_servers) == 1
+        assert (
+            str(provider.authorization_servers[0])
+            == "https://my-env.scalekit.com/resources/sk_resource_456"
+        )
+
+
+def run_mcp_server(host: str, port: int) -> None:
+    mcp = FastMCP(
+        auth=ScalekitProvider(
+            environment_url="https://test-env.scalekit.com",
+            client_id="sk_client_test_123",
+            resource_id="sk_resource_test_456",
+            mcp_url="http://localhost:4321",
+        )
+    )
+
+    @mcp.tool
+    def add(a: int, b: int) -> int:
+        return a + b
+
+    mcp.run(host=host, port=port, transport="http")
+
+
+@pytest.fixture
+def mcp_server_url() -> Generator[str]:
+    with run_server_in_process(run_mcp_server) as url:
+        yield f"{url}/mcp"
+
+
+@pytest.fixture()
+def client_with_headless_oauth(
+    mcp_server_url: str,
+) -> Generator[Client, None, None]:
+    """Client with headless OAuth that bypasses browser interaction."""
+    client = Client(
+        transport=StreamableHttpTransport(mcp_server_url),
+        auth=HeadlessOAuth(mcp_url=mcp_server_url),
+    )
+    yield client
+
+
+class TestScalekitProviderIntegration:
+    async def test_unauthorized_access(self, mcp_server_url: str):
+        with pytest.raises(httpx.HTTPStatusError) as exc_info:
+            async with Client(mcp_server_url) as client:
+                tools = await client.list_tools()  # noqa: F841
+
+        assert isinstance(exc_info.value, httpx.HTTPStatusError)
+        assert exc_info.value.response.status_code == 401
+        assert "tools" not in locals()
+
+    # async def test_authorized_access(self, client_with_headless_oauth: Client):
+    #     async with client_with_headless_oauth:
+    #         tools = await client_with_headless_oauth.list_tools()
+    #     assert tools is not None
+    #     assert len(tools) > 0
+    #     assert "add" in tools

tests/server/auth/providers/test_supabase.py

@@ -0,0 +1,165 @@
+"""Tests for Supabase Auth provider."""
+
+import os
+from collections.abc import Generator
+from unittest.mock import patch
+
+import httpx
+import pytest
+
+from fastmcp import Client, FastMCP
+from fastmcp.client.transports import StreamableHttpTransport
+from fastmcp.server.auth.providers.supabase import SupabaseProvider
+from fastmcp.utilities.tests import HeadlessOAuth, run_server_in_process
+
+
+class TestSupabaseProvider:
+    """Test Supabase Auth provider functionality."""
+
+    def test_init_with_explicit_params(self):
+        """Test SupabaseProvider initialization with explicit parameters."""
+        provider = SupabaseProvider(
+            project_url="https://abc123.supabase.co",
+            base_url="https://myserver.com",
+        )
+
+        assert provider.project_url == "https://abc123.supabase.co"
+        assert str(provider.base_url) == "https://myserver.com/"
+
+    @pytest.mark.parametrize(
+        "scopes_env",
+        [
+            "openid,email",
+            '["openid", "email"]',
+        ],
+    )
+    def test_init_with_env_vars(self, scopes_env):
+        """Test SupabaseProvider initialization from environment variables."""
+        with patch.dict(
+            os.environ,
+            {
+                "FASTMCP_SERVER_AUTH_SUPABASE_PROJECT_URL": "https://env123.supabase.co",
+                "FASTMCP_SERVER_AUTH_SUPABASE_BASE_URL": "https://envserver.com",
+            },
+        ):
+            provider = SupabaseProvider()
+
+            assert provider.project_url == "https://env123.supabase.co"
+            assert str(provider.base_url) == "https://envserver.com/"
+
+    def test_environment_variable_loading(self):
+        """Test that environment variables are loaded correctly."""
+        provider = SupabaseProvider(
+            project_url="https://env123.supabase.co",
+            base_url="http://env-server.com",
+        )
+
+        assert provider.project_url == "https://env123.supabase.co"
+        assert str(provider.base_url) == "http://env-server.com/"
+
+    def test_project_url_normalization(self):
+        """Test that project_url handles trailing slashes correctly."""
+        # Without trailing slash
+        provider1 = SupabaseProvider(
+            project_url="https://abc123.supabase.co",
+            base_url="https://myserver.com",
+        )
+        assert provider1.project_url == "https://abc123.supabase.co"
+
+        # With trailing slash - should be stripped
+        provider2 = SupabaseProvider(
+            project_url="https://abc123.supabase.co/",
+            base_url="https://myserver.com",
+        )
+        assert provider2.project_url == "https://abc123.supabase.co"
+
+    def test_jwt_verifier_configured_correctly(self):
+        """Test that JWT verifier is configured correctly."""
+        provider = SupabaseProvider(
+            project_url="https://abc123.supabase.co",
+            base_url="https://myserver.com",
+        )
+
+        # Check that JWT verifier uses the correct endpoints
+        assert (
+            provider.token_verifier.jwks_uri  # type: ignore[attr-defined]
+            == "https://abc123.supabase.co/auth/v1/.well-known/jwks.json"
+        )
+        assert (
+            provider.token_verifier.issuer == "https://abc123.supabase.co/auth/v1"  # type: ignore[attr-defined]
+        )
+        assert provider.token_verifier.algorithm == "ES256"  # type: ignore[attr-defined]
+
+    def test_jwt_verifier_with_required_scopes(self):
+        """Test that JWT verifier respects required_scopes."""
+        provider = SupabaseProvider(
+            project_url="https://abc123.supabase.co",
+            base_url="https://myserver.com",
+            required_scopes=["openid", "email"],
+        )
+
+        assert provider.token_verifier.required_scopes == ["openid", "email"]  # type: ignore[attr-defined]
+
+    def test_authorization_servers_configured(self):
+        """Test that authorization servers list is configured correctly."""
+        provider = SupabaseProvider(
+            project_url="https://abc123.supabase.co",
+            base_url="https://myserver.com",
+        )
+
+        assert len(provider.authorization_servers) == 1
+        assert (
+            str(provider.authorization_servers[0])
+            == "https://abc123.supabase.co/auth/v1"
+        )
+
+
+def run_mcp_server(host: str, port: int) -> None:
+    mcp = FastMCP(
+        auth=SupabaseProvider(
+            project_url="https://test123.supabase.co",
+            base_url="http://localhost:4321",
+        )
+    )
+
+    @mcp.tool
+    def add(a: int, b: int) -> int:
+        return a + b
+
+    mcp.run(host=host, port=port, transport="http")
+
+
+@pytest.fixture
+def mcp_server_url() -> Generator[str]:
+    with run_server_in_process(run_mcp_server) as url:
+        yield f"{url}/mcp"
+
+
+@pytest.fixture()
+def client_with_headless_oauth(
+    mcp_server_url: str,
+) -> Generator[Client, None, None]:
+    """Client with headless OAuth that bypasses browser interaction."""
+    client = Client(
+        transport=StreamableHttpTransport(mcp_server_url),
+        auth=HeadlessOAuth(mcp_url=mcp_server_url),
+    )
+    yield client
+
+
+class TestSupabaseProviderIntegration:
+    async def test_unauthorized_access(self, mcp_server_url: str):
+        with pytest.raises(httpx.HTTPStatusError) as exc_info:
+            async with Client(mcp_server_url) as client:
+                tools = await client.list_tools()  # noqa: F841
+
+        assert isinstance(exc_info.value, httpx.HTTPStatusError)
+        assert exc_info.value.response.status_code == 401
+        assert "tools" not in locals()
+
+    # async def test_authorized_access(self, client_with_headless_oauth: Client):
+    #     async with client_with_headless_oauth:
+    #         tools = await client_with_headless_oauth.list_tools()
+    #     assert tools is not None
+    #     assert len(tools) > 0
+    #     assert "add" in tools

tests/server/auth/providers/test_workos.py

@@ -172,7 +172,7 @@ def run_mcp_server(host: str, port: int) -> None:
     mcp.run(host=host, port=port, transport="http")
 
 
-@pytest.fixture(scope="module")
+@pytest.fixture
 def mcp_server_url() -> Generator[str]:
     with run_server_in_process(run_mcp_server) as url:
         yield f"{url}/mcp"

tests/server/auth/test_jwt_provider.py

@@ -132,7 +132,7 @@ def run_mcp_server(
     mcp.run(host=host, port=port, **run_kwargs or {})
 
 
-@pytest.fixture(scope="module")
+@pytest.fixture
 def mcp_server_url(rsa_key_pair: RSAKeyPair) -> Generator[str]:
     with run_server_in_process(
         run_mcp_server,

tests/server/auth/test_oauth_proxy.py

@@ -44,7 +44,7 @@ class MockOAuthProvider:
     - Network calls to external services
     """
 
-    def __init__(self, port: int = 9999):
+    def __init__(self, port: int = 0):
         self.port = port
         self.base_url = f"http://localhost:{port}"
         self.app = None
@@ -229,23 +229,40 @@ class MockOAuthProvider:
 
     async def start(self):
         """Start the mock OAuth server."""
+        import socket
+
         from uvicorn import Config, Server
 
         self.app = self.create_app()
-        config = Config(self.app, host="localhost", port=self.port, log_level="error")
+
+        # If port is 0, find an available port
+        if self.port == 0:
+            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
+                s.bind(("127.0.0.1", 0))
+                s.listen(1)
+                self.port = s.getsockname()[1]
+
+        self.base_url = f"http://localhost:{self.port}"
+        config = Config(
+            self.app,
+            host="localhost",
+            port=self.port,
+            log_level="error",
+            ws="websockets-sansio",
+        )
         self.server = Server(config)
 
         # Start server in background
         asyncio.create_task(self.server.serve())
 
         # Wait for server to be ready
-        await asyncio.sleep(0.5)
+        await asyncio.sleep(0.05)
 
     async def stop(self):
         """Stop the mock OAuth server."""
         if self.server:
             self.server.should_exit = True
-            await asyncio.sleep(0.1)
+            await asyncio.sleep(0.01)
 
     def reset(self):
         """Reset all state for next test."""
@@ -308,7 +325,7 @@ def oauth_proxy(jwt_verifier):
 @pytest.fixture
 async def mock_oauth_provider():
     """Create and start a mock OAuth provider."""
-    provider = MockOAuthProvider(port=9999)
+    provider = MockOAuthProvider()
     await provider.start()
     yield provider
     await provider.stop()
@@ -395,8 +412,8 @@ class TestOAuthProxyClientRegistration:
 
         await oauth_proxy.register_client(client_info)
 
-        # Client should be stored with original credentials
-        stored = oauth_proxy._clients.get("original-client")
+        # Client should be retrievable with original credentials
+        stored = await oauth_proxy.get_client("original-client")
         assert stored is not None
         assert stored.client_id == "original-client"
         assert stored.client_secret == "original-secret"
@@ -962,3 +979,126 @@ class TestParameterForwarding:
         assert query_params["audience"][0] == "https://api.example.com"
         assert query_params["prompt"][0] == "consent"
         assert query_params["max_age"][0] == "3600"
+
+    @pytest.mark.asyncio
+    async def test_token_endpoint_invalid_client_error(self, jwt_verifier):
+        """Test that invalid client_id returns OAuth 2.1 compliant error response.
+
+        When a client ID is not found during token exchange, the proxy should:
+        1. Return HTTP 401 status code
+        2. Use 'invalid_client' error code instead of 'unauthorized_client'
+
+        This aligns with OAuth 2.1 spec and enables Claude's automatic client re-registration.
+        """
+        from starlette.applications import Starlette
+        from starlette.testclient import TestClient
+
+        proxy = OAuthProxy(
+            upstream_authorization_endpoint="https://oauth.example.com/authorize",
+            upstream_token_endpoint="https://oauth.example.com/token",
+            upstream_client_id="upstream-client",
+            upstream_client_secret="upstream-secret",
+            token_verifier=jwt_verifier,
+            base_url="https://proxy.example.com",
+        )
+
+        # Create a test app with OAuth routes
+        app = Starlette(routes=proxy.get_routes())
+
+        # Test the token endpoint with an invalid (non-existent) client_id
+        with TestClient(app) as client:
+            response = client.post(
+                "/token",
+                data={
+                    "grant_type": "authorization_code",
+                    "code": "test-auth-code",
+                    "client_id": "non-existent-client-id",
+                    "code_verifier": "test-code-verifier",
+                    "redirect_uri": "http://localhost:12345/callback",
+                },
+                headers={
+                    "Content-Type": "application/x-www-form-urlencoded",
+                },
+            )
+
+            # Verify OAuth 2.1 compliant error response
+            assert response.status_code == 401, (
+                f"Expected 401 but got {response.status_code}"
+            )
+
+            error_data = response.json()
+            assert error_data["error"] == "invalid_client", (
+                f"Expected 'invalid_client' but got '{error_data.get('error')}'"
+            )
+            assert "Invalid client_id" in error_data["error_description"]
+
+            # Verify proper cache headers are set
+            assert response.headers.get("Cache-Control") == "no-store"
+            assert response.headers.get("Pragma") == "no-cache"
+
+
+class TestTokenHandlerErrorTransformation:
+    """Tests for TokenHandler's OAuth 2.1 compliant error transformation."""
+
+    def test_transforms_client_auth_failure_to_invalid_client_401(self):
+        """Test that client authentication failures return invalid_client with 401."""
+        from mcp.server.auth.handlers.token import TokenErrorResponse
+
+        from fastmcp.server.auth.oauth_proxy import TokenHandler
+
+        handler = TokenHandler(provider=Mock(), client_authenticator=Mock())
+
+        # Simulate error from ClientAuthenticator.authenticate() failure
+        error_response = TokenErrorResponse(
+            error="unauthorized_client",
+            error_description="Invalid client_id 'test-client-id'",
+        )
+
+        response = handler.response(error_response)
+
+        # Should transform to OAuth 2.1 compliant response
+        assert response.status_code == 401
+        assert b'"error":"invalid_client"' in response.body
+        assert (
+            b'"error_description":"Invalid client_id \'test-client-id\'"'
+            in response.body
+        )
+
+    def test_does_not_transform_grant_type_unauthorized_to_invalid_client(self):
+        """Test that grant type authorization errors stay as unauthorized_client with 400."""
+        from mcp.server.auth.handlers.token import TokenErrorResponse
+
+        from fastmcp.server.auth.oauth_proxy import TokenHandler
+
+        handler = TokenHandler(provider=Mock(), client_authenticator=Mock())
+
+        # Simulate error from grant_type not in client_info.grant_types
+        error_response = TokenErrorResponse(
+            error="unauthorized_client",
+            error_description="Client not authorized for this grant type",
+        )
+
+        response = handler.response(error_response)
+
+        # Should NOT transform - keep as 400 unauthorized_client
+        assert response.status_code == 400
+        assert b'"error":"unauthorized_client"' in response.body
+
+    def test_does_not_transform_other_errors(self):
+        """Test that other error types pass through unchanged."""
+        from mcp.server.auth.handlers.token import TokenErrorResponse
+
+        from fastmcp.server.auth.oauth_proxy import TokenHandler
+
+        handler = TokenHandler(provider=Mock(), client_authenticator=Mock())
+
+        error_response = TokenErrorResponse(
+            error="invalid_grant",
+            error_description="Authorization code has expired",
+        )
+
+        response = handler.response(error_response)
+
+        # Should pass through unchanged
+        assert response.status_code == 400
+        assert b'"error":"invalid_grant"' in response.body

tests/server/auth/test_oauth_proxy_redirect_validation.py

@@ -176,7 +176,7 @@ class TestOAuthProxyRedirectValidation:
             "new-client"
         )  # Use the client ID we registered
         assert isinstance(registered, ProxyDCRClient)
-        assert registered._allowed_redirect_uri_patterns == custom_patterns
+        assert registered.allowed_redirect_uri_patterns == custom_patterns
 
     @pytest.mark.asyncio
     async def test_proxy_unregistered_client_returns_none(self):

tests/server/auth/test_oauth_proxy_storage.py

@@ -0,0 +1,204 @@
+"""Tests for OAuth proxy with persistent storage."""
+
+from collections.abc import AsyncGenerator
+from pathlib import Path
+from unittest.mock import AsyncMock, Mock
+
+import pytest
+from diskcache.core import tempfile
+from inline_snapshot import snapshot
+from key_value.aio.stores.disk import MultiDiskStore
+from key_value.aio.stores.memory import MemoryStore
+from mcp.shared.auth import OAuthClientInformationFull
+from pydantic import AnyUrl
+
+from fastmcp.server.auth.oauth_proxy import OAuthProxy
+
+
+class TestOAuthProxyStorage:
+    """Tests for OAuth proxy client storage functionality."""
+
+    @pytest.fixture
+    def jwt_verifier(self):
+        """Create a mock JWT verifier."""
+        verifier = Mock()
+        verifier.required_scopes = ["read", "write"]
+        verifier.verify_token = AsyncMock(return_value=None)
+        return verifier
+
+    @pytest.fixture
+    async def temp_storage(self) -> AsyncGenerator[MultiDiskStore, None]:
+        """Create file-based storage for testing."""
+        with tempfile.TemporaryDirectory() as temp_dir:
+            disk_store = MultiDiskStore(base_directory=Path(temp_dir))
+            yield disk_store
+            await disk_store.close()
+
+    @pytest.fixture
+    def memory_storage(self) -> MemoryStore:
+        """Create in-memory storage for testing."""
+        return MemoryStore()
+
+    def create_proxy(self, jwt_verifier, storage=None) -> OAuthProxy:
+        """Create an OAuth proxy with specified storage."""
+        return OAuthProxy(
+            upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
+            upstream_token_endpoint="https://github.com/login/oauth/access_token",
+            upstream_client_id="test-client-id",
+            upstream_client_secret="test-client-secret",
+            token_verifier=jwt_verifier,
+            base_url="https://myserver.com",
+            redirect_path="/auth/callback",
+            client_storage=storage,
+        )
+
+    async def test_default_storage_is_file_based(self, jwt_verifier):
+        """Test that proxy defaults to file-based storage."""
+        proxy = self.create_proxy(jwt_verifier, storage=None)
+        assert isinstance(proxy._client_storage, MemoryStore)
+
+    async def test_register_and_get_client(self, jwt_verifier, temp_storage):
+        """Test registering and retrieving a client."""
+        proxy = self.create_proxy(jwt_verifier, storage=temp_storage)
+
+        # Register client
+        client_info = OAuthClientInformationFull(
+            client_id="test-client-123",
+            client_secret="secret-456",
+            redirect_uris=[AnyUrl("http://localhost:8080/callback")],
+            grant_types=["authorization_code", "refresh_token"],
+            scope="read write",
+        )
+        await proxy.register_client(client_info)
+
+        # Get client back
+        client = await proxy.get_client("test-client-123")
+        assert client is not None
+        assert client.client_id == "test-client-123"
+        assert client.client_secret == "secret-456"
+        assert client.scope == "read write"
+
+    async def test_client_persists_across_proxy_instances(
+        self, jwt_verifier, temp_storage
+    ):
+        """Test that clients persist when proxy is recreated."""
+        # First proxy registers client
+        proxy1 = self.create_proxy(jwt_verifier, storage=temp_storage)
+        client_info = OAuthClientInformationFull(
+            client_id="persistent-client",
+            client_secret="persistent-secret",
+            redirect_uris=[AnyUrl("http://localhost:9999/callback")],
+            scope="openid profile",
+        )
+        await proxy1.register_client(client_info)
+
+        # Second proxy can retrieve it
+        proxy2 = self.create_proxy(jwt_verifier, storage=temp_storage)
+        client = await proxy2.get_client("persistent-client")
+        assert client is not None
+        assert client.client_secret == "persistent-secret"
+        assert client.scope == "openid profile"
+
+    async def test_nonexistent_client_returns_none(self, jwt_verifier, temp_storage):
+        """Test that requesting non-existent client returns None."""
+        proxy = self.create_proxy(jwt_verifier, storage=temp_storage)
+        client = await proxy.get_client("does-not-exist")
+        assert client is None
+
+    async def test_proxy_dcr_client_redirect_validation(
+        self, jwt_verifier, temp_storage
+    ):
+        """Test that ProxyDCRClient is created with redirect URI patterns."""
+        proxy = OAuthProxy(
+            upstream_authorization_endpoint="https://github.com/login/oauth/authorize",
+            upstream_token_endpoint="https://github.com/login/oauth/access_token",
+            upstream_client_id="test-client-id",
+            upstream_client_secret="test-client-secret",
+            token_verifier=jwt_verifier,
+            base_url="https://myserver.com",
+            allowed_client_redirect_uris=["http://localhost:*"],
+            client_storage=temp_storage,
+        )
+
+        client_info = OAuthClientInformationFull(
+            client_id="test-proxy-client",
+            client_secret="secret",
+            redirect_uris=[AnyUrl("http://localhost:8080/callback")],
+        )
+        await proxy.register_client(client_info)
+
+        # Get client back - should be ProxyDCRClient
+        client = await proxy.get_client("test-proxy-client")
+        assert client is not None
+
+        # ProxyDCRClient should validate dynamic localhost ports
+        validated = client.validate_redirect_uri(
+            AnyUrl("http://localhost:12345/callback")
+        )
+        assert validated is not None
+
+    async def test_in_memory_storage_option(self, jwt_verifier):
+        """Test using in-memory storage explicitly."""
+        storage = MemoryStore()
+        proxy = self.create_proxy(jwt_verifier, storage=storage)
+
+        client_info = OAuthClientInformationFull(
+            client_id="memory-client",
+            client_secret="memory-secret",
+            redirect_uris=[AnyUrl("http://localhost:8080/callback")],
+        )
+        await proxy.register_client(client_info)
+
+        client = await proxy.get_client("memory-client")
+        assert client is not None
+
+        # Create new proxy with same storage instance
+        proxy2 = self.create_proxy(jwt_verifier, storage=storage)
+        client2 = await proxy2.get_client("memory-client")
+        assert client2 is not None
+
+        # But new storage instance won't have it
+        proxy3 = self.create_proxy(jwt_verifier, storage=MemoryStore())
+        client3 = await proxy3.get_client("memory-client")
+        assert client3 is None
+
+    async def test_storage_data_structure(self, jwt_verifier, temp_storage):
+        """Test that storage uses proper structured format."""
+        proxy = self.create_proxy(jwt_verifier, storage=temp_storage)
+
+        client_info = OAuthClientInformationFull(
+            client_id="structured-client",
+            client_secret="secret",
+            redirect_uris=[AnyUrl("http://localhost:8080/callback")],
+        )
+        await proxy.register_client(client_info)
+
+        # Check raw storage data
+        raw_data = await temp_storage.get(
+            collection="mcp-oauth-proxy-clients", key="structured-client"
+        )
+        assert raw_data is not None
+        assert raw_data == snapshot(
+            {
+                "redirect_uris": ["http://localhost:8080/callback"],
+                "token_endpoint_auth_method": "none",
+                "grant_types": ["authorization_code", "refresh_token"],
+                "response_types": ["code"],
+                "scope": "read write",
+                "client_name": None,
+                "client_uri": None,
+                "logo_uri": None,
+                "contacts": None,
+                "tos_uri": None,
+                "policy_uri": None,
+                "jwks_uri": None,
+                "jwks": None,
+                "software_id": None,
+                "software_version": None,
+                "client_id": "structured-client",
+                "client_secret": "secret",
+                "client_id_issued_at": None,
+                "client_secret_expires_at": None,
+                "allowed_redirect_uri_patterns": None,
+            }
+        )

tests/server/auth/test_oidc_proxy.py

@@ -1,5 +1,6 @@
 """Comprehensive tests for OIDC Proxy Provider functionality."""
 
+import json
 from unittest.mock import MagicMock, patch
 
 import pytest
@@ -9,6 +10,7 @@ from pydantic import AnyHttpUrl
 from fastmcp.server.auth.oidc_proxy import OIDCConfiguration, OIDCProxy
 from fastmcp.server.auth.providers.jwt import JWTVerifier
 
+TEST_ISSUER = "https://example.com"
 TEST_AUTHORIZATION_ENDPOINT = "https://example.com/authorize"
 TEST_TOKEN_ENDPOINT = "https://example.com/oauth/token"
 
@@ -27,7 +29,7 @@ TEST_BASE_URL = "https://example.com:8000/"
 def valid_oidc_configuration_dict():
     """Create a valid OIDC configuration dict for testing."""
     return {
-        "issuer": "https://example.com/",
+        "issuer": TEST_ISSUER,
         "authorization_endpoint": TEST_AUTHORIZATION_ENDPOINT,
         "token_endpoint": TEST_TOKEN_ENDPOINT,
         "jwks_uri": "https://example.com/.well-known/jwks.json",
@@ -41,27 +43,218 @@ def valid_oidc_configuration_dict():
 def invalid_oidc_configuration_dict():
     """Create an invalid OIDC configuration dict for testing."""
     return {
-        "issuer": "https://example.com/",
+        "issuer": TEST_ISSUER,
         "authorization_endpoint": TEST_AUTHORIZATION_ENDPOINT,
         "token_endpoint": TEST_TOKEN_ENDPOINT,
         "jwks_uri": "https://example.com/.well-known/jwks.json",
     }
 
 
+@pytest.fixture
+def valid_google_oidc_configuration_dict():
+    """Create a valid Google OIDC configuration dict for testing.
+
+    See: https://accounts.google.com/.well-known/openid-configuration
+    """
+    google_config_str = """
+    {
+      "issuer": "https://accounts.google.com",
+      "authorization_endpoint": "https://accounts.google.com/o/oauth2/v2/auth",
+      "device_authorization_endpoint": "https://oauth2.googleapis.com/device/code",
+      "token_endpoint": "https://oauth2.googleapis.com/token",
+      "userinfo_endpoint": "https://openidconnect.googleapis.com/v1/userinfo",
+      "revocation_endpoint": "https://oauth2.googleapis.com/revoke",
+      "jwks_uri": "https://www.googleapis.com/oauth2/v3/certs",
+      "response_types_supported": [
+        "code",
+        "token",
+        "id_token",
+        "code token",
+        "code id_token",
+        "token id_token",
+        "code token id_token",
+        "none"
+      ],
+      "response_modes_supported": [
+        "query",
+        "fragment",
+        "form_post"
+      ],
+      "subject_types_supported": [
+        "public"
+      ],
+      "id_token_signing_alg_values_supported": [
+        "RS256"
+      ],
+      "scopes_supported": [
+        "openid",
+        "email",
+        "profile"
+      ],
+      "token_endpoint_auth_methods_supported": [
+        "client_secret_post",
+        "client_secret_basic"
+      ],
+      "claims_supported": [
+        "aud",
+        "email",
+        "email_verified",
+        "exp",
+        "family_name",
+        "given_name",
+        "iat",
+        "iss",
+        "name",
+        "picture",
+        "sub"
+      ],
+      "code_challenge_methods_supported": [
+        "plain",
+        "S256"
+      ],
+      "grant_types_supported": [
+        "authorization_code",
+        "refresh_token",
+        "urn:ietf:params:oauth:grant-type:device_code",
+        "urn:ietf:params:oauth:grant-type:jwt-bearer"
+      ]
+    }
+    """
+
+    return json.loads(google_config_str)
+
+
+@pytest.fixture
+def valid_auth0_oidc_configuration_dict():
+    """Create a valid Auth0 OIDC configuration dict for testing.
+
+    See: https://<tenant>.us.auth0.com/.well-known/openid-configuration
+    """
+    auth0_config_str = """
+    {
+      "issuer": "https://example.us.auth0.com/",
+      "authorization_endpoint": "https://example.us.auth0.com/authorize",
+      "token_endpoint": "https://example.us.auth0.com/oauth/token",
+      "device_authorization_endpoint": "https://example.us.auth0.com/oauth/device/code",
+      "userinfo_endpoint": "https://example.us.auth0.com/userinfo",
+      "mfa_challenge_endpoint": "https://example.us.auth0.com/mfa/challenge",
+      "jwks_uri": "https://example.us.auth0.com/.well-known/jwks.json",
+      "registration_endpoint": "https://example.us.auth0.com/oidc/register",
+      "revocation_endpoint": "https://example.us.auth0.com/oauth/revoke",
+      "scopes_supported": [
+        "openid",
+        "profile",
+        "offline_access",
+        "name",
+        "given_name",
+        "family_name",
+        "nickname",
+        "email",
+        "email_verified",
+        "picture",
+        "created_at",
+        "identities",
+        "phone",
+        "address"
+      ],
+      "response_types_supported": [
+        "code",
+        "token",
+        "id_token",
+        "code token",
+        "code id_token",
+        "token id_token",
+        "code token id_token"
+      ],
+      "code_challenge_methods_supported": [
+        "S256",
+        "plain"
+      ],
+      "response_modes_supported": [
+        "query",
+        "fragment",
+        "form_post"
+      ],
+      "subject_types_supported": [
+        "public"
+      ],
+      "token_endpoint_auth_methods_supported": [
+        "client_secret_basic",
+        "client_secret_post",
+        "private_key_jwt",
+        "tls_client_auth",
+        "self_signed_tls_client_auth"
+      ],
+      "token_endpoint_auth_signing_alg_values_supported": [
+        "RS256",
+        "RS384",
+        "PS256"
+      ],
+      "claims_supported": [
+        "aud",
+        "auth_time",
+        "created_at",
+        "email",
+        "email_verified",
+        "exp",
+        "family_name",
+        "given_name",
+        "iat",
+        "identities",
+        "iss",
+        "name",
+        "nickname",
+        "phone_number",
+        "picture",
+        "sub"
+      ],
+      "request_uri_parameter_supported": false,
+      "request_parameter_supported": true,
+      "id_token_signing_alg_values_supported": [
+        "HS256",
+        "RS256",
+        "PS256"
+      ],
+      "tls_client_certificate_bound_access_tokens": true,
+      "request_object_signing_alg_values_supported": [
+        "RS256",
+        "RS384",
+        "PS256"
+      ],
+      "backchannel_logout_supported": true,
+      "backchannel_logout_session_supported": true,
+      "end_session_endpoint": "https://example.us.auth0.com/oidc/logout",
+      "backchannel_authentication_endpoint": "https://example.us.auth0.com/bc-authorize",
+      "backchannel_token_delivery_modes_supported": [
+        "poll"
+      ],
+      "global_token_revocation_endpoint": "https://example.us.auth0.com/oauth/global-token-revocation/connection/{connectionName}",
+      "global_token_revocation_endpoint_auth_methods_supported": [
+        "global-token-revocation+jwt"
+      ]
+    }
+    """
+
+    return json.loads(auth0_config_str)
+
+
 # =============================================================================
 # Test Classes
 # =============================================================================
 
 
-def validate_config(config):
-    """Validate an OIDC configuration."""
-    assert str(config.issuer) == "https://example.com/"
-    assert str(config.authorization_endpoint) == TEST_AUTHORIZATION_ENDPOINT
-    assert str(config.token_endpoint) == TEST_TOKEN_ENDPOINT
-    assert str(config.jwks_uri) == "https://example.com/.well-known/jwks.json"
-    assert config.response_types_supported == ["code"]
-    assert config.subject_types_supported == ["public"]
-    assert config.id_token_signing_alg_values_supported == ["RS256"]
+def validate_config(config, source_dict):
+    """Validate an OIDC configuration against the source dict."""
+    for source_key, source_value in source_dict.items():
+        config_value = getattr(config, source_key, None)
+        if not hasattr(config, source_key):
+            continue
+
+        config_value = getattr(config, source_key, None)
+        if isinstance(config_value, AnyHttpUrl):
+            config_value = str(config_value)
+
+        assert config_value == source_value
 
 
 class TestOIDCConfiguration:
@@ -70,13 +263,29 @@ class TestOIDCConfiguration:
     def test_default_configuration(self, valid_oidc_configuration_dict):
         """Test default configuration with valid dict."""
         config = OIDCConfiguration.model_validate(valid_oidc_configuration_dict)
-        validate_config(config)
+        validate_config(config, valid_oidc_configuration_dict)
+
+    def test_default_configuration_with_issuer_trailing_slash(
+        self, valid_oidc_configuration_dict
+    ):
+        """Test default configuration with valid dict and issuer trailing slash."""
+        valid_oidc_configuration_dict["issuer"] += "/"
+        config = OIDCConfiguration.model_validate(valid_oidc_configuration_dict)
+        validate_config(config, valid_oidc_configuration_dict)
 
     def test_explicit_strict_configuration(self, valid_oidc_configuration_dict):
         """Test default configuration with explicit True strict setting and valid dict."""
         valid_oidc_configuration_dict["strict"] = True
         config = OIDCConfiguration.model_validate(valid_oidc_configuration_dict)
-        validate_config(config)
+        validate_config(config, valid_oidc_configuration_dict)
+
+    def test_explicit_strict_configuration_with_issuer_trailing_slash(
+        self, valid_oidc_configuration_dict
+    ):
+        """Test default configuration with explicit True strict setting, valid dict and issuer trailing slash."""
+        valid_oidc_configuration_dict["issuer"] += "/"
+        config = OIDCConfiguration.model_validate(valid_oidc_configuration_dict)
+        validate_config(config, valid_oidc_configuration_dict)
 
     def test_default_configuration_raises_error(self, invalid_oidc_configuration_dict):
         """Test default configuration with invalid dict."""
@@ -91,6 +300,21 @@ class TestOIDCConfiguration:
         with pytest.raises(ValueError, match="Missing required configuration metadata"):
             OIDCConfiguration.model_validate(invalid_oidc_configuration_dict)
 
+    def test_bad_url_raises_error(self, valid_oidc_configuration_dict):
+        """Test default configuration with bad URL setting."""
+        valid_oidc_configuration_dict["issuer"] = "not-a-URL"
+        with pytest.raises(ValueError, match="Invalid URL for configuration metadata"):
+            OIDCConfiguration.model_validate(valid_oidc_configuration_dict)
+
+    def test_explict_strict_with_bad_url_raises_error(
+        self, valid_oidc_configuration_dict
+    ):
+        """Test default configuration with explicit True strict setting and bad URL setting."""
+        valid_oidc_configuration_dict["strict"] = True
+        valid_oidc_configuration_dict["issuer"] = "not-a-URL"
+        with pytest.raises(ValueError, match="Invalid URL for configuration metadata"):
+            OIDCConfiguration.model_validate(valid_oidc_configuration_dict)
+
     def test_not_strict_configuration(self):
         """Test default configuration with explicit False strict setting."""
         config = OIDCConfiguration.model_validate({"strict": False})
@@ -103,6 +327,35 @@ class TestOIDCConfiguration:
         assert config.subject_types_supported is None
         assert config.id_token_signing_alg_values_supported is None
 
+    def test_not_strict_configuration_with_invalid_config(
+        self, invalid_oidc_configuration_dict
+    ):
+        """Test default configuration with explicit False strict setting."""
+        invalid_oidc_configuration_dict["strict"] = False
+        config = OIDCConfiguration.model_validate(invalid_oidc_configuration_dict)
+
+        validate_config(config, invalid_oidc_configuration_dict)
+
+    def test_not_strict_configuration_with_bad_url(self, valid_oidc_configuration_dict):
+        """Test default configuration with explicit False strict setting."""
+        valid_oidc_configuration_dict["strict"] = False
+        valid_oidc_configuration_dict["issuer"] = "not-a-url"
+        config = OIDCConfiguration.model_validate(valid_oidc_configuration_dict)
+
+        validate_config(config, valid_oidc_configuration_dict)
+
+    def test_google_configuration(self, valid_google_oidc_configuration_dict):
+        """Test Google configuration."""
+        config = OIDCConfiguration.model_validate(valid_google_oidc_configuration_dict)
+
+        validate_config(config, valid_google_oidc_configuration_dict)
+
+    def test_auth0_configuration(self, valid_auth0_oidc_configuration_dict):
+        """Test Auth0 configuration."""
+        config = OIDCConfiguration.model_validate(valid_auth0_oidc_configuration_dict)
+
+        validate_config(config, valid_auth0_oidc_configuration_dict)
+
 
 def validate_get_oidc_configuration(oidc_configuration, strict, timeout_seconds):
     """Validate get_oidc_configuation call."""
@@ -117,7 +370,7 @@ def validate_get_oidc_configuration(oidc_configuration, strict, timeout_seconds)
             timeout_seconds=timeout_seconds,
         )
 
-        validate_config(config)
+        validate_config(config, oidc_configuration)
 
         mock_get.assert_called_once()
 

tests/server/auth/test_remote_auth_provider.py

@@ -105,6 +105,28 @@ class TestRemoteAuthProvider:
             "https://api.example.com/.well-known/oauth-protected-resource"
         )
 
+    def test_get_resource_url_with_nested_base_url(self):
+        """Test _get_resource_url returns correct URL for .well-known path with nested base_url."""
+        tokens = {
+            "test_token": {
+                "client_id": "test-client",
+                "scopes": ["read"],
+            }
+        }
+        token_verifier = StaticTokenVerifier(tokens=tokens)
+        provider = RemoteAuthProvider(
+            token_verifier=token_verifier,
+            authorization_servers=[AnyHttpUrl("https://auth.example.com")],
+            base_url="https://api.example.com/v1/",
+        )
+
+        metadata_url = provider._get_resource_url(
+            "/.well-known/oauth-protected-resource"
+        )
+        assert metadata_url == AnyHttpUrl(
+            "https://api.example.com/v1/.well-known/oauth-protected-resource"
+        )
+
     def test_get_resource_url_handles_trailing_slash(self):
         """Test _get_resource_url handles trailing slash correctly."""
         tokens = {
@@ -216,6 +238,7 @@ class TestRemoteAuthProviderIntegration:
         [
             ("https://api.example.com", "https://api.example.com/mcp"),
             ("https://api.example.com/", "https://api.example.com/mcp"),
+            ("https://api.example.com/v1/", "https://api.example.com/v1/mcp"),
         ],
     )
     async def test_base_url_configurations(self, base_url: str, expected_resource: str):

tests/server/http/test_custom_routes.py

@@ -19,7 +19,7 @@ class TestCustomRoutes:
 
         return server
 
-    def test_custom_routes_via_server_http_app(self, server_with_custom_route):
+    def test_custom_routes_apply_filtering_http_app(self, server_with_custom_route):
         """Test that custom routes are included when using server.http_app()."""
         # Get the app via server.http_app()
         app = server_with_custom_route.http_app()

tests/server/http/test_http_dependencies.py

@@ -42,13 +42,13 @@ def run_server(host: str, port: int, **kwargs) -> None:
     fastmcp_server().run(host=host, port=port, **kwargs)
 
 
-@pytest.fixture(autouse=True, scope="module")
+@pytest.fixture(autouse=True)
 def shttp_server() -> Generator[str, None, None]:
     with run_server_in_process(run_server, transport="http") as url:
         yield f"{url}/mcp"
 
 
-@pytest.fixture(autouse=True, scope="module")
+@pytest.fixture(autouse=True)
 def sse_server() -> Generator[str, None, None]:
     with run_server_in_process(run_server, transport="sse") as url:
         yield f"{url}/sse"

tests/server/middleware/test_error_handling.py

@@ -11,6 +11,7 @@ from fastmcp.server.middleware.error_handling import (
     RetryMiddleware,
 )
 from fastmcp.server.middleware.middleware import MiddlewareContext
+from fastmcp.utilities.tests import caplog_for_fastmcp
 
 
 @pytest.fixture
@@ -60,8 +61,9 @@ class TestErrorHandlingMiddleware:
         middleware = ErrorHandlingMiddleware()
         error = ValueError("test error")
 
-        with caplog.at_level(logging.ERROR):
-            middleware._log_error(error, mock_context)
+        with caplog_for_fastmcp(caplog):
+            with caplog.at_level(logging.ERROR):
+                middleware._log_error(error, mock_context)
 
         assert "Error in test_method: ValueError: test error" in caplog.text
         assert "ValueError:test_method" in middleware.error_counts
@@ -72,8 +74,9 @@ class TestErrorHandlingMiddleware:
         middleware = ErrorHandlingMiddleware(include_traceback=True)
         error = ValueError("test error")
 
-        with caplog.at_level(logging.ERROR):
-            middleware._log_error(error, mock_context)
+        with caplog_for_fastmcp(caplog):
+            with caplog.at_level(logging.ERROR):
+                middleware._log_error(error, mock_context)
 
         assert "Error in test_method: ValueError: test error" in caplog.text
         # The traceback is added to the log message
@@ -95,8 +98,9 @@ class TestErrorHandlingMiddleware:
         middleware = ErrorHandlingMiddleware(error_callback=callback)
         error = ValueError("test error")
 
-        with caplog.at_level(logging.ERROR):
-            middleware._log_error(error, mock_context)
+        with caplog_for_fastmcp(caplog):
+            with caplog.at_level(logging.ERROR):
+                middleware._log_error(error, mock_context)
 
         assert "Error in error callback: callback error" in caplog.text
 
@@ -189,9 +193,10 @@ class TestErrorHandlingMiddleware:
         middleware = ErrorHandlingMiddleware()
         mock_call_next = AsyncMock(side_effect=ValueError("test error"))
 
-        with caplog.at_level(logging.ERROR):
-            with pytest.raises(McpError) as exc_info:
-                await middleware.on_message(mock_context, mock_call_next)
+        with caplog_for_fastmcp(caplog):
+            with caplog.at_level(logging.ERROR):
+                with pytest.raises(McpError) as exc_info:
+                    await middleware.on_message(mock_context, mock_call_next)
 
         assert isinstance(exc_info.value, McpError)
         assert exc_info.value.error.code == -32602
@@ -293,8 +298,9 @@ class TestRetryMiddleware:
             ]
         )
 
-        with caplog.at_level(logging.WARNING):
-            result = await middleware.on_request(mock_context, mock_call_next)
+        with caplog_for_fastmcp(caplog):
+            with caplog.at_level(logging.WARNING):
+                result = await middleware.on_request(mock_context, mock_call_next)
 
         assert result == "test_result"
         assert mock_call_next.call_count == 3
@@ -307,9 +313,10 @@ class TestRetryMiddleware:
         # Fail all attempts
         mock_call_next = AsyncMock(side_effect=ConnectionError("connection failed"))
 
-        with caplog.at_level(logging.WARNING):
-            with pytest.raises(ConnectionError):
-                await middleware.on_request(mock_context, mock_call_next)
+        with caplog_for_fastmcp(caplog):
+            with caplog.at_level(logging.WARNING):
+                with pytest.raises(ConnectionError):
+                    await middleware.on_request(mock_context, mock_call_next)
 
         assert mock_call_next.call_count == 3  # initial + 2 retries
         assert "Retrying in" in caplog.text
@@ -385,14 +392,19 @@ class TestErrorHandlingMiddlewareIntegration:
 
         error_handling_server.add_middleware(ErrorHandlingMiddleware())
 
-        with caplog.at_level(logging.ERROR):
-            async with Client(error_handling_server) as client:
-                # Test different types of errors
-                with pytest.raises(Exception):
-                    await client.call_tool("failing_operation", {"error_type": "value"})
+        with caplog_for_fastmcp(caplog):
+            with caplog.at_level(logging.ERROR):
+                async with Client(error_handling_server) as client:
+                    # Test different types of errors
+                    with pytest.raises(Exception):
+                        await client.call_tool(
+                            "failing_operation", {"error_type": "value"}
+                        )
 
-                with pytest.raises(Exception):
-                    await client.call_tool("failing_operation", {"error_type": "file"})
+                    with pytest.raises(Exception):
+                        await client.call_tool(
+                            "failing_operation", {"error_type": "file"}
+                        )
 
         log_text = caplog.text
 
@@ -443,17 +455,20 @@ class TestErrorHandlingMiddlewareIntegration:
 
         error_handling_server.add_middleware(ErrorHandlingMiddleware())
 
-        with caplog.at_level(logging.ERROR):
-            async with Client(error_handling_server) as client:
-                # Successful operation (should not generate error logs)
-                await client.call_tool("reliable_operation", {"data": "test"})
+        with caplog_for_fastmcp(caplog):
+            with caplog.at_level(logging.ERROR):
+                async with Client(error_handling_server) as client:
+                    # Successful operation (should not generate error logs)
+                    await client.call_tool("reliable_operation", {"data": "test"})
 
-                # Failed operation (should generate error log)
-                with pytest.raises(Exception):
-                    await client.call_tool("failing_operation", {"error_type": "value"})
+                    # Failed operation (should generate error log)
+                    with pytest.raises(Exception):
+                        await client.call_tool(
+                            "failing_operation", {"error_type": "value"}
+                        )
 
-                # Another successful operation
-                await client.call_tool("reliable_operation", {"data": "test2"})
+                    # Another successful operation
+                    await client.call_tool("reliable_operation", {"data": "test2"})
 
         log_text = caplog.text
 
@@ -533,18 +548,19 @@ class TestRetryMiddlewareIntegration:
             )
         )
 
-        with caplog.at_level(logging.WARNING):
-            async with Client(error_handling_server) as client:
-                # This operation fails intermittently - try several times
-                success_count = 0
-                for _ in range(5):
-                    try:
-                        await client.call_tool(
-                            "intermittent_operation", {"fail_rate": 0.7}
-                        )
-                        success_count += 1
-                    except Exception:
-                        pass  # Some failures expected even with retries
+        with caplog_for_fastmcp(caplog):
+            with caplog.at_level(logging.WARNING):
+                async with Client(error_handling_server) as client:
+                    # This operation fails intermittently - try several times
+                    success_count = 0
+                    for _ in range(5):
+                        try:
+                            await client.call_tool(
+                                "intermittent_operation", {"fail_rate": 0.7}
+                            )
+                            success_count += 1
+                        except Exception:
+                            pass  # Some failures expected even with retries
 
         # Should have some retry log messages
         # Note: Retry logs might not appear if the underlying errors are wrapped by FastMCP
@@ -584,17 +600,22 @@ class TestRetryMiddlewareIntegration:
             )
         )
 
-        with caplog.at_level(logging.ERROR):
-            async with Client(error_handling_server) as client:
-                # Try intermittent operation
-                try:
-                    await client.call_tool("intermittent_operation", {"fail_rate": 0.9})
-                except Exception:
-                    pass  # May still fail even with retries
+        with caplog_for_fastmcp(caplog):
+            with caplog.at_level(logging.ERROR):
+                async with Client(error_handling_server) as client:
+                    # Try intermittent operation
+                    try:
+                        await client.call_tool(
+                            "intermittent_operation", {"fail_rate": 0.9}
+                        )
+                    except Exception:
+                        pass  # May still fail even with retries
 
-                # Try permanent failure
-                with pytest.raises(Exception):
-                    await client.call_tool("failing_operation", {"error_type": "value"})
+                    # Try permanent failure
+                    with pytest.raises(Exception):
+                        await client.call_tool(
+                            "failing_operation", {"error_type": "value"}
+                        )
 
         log_text = caplog.text
 

tests/server/middleware/test_initialization_middleware.py

@@ -0,0 +1,251 @@
+"""Tests for middleware support during initialization."""
+
+from typing import Any
+
+import mcp.types as mt
+
+from fastmcp import Client, FastMCP
+from fastmcp.server.middleware import CallNext, Middleware, MiddlewareContext
+
+
+class InitializationMiddleware(Middleware):
+    """Middleware that captures initialization details."""
+
+    def __init__(self):
+        super().__init__()
+        self.initialized = False
+        self.client_info = None
+        self.session_data = {}
+
+    async def on_initialize(
+        self,
+        context: MiddlewareContext[mt.InitializeRequest],
+        call_next: CallNext[mt.InitializeRequest, None],
+    ) -> None:
+        """Capture initialization details and store session data."""
+        self.initialized = True
+
+        # Extract client info from the initialize params
+        if hasattr(context.message, "params") and hasattr(
+            context.message.params, "clientInfo"
+        ):
+            self.client_info = context.message.params.clientInfo
+
+        # Store data in the context state for cross-request access
+        if context.fastmcp_context:
+            context.fastmcp_context.set_state("client_initialized", True)
+            if self.client_info:
+                context.fastmcp_context.set_state(
+                    "client_name", getattr(self.client_info, "name", "unknown")
+                )
+
+        return await call_next(context)
+
+
+class ClientDetectionMiddleware(Middleware):
+    """Middleware that detects specific clients and modifies behavior.
+
+    This demonstrates storing data in the middleware instance itself
+    for cross-request access, since context state is request-scoped.
+    """
+
+    def __init__(self):
+        super().__init__()
+        self.is_test_client = False
+        self.tools_modified = False
+        self.initialization_called = False
+
+    async def on_initialize(
+        self,
+        context: MiddlewareContext[mt.InitializeRequest],
+        call_next: CallNext[mt.InitializeRequest, None],
+    ) -> None:
+        """Detect test client during initialization."""
+        self.initialization_called = True
+
+        # For testing purposes, always set it to true
+        # Store in instance variable for cross-request access
+        self.is_test_client = True
+
+        return await call_next(context)
+
+    async def on_list_tools(
+        self,
+        context: MiddlewareContext[mt.ListToolsRequest],
+        call_next: CallNext[mt.ListToolsRequest, list],
+    ) -> list:
+        """Modify tools based on client detection."""
+        tools = await call_next(context)
+
+        # Use the instance variable set during initialization
+        if self.is_test_client:
+            # Add a special annotation to tools for test clients
+            for tool in tools:
+                if not hasattr(tool, "annotations"):
+                    tool.annotations = mt.ToolAnnotations()
+                if tool.annotations is None:
+                    tool.annotations = mt.ToolAnnotations()
+                # Mark as read-only for test clients
+                tool.annotations.readOnlyHint = True
+            self.tools_modified = True
+
+        return tools
+
+
+async def test_simple_initialization_hook():
+    """Test that the on_initialize hook is called."""
+    server = FastMCP("TestServer")
+
+    class SimpleInitMiddleware(Middleware):
+        def __init__(self):
+            super().__init__()
+            self.called = False
+
+        async def on_initialize(
+            self,
+            context: MiddlewareContext[mt.InitializeRequest],
+            call_next: CallNext[mt.InitializeRequest, None],
+        ) -> None:
+            self.called = True
+            return await call_next(context)
+
+    middleware = SimpleInitMiddleware()
+    server.add_middleware(middleware)
+
+    # Connect client
+    async with Client(server):
+        # Middleware should have been called
+        assert middleware.called is True, "on_initialize was not called"
+
+
+async def test_middleware_receives_initialization():
+    """Test that middleware can intercept initialization requests."""
+    server = FastMCP("TestServer")
+    middleware = InitializationMiddleware()
+    server.add_middleware(middleware)
+
+    @server.tool
+    def test_tool(x: int) -> str:
+        return f"Result: {x}"
+
+    # Connect client
+    async with Client(server) as client:
+        # Middleware should have been called during initialization
+        assert middleware.initialized is True
+
+        # Test that the tool still works
+        result = await client.call_tool("test_tool", {"x": 42})
+        assert result.content[0].text == "Result: 42"  # type: ignore[attr-defined]
+
+
+async def test_client_detection_middleware():
+    """Test middleware that detects specific clients and modifies behavior."""
+    server = FastMCP("TestServer")
+    middleware = ClientDetectionMiddleware()
+    server.add_middleware(middleware)
+
+    @server.tool
+    def example_tool() -> str:
+        return "example"
+
+    # Connect with a client
+    async with Client(server) as client:
+        # Middleware should have been called during initialization
+        assert middleware.initialization_called is True
+        assert middleware.is_test_client is True
+
+        # List tools to trigger modification
+        tools = await client.list_tools()
+        assert len(tools) == 1
+        assert middleware.tools_modified is True
+
+        # Check that the tool has the modified annotation
+        tool = tools[0]
+        assert tool.annotations is not None
+        assert tool.annotations.readOnlyHint is True
+
+
+async def test_multiple_middleware_initialization():
+    """Test that multiple middleware can handle initialization."""
+    server = FastMCP("TestServer")
+
+    init_mw = InitializationMiddleware()
+    detect_mw = ClientDetectionMiddleware()
+
+    server.add_middleware(init_mw)
+    server.add_middleware(detect_mw)
+
+    @server.tool
+    def test_tool() -> str:
+        return "test"
+
+    async with Client(server) as client:
+        # Both middleware should have processed initialization
+        assert init_mw.initialized is True
+        assert detect_mw.initialization_called is True
+        assert detect_mw.is_test_client is True
+
+        # List tools to check detection worked
+        await client.list_tools()
+        assert detect_mw.tools_modified is True
+
+
+async def test_initialization_middleware_with_state_sharing():
+    """Test that state set during initialization is available in later requests."""
+    server = FastMCP("TestServer")
+
+    class StateTrackingMiddleware(Middleware):
+        def __init__(self):
+            super().__init__()
+            self.init_state = {}
+            self.tool_state = {}
+
+        async def on_initialize(
+            self,
+            context: MiddlewareContext[mt.InitializeRequest],
+            call_next: CallNext[mt.InitializeRequest, None],
+        ) -> None:
+            # Store some state during initialization
+            if context.fastmcp_context:
+                context.fastmcp_context.set_state("init_timestamp", "2024-01-01")
+                context.fastmcp_context.set_state("client_id", "test-123")
+                self.init_state["timestamp"] = "2024-01-01"
+                self.init_state["client_id"] = "test-123"
+
+            return await call_next(context)
+
+        async def on_call_tool(
+            self,
+            context: MiddlewareContext[mt.CallToolRequestParams],
+            call_next: CallNext[mt.CallToolRequestParams, Any],
+        ) -> Any:
+            # Try to access state from initialization
+            if context.fastmcp_context:
+                timestamp = context.fastmcp_context.get_state("init_timestamp")
+                client_id = context.fastmcp_context.get_state("client_id")
+                self.tool_state["timestamp"] = timestamp
+                self.tool_state["client_id"] = client_id
+
+            return await call_next(context)
+
+    middleware = StateTrackingMiddleware()
+    server.add_middleware(middleware)
+
+    @server.tool
+    def test_tool() -> str:
+        return "success"
+
+    async with Client(server) as client:
+        # Initialization should have set state
+        assert middleware.init_state["timestamp"] == "2024-01-01"
+        assert middleware.init_state["client_id"] == "test-123"
+
+        # Call a tool - state should be accessible
+        result = await client.call_tool("test_tool", {})
+        assert result.content[0].text == "success"  # type: ignore[attr-defined]
+
+        # State should have been accessible during tool call
+        # Note: State is request-scoped, so it won't persist across requests
+        # This test shows the pattern, but actual cross-request state would need
+        # external storage (Redis, DB, etc.)
+        # The middleware.tool_state might be None if state doesn't persist

tests/server/middleware/test_logging.py

@@ -2,37 +2,40 @@
 
 import datetime
 import logging
-import re
+from collections.abc import Generator
 from typing import Any, Literal, TypeVar
-from unittest.mock import AsyncMock, MagicMock
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import mcp
 import mcp.types
 import pytest
 from inline_snapshot import snapshot
+from pydantic import AnyUrl
 
 from fastmcp import FastMCP
 from fastmcp.client import Client
+from fastmcp.resources.template import ResourceTemplate
 from fastmcp.server.middleware.logging import (
     LoggingMiddleware,
     StructuredLoggingMiddleware,
 )
-from fastmcp.server.middleware.middleware import MiddlewareContext
+from fastmcp.server.middleware.middleware import CallNext, MiddlewareContext
+from fastmcp.utilities.tests import caplog_for_fastmcp
 
 FIXED_DATE = datetime.datetime(2023, 1, 1, tzinfo=datetime.timezone.utc)
 
 T = TypeVar("T")
 
 
-def remove_line_numbers(logs: str) -> str:
-    """Remove line numbers from log messages."""
-    trimmed_logs = ""
-    lines = logs.split("\n")
-    for line in lines:
-        # Match only the first `:\d+ `
-        line = re.sub(pattern=r":\d+ ", repl=":LINE_NUMBER ", string=line, count=1)
-        trimmed_logs += line + "\n"
-    return trimmed_logs
+def get_log_lines(
+    caplog: pytest.LogCaptureFixture, module: str | None = None
+) -> list[str]:
+    """Get log lines from a caplog fixture."""
+    return [
+        record.message
+        for record in caplog.records
+        if (module or "logging") in record.name
+    ]
 
 
 def new_mock_context(
@@ -51,6 +54,17 @@ def new_mock_context(
     return context
 
 
+@pytest.fixture(autouse=True)
+def mock_duration_ms() -> Generator[float, None]:
+    """Mock duration_ms."""
+    patched = patch(
+        "fastmcp.server.middleware.logging._get_duration_ms", return_value=0.02
+    )
+    patched.start()
+    yield
+    patched.stop()
+
+
 @pytest.fixture
 def mock_context():
     """Create a mock middleware context."""
@@ -77,15 +91,14 @@ class TestStructuredLoggingMiddleware:
 
     def test_init_default(self):
         """Test default initialization."""
-        middleware = LoggingMiddleware()
+        middleware = StructuredLoggingMiddleware()
 
-        assert middleware.logger.name == "fastmcp.requests"
+        assert middleware.logger.name == "fastmcp.middleware.structured_logging"
         assert middleware.log_level == logging.INFO
         assert middleware.include_payloads is False
-        assert middleware.max_payload_length == 1000
         assert middleware.include_payload_length is False
         assert middleware.estimate_payload_tokens is False
-        assert middleware.structured_logging is False
+        assert middleware.structured_logging is True
 
     def test_init_custom(self):
         """Test custom initialization."""
@@ -108,14 +121,12 @@ class TestStructuredLoggingMiddleware:
             """Test message formatting without payloads."""
             middleware = StructuredLoggingMiddleware()
 
-            message = middleware._create_before_message(mock_context, "test_event")
+            message = middleware._create_before_message(mock_context)
 
             assert message == snapshot(
                 {
-                    "event": "test_event",
-                    "timestamp": "2023-01-01T00:00:00+00:00",
+                    "event": "request_start",
                     "source": "client",
-                    "type": "request",
                     "method": "test_method",
                 }
             )
@@ -126,14 +137,12 @@ class TestStructuredLoggingMiddleware:
             """Test message formatting with payloads."""
             middleware = StructuredLoggingMiddleware(include_payloads=True)
 
-            message = middleware._create_before_message(mock_context, "test_event")
+            message = middleware._create_before_message(mock_context)
 
             assert message == snapshot(
                 {
-                    "event": "test_event",
-                    "timestamp": "2023-01-01T00:00:00+00:00",
+                    "event": "request_start",
                     "source": "client",
-                    "type": "request",
                     "method": "test_method",
                     "payload": '{"method":"tools/call","params":{"_meta":null,"name":"test_method","arguments":{"param":"value"}}}',
                     "payload_type": "CallToolRequest",
@@ -143,14 +152,12 @@ class TestStructuredLoggingMiddleware:
         def test_calculate_response_size(self, mock_context: MiddlewareContext[Any]):
             """Test response size calculation."""
             middleware = StructuredLoggingMiddleware(include_payload_length=True)
-            message = middleware._create_before_message(mock_context, "test_event")
+            message = middleware._create_before_message(mock_context)
 
             assert message == snapshot(
                 {
-                    "event": "test_event",
-                    "timestamp": "2023-01-01T00:00:00+00:00",
+                    "event": "request_start",
                     "source": "client",
-                    "type": "request",
                     "method": "test_method",
                     "payload_length": 98,
                 }
@@ -163,14 +170,12 @@ class TestStructuredLoggingMiddleware:
             middleware = StructuredLoggingMiddleware(
                 include_payload_length=True, estimate_payload_tokens=True
             )
-            message = middleware._create_before_message(mock_context, "test_event")
+            message = middleware._create_before_message(mock_context)
 
             assert message == snapshot(
                 {
-                    "event": "test_event",
-                    "timestamp": "2023-01-01T00:00:00+00:00",
+                    "event": "request_start",
                     "source": "client",
-                    "type": "request",
                     "method": "test_method",
                     "payload_tokens": 24,
                     "payload_length": 98,
@@ -186,16 +191,18 @@ class TestStructuredLoggingMiddleware:
         middleware = StructuredLoggingMiddleware()
         mock_call_next = AsyncMock(return_value="test_result")
 
-        with caplog.at_level(logging.INFO):
+        with caplog_for_fastmcp(caplog):
             result = await middleware.on_message(mock_context, mock_call_next)
 
         assert result == "test_result"
         assert mock_call_next.called
-        assert remove_line_numbers(caplog.text) == snapshot("""\
-INFO     fastmcp.structured:logging.py:LINE_NUMBER Processing message: {"event": "request_start", "timestamp": "2023-01-01T00:00:00+00:00", "method": "test_method", "type": "request", "source": "client"}
-INFO     fastmcp.structured:logging.py:LINE_NUMBER Completed message: {"event": "request_success", "timestamp": "2023-01-01T00:00:00+00:00", "method": "test_method", "type": "request", "source": "client"}
 
-""")
+        assert get_log_lines(caplog) == snapshot(
+            [
+                '{"event": "request_start", "method": "test_method", "source": "client"}',
+                '{"event": "request_success", "method": "test_method", "source": "client", "duration_ms": 0.02}',
+            ]
+        )
 
     async def test_on_message_failure(
         self, mock_context: MiddlewareContext[Any], caplog: pytest.LogCaptureFixture
@@ -204,12 +211,16 @@ INFO     fastmcp.structured:logging.py:LINE_NUMBER Completed message: {"event":
         middleware = StructuredLoggingMiddleware()
         mock_call_next = AsyncMock(side_effect=ValueError("test error"))
 
-        with caplog.at_level(logging.INFO):
+        with caplog_for_fastmcp(caplog):
             with pytest.raises(ValueError):
                 await middleware.on_message(mock_context, mock_call_next)
 
-        assert "Processing message:" in caplog.text
-        assert "Failed message: test_method - test error" in caplog.text
+        assert get_log_lines(caplog) == snapshot(
+            [
+                '{"event": "request_start", "method": "test_method", "source": "client"}',
+                '{"event": "request_error", "method": "test_method", "source": "client", "duration_ms": 0.02, "error": "test error"}',
+            ]
+        )
 
 
 class TestLoggingMiddleware:
@@ -218,7 +229,7 @@ class TestLoggingMiddleware:
     def test_init_default(self):
         """Test default initialization."""
         middleware = LoggingMiddleware()
-        assert middleware.logger.name == "fastmcp.requests"
+        assert middleware.logger.name == "fastmcp.middleware.logging"
         assert middleware.log_level == logging.INFO
         assert middleware.include_payloads is False
         assert middleware.include_payload_length is False
@@ -227,11 +238,11 @@ class TestLoggingMiddleware:
     def test_format_message(self, mock_context: MiddlewareContext[Any]):
         """Test message formatting."""
         middleware = LoggingMiddleware()
-        message = middleware._create_before_message(mock_context, "test_event")
+        message = middleware._create_before_message(mock_context)
         formatted = middleware._format_message(message)
 
         assert formatted == snapshot(
-            "event=test_event timestamp=2023-01-01T00:00:00+00:00 method=test_method type=request source=client"
+            "event=request_start method=test_method source=client"
         )
 
     def test_create_before_message_long_payload(
@@ -240,12 +251,196 @@ class TestLoggingMiddleware:
         """Test message formatting with long payload truncation."""
         middleware = LoggingMiddleware(include_payloads=True, max_payload_length=10)
 
-        message = middleware._create_before_message(mock_context, "test_event")
+        message = middleware._create_before_message(mock_context)
 
         formatted = middleware._format_message(message)
 
-        assert "payload=" in formatted
-        assert "..." in formatted
+        assert formatted == snapshot(
+            'event=request_start method=test_method source=client payload={"method":... payload_type=CallToolRequest'
+        )
+
+    async def test_on_message_failure(
+        self, mock_context: MiddlewareContext[Any], caplog: pytest.LogCaptureFixture
+    ):
+        """Test structured logging of failed messages."""
+        middleware = StructuredLoggingMiddleware()
+        mock_call_next = AsyncMock(side_effect=ValueError("test error"))
+
+        with caplog_for_fastmcp(caplog):
+            with pytest.raises(ValueError):
+                await middleware.on_message(mock_context, mock_call_next)
+
+        # Check that we have structured JSON logs
+        assert get_log_lines(caplog) == snapshot(
+            [
+                '{"event": "request_start", "method": "test_method", "source": "client"}',
+                '{"event": "request_error", "method": "test_method", "source": "client", "duration_ms": 0.02, "error": "test error"}',
+            ]
+        )
+
+    async def test_on_message_with_pydantic_types_in_payload(
+        self,
+        mock_call_next: CallNext[Any, Any],
+        caplog: pytest.LogCaptureFixture,
+    ):
+        """Ensure Pydantic AnyUrl in payload serializes correctly when include_payloads=True."""
+
+        mock_context = new_mock_context(
+            message=mcp.types.ReadResourceRequest(
+                method="resources/read",
+                params=mcp.types.ReadResourceRequestParams(
+                    uri=AnyUrl("test://example/1"),
+                ),
+            )
+        )
+
+        middleware = StructuredLoggingMiddleware(include_payloads=True)
+
+        with caplog_for_fastmcp(caplog):
+            result = await middleware.on_message(mock_context, mock_call_next)
+
+        assert result == "test_result"
+
+        assert get_log_lines(caplog) == snapshot(
+            [
+                '{"event": "request_start", "method": "test_method", "source": "client", "payload": "{\\"method\\":\\"resources/read\\",\\"params\\":{\\"_meta\\":null,\\"uri\\":\\"test://example/1\\"}}", "payload_type": "ReadResourceRequest"}',
+                '{"event": "request_success", "method": "test_method", "source": "client", "duration_ms": 0.02}',
+            ]
+        )
+
+    async def test_on_message_with_resource_template_in_payload(
+        self,
+        mock_call_next: CallNext[Any, Any],
+        caplog: pytest.LogCaptureFixture,
+    ):
+        """Ensure ResourceTemplate in payload serializes via pydantic conversion without errors."""
+
+        mock_context = new_mock_context(
+            message=ResourceTemplate(
+                name="tmpl",
+                uri_template="tmpl://{id}",
+                parameters={"id": {"type": "string"}},
+            )
+        )
+
+        middleware = StructuredLoggingMiddleware(include_payloads=True)
+
+        with caplog_for_fastmcp(caplog):
+            result = await middleware.on_message(mock_context, mock_call_next)
+
+        assert result == "test_result"
+
+        assert get_log_lines(caplog) == snapshot(
+            [
+                '{"event": "request_start", "method": "test_method", "source": "client", "payload": "{\\"name\\":\\"tmpl\\",\\"title\\":null,\\"description\\":null,\\"tags\\":[],\\"meta\\":null,\\"enabled\\":true,\\"uri_template\\":\\"tmpl://{id}\\",\\"mime_type\\":\\"text/plain\\",\\"parameters\\":{\\"id\\":{\\"type\\":\\"string\\"}},\\"annotations\\":null}", "payload_type": "ResourceTemplate"}',
+                '{"event": "request_success", "method": "test_method", "source": "client", "duration_ms": 0.02}',
+            ]
+        )
+
+    async def test_on_message_with_nonserializable_payload_falls_back_to_str(
+        self, mock_call_next: CallNext[Any, Any], caplog: pytest.LogCaptureFixture
+    ):
+        """Ensure non-JSONable objects fall back to string serialization in payload."""
+
+        class NonSerializable:
+            def __str__(self) -> str:
+                return "NON_SERIALIZABLE"
+
+        mock_context = new_mock_context(
+            message=mcp.types.CallToolRequest(
+                method="tools/call",
+                params=mcp.types.CallToolRequestParams(
+                    name="test_method",
+                    arguments={"obj": NonSerializable()},
+                ),
+            )
+        )
+
+        middleware = StructuredLoggingMiddleware(include_payloads=True)
+
+        with caplog_for_fastmcp(caplog):
+            result = await middleware.on_message(mock_context, mock_call_next)
+
+        assert result == "test_result"
+
+        assert get_log_lines(caplog) == snapshot(
+            [
+                '{"event": "request_start", "method": "test_method", "source": "client", "payload": "{\\"method\\":\\"tools/call\\",\\"params\\":{\\"_meta\\":null,\\"name\\":\\"test_method\\",\\"arguments\\":{\\"obj\\":\\"NON_SERIALIZABLE\\"}}}", "payload_type": "CallToolRequest"}',
+                '{"event": "request_success", "method": "test_method", "source": "client", "duration_ms": 0.02}',
+            ]
+        )
+
+    async def test_on_message_with_custom_serializer_applied(
+        self, mock_call_next: CallNext[Any, Any], caplog: pytest.LogCaptureFixture
+    ):
+        """Ensure a custom serializer is used for non-JSONable payloads."""
+
+        # Provide a serializer that replaces entire payload with a fixed string
+        def custom_serializer(_: Any) -> str:
+            return "CUSTOM_PAYLOAD"
+
+        mock_context = new_mock_context(
+            message=mcp.types.CallToolRequest(
+                method="tools/call",
+                params=mcp.types.CallToolRequestParams(
+                    name="test_method",
+                    arguments={"obj": "OBJECT"},
+                ),
+            )
+        )
+
+        middleware = StructuredLoggingMiddleware(
+            include_payloads=True, payload_serializer=custom_serializer
+        )
+
+        with caplog_for_fastmcp(caplog):
+            result = await middleware.on_message(mock_context, mock_call_next)
+
+        assert result == "test_result"
+
+        assert get_log_lines(caplog) == snapshot(
+            [
+                '{"event": "request_start", "method": "test_method", "source": "client", "payload": "CUSTOM_PAYLOAD", "payload_type": "CallToolRequest"}',
+                '{"event": "request_success", "method": "test_method", "source": "client", "duration_ms": 0.02}',
+            ]
+        )
+
+
+@pytest.fixture
+def logging_server():
+    """Create a FastMCP server specifically for logging middleware tests."""
+    from fastmcp import FastMCP
+
+    mcp = FastMCP("LoggingTestServer")
+
+    @mcp.tool
+    def simple_operation(data: str) -> str:
+        """A simple operation for testing logging."""
+        return f"Processed: {data}"
+
+    @mcp.tool
+    def complex_operation(items: list[str], mode: str = "default") -> dict:
+        """A complex operation with structured data."""
+        return {"processed_items": len(items), "mode": mode, "result": "success"}
+
+    @mcp.tool
+    def operation_with_error(should_fail: bool = False) -> str:
+        """An operation that can be made to fail."""
+        if should_fail:
+            raise ValueError("Operation failed intentionally")
+        return "Operation completed successfully"
+
+    @mcp.resource("log://test")
+    def test_resource() -> str:
+        """A test resource for logging."""
+        return "Test resource content"
+
+    @mcp.prompt
+    def test_prompt() -> str:
+        """A test prompt for logging."""
+        return "Test prompt content"
+
+    return mcp
 
 
 class TestLoggingMiddlewareIntegration:
@@ -290,33 +485,29 @@ class TestLoggingMiddlewareIntegration:
     ):
         """Test that logging middleware captures successful operations."""
         logging_middleware = LoggingMiddleware(methods=["tools/call"])
-        logging_middleware._get_timestamp_from_context = (  # ty: ignore[invalid-assignment]
-            lambda _: FIXED_DATE.isoformat()
-        )
 
         logging_server.add_middleware(logging_middleware)
 
-        with caplog.at_level(logging.INFO):
-            async with Client(logging_server) as client:
-                await client.call_tool(
-                    name="simple_operation", arguments={"data": "test_data"}
-                )
-                await client.call_tool(
-                    name="complex_operation",
-                    arguments={"items": ["a", "b", "c"], "mode": "batch"},
-                )
+        with caplog_for_fastmcp(caplog):
+            with caplog.at_level(logging.INFO):
+                async with Client(logging_server) as client:
+                    await client.call_tool(
+                        name="simple_operation", arguments={"data": "test_data"}
+                    )
+                    await client.call_tool(
+                        name="complex_operation",
+                        arguments={"items": ["a", "b", "c"], "mode": "batch"},
+                    )
 
         # Should have processing and completion logs for both operations
-        assert remove_line_numbers(caplog.text) == snapshot("""\
-INFO     mcp.server.lowlevel.server:server.py:LINE_NUMBER Processing request of type CallToolRequest
-INFO     fastmcp.requests:logging.py:LINE_NUMBER Processing message: event=request_start timestamp=2023-01-01T00:00:00+00:00 method=tools/call type=request source=client
-INFO     fastmcp.requests:logging.py:LINE_NUMBER Completed message: event=request_success timestamp=2023-01-01T00:00:00+00:00 method=tools/call type=request source=client
-INFO     mcp.server.lowlevel.server:server.py:LINE_NUMBER Processing request of type ListToolsRequest
-INFO     mcp.server.lowlevel.server:server.py:LINE_NUMBER Processing request of type CallToolRequest
-INFO     fastmcp.requests:logging.py:LINE_NUMBER Processing message: event=request_start timestamp=2023-01-01T00:00:00+00:00 method=tools/call type=request source=client
-INFO     fastmcp.requests:logging.py:LINE_NUMBER Completed message: event=request_success timestamp=2023-01-01T00:00:00+00:00 method=tools/call type=request source=client
-
-""")
+        assert get_log_lines(caplog) == snapshot(
+            [
+                "event=request_start method=tools/call source=client",
+                "event=request_success method=tools/call source=client duration_ms=0.02",
+                "event=request_start method=tools/call source=client",
+                "event=request_success method=tools/call source=client duration_ms=0.02",
+            ]
+        )
 
     async def test_logging_middleware_logs_failures(
         self, logging_server: FastMCP, caplog: pytest.LogCaptureFixture
@@ -324,7 +515,7 @@ INFO     fastmcp.requests:logging.py:LINE_NUMBER Completed message: event=reques
         """Test that logging middleware captures failed operations."""
         logging_server.add_middleware(LoggingMiddleware(methods=["tools/call"]))
 
-        with caplog.at_level(logging.INFO):
+        with caplog_for_fastmcp(caplog):
             async with Client(logging_server) as client:
                 # This should fail and be logged
                 with pytest.raises(Exception):
@@ -335,8 +526,9 @@ INFO     fastmcp.requests:logging.py:LINE_NUMBER Completed message: event=reques
         log_text = caplog.text
 
         # Should have processing and failure logs
-        assert "Processing message:" in log_text
-        assert "Failed message: tools/call" in log_text
+        assert log_text.splitlines()[-1] == snapshot(
+            "ERROR    fastmcp.middleware.logging:logging.py:122 event=request_error method=tools/call source=client duration_ms=0.02 error=Error calling tool 'operation_with_error': Operation failed intentionally"
+        )
 
     async def test_logging_middleware_with_payloads(
         self, logging_server: FastMCP, caplog: pytest.LogCaptureFixture
@@ -346,24 +538,18 @@ INFO     fastmcp.requests:logging.py:LINE_NUMBER Completed message: event=reques
         middleware = LoggingMiddleware(
             include_payloads=True, max_payload_length=500, methods=["tools/call"]
         )
-        middleware._get_timestamp_from_context = (  # ty: ignore[invalid-assignment]
-            lambda _: FIXED_DATE.isoformat()
-        )
         logging_server.add_middleware(middleware)
 
-        with caplog.at_level(logging.INFO):
+        with caplog_for_fastmcp(caplog):
             async with Client(logging_server) as client:
                 await client.call_tool("simple_operation", {"data": "payload_test"})
 
-        log_text = caplog.text
-
-        assert remove_line_numbers(log_text) == snapshot("""\
-INFO     mcp.server.lowlevel.server:server.py:LINE_NUMBER Processing request of type CallToolRequest
-INFO     fastmcp.requests:logging.py:LINE_NUMBER Processing message: event=request_start timestamp=2023-01-01T00:00:00+00:00 method=tools/call type=request source=client payload={"_meta":null,"name":"simple_operation","arguments":{"data":"payload_test"}} payload_type=CallToolRequestParams
-INFO     fastmcp.requests:logging.py:LINE_NUMBER Completed message: event=request_success timestamp=2023-01-01T00:00:00+00:00 method=tools/call type=request source=client
-INFO     mcp.server.lowlevel.server:server.py:LINE_NUMBER Processing request of type ListToolsRequest
-
-""")
+        assert get_log_lines(caplog) == snapshot(
+            [
+                'event=request_start method=tools/call source=client payload={"_meta":null,"name":"simple_operation","arguments":{"data":"payload_test"}} payload_type=CallToolRequestParams',
+                "event=request_success method=tools/call source=client duration_ms=0.02",
+            ]
+        )
 
     async def test_structured_logging_middleware_produces_json(
         self, logging_server: FastMCP, caplog: pytest.LogCaptureFixture
@@ -373,34 +559,21 @@ INFO     mcp.server.lowlevel.server:server.py:LINE_NUMBER Processing request of
         logging_middleware = StructuredLoggingMiddleware(
             include_payloads=True, methods=["tools/call"]
         )
-        logging_middleware._get_timestamp_from_context = (  # ty: ignore[invalid-assignment]
-            lambda _: FIXED_DATE.isoformat()
-        )
 
         logging_server.add_middleware(logging_middleware)
 
-        with caplog.at_level(logging.INFO):
+        with caplog_for_fastmcp(caplog):
             async with Client(logging_server) as client:
                 await client.call_tool(
                     name="simple_operation", arguments={"data": "json_test"}
                 )
 
-        # Extract JSON log entries
-        log_lines = [
-            record.message
-            for record in caplog.records
-            if record.name == "fastmcp.structured"
-        ]
-
-        assert len(log_lines) >= 2  # Should have start and success entries
-
-        assert remove_line_numbers(caplog.text) == snapshot("""\
-INFO     mcp.server.lowlevel.server:server.py:LINE_NUMBER Processing request of type CallToolRequest
-INFO     fastmcp.structured:logging.py:LINE_NUMBER Processing message: {"event": "request_start", "timestamp": "2023-01-01T00:00:00+00:00", "method": "tools/call", "type": "request", "source": "client", "payload": "{\\"_meta\\":null,\\"name\\":\\"simple_operation\\",\\"arguments\\":{\\"data\\":\\"json_test\\"}}", "payload_type": "CallToolRequestParams"}
-INFO     fastmcp.structured:logging.py:LINE_NUMBER Completed message: {"event": "request_success", "timestamp": "2023-01-01T00:00:00+00:00", "method": "tools/call", "type": "request", "source": "client"}
-INFO     mcp.server.lowlevel.server:server.py:LINE_NUMBER Processing request of type ListToolsRequest
-
-""")
+        assert get_log_lines(caplog) == snapshot(
+            [
+                '{"event": "request_start", "method": "tools/call", "source": "client", "payload": "{\\"_meta\\":null,\\"name\\":\\"simple_operation\\",\\"arguments\\":{\\"data\\":\\"json_test\\"}}", "payload_type": "CallToolRequestParams"}',
+                '{"event": "request_success", "method": "tools/call", "source": "client", "duration_ms": 0.02}',
+            ]
+        )
 
     async def test_structured_logging_middleware_handles_errors(
         self, logging_server: FastMCP, caplog: pytest.LogCaptureFixture
@@ -408,25 +581,23 @@ INFO     mcp.server.lowlevel.server:server.py:LINE_NUMBER Processing request of
         """Test structured logging of errors with JSON format."""
 
         logging_middleware = StructuredLoggingMiddleware(methods=["tools/call"])
-        logging_middleware._get_timestamp_from_context = (  # ty: ignore[invalid-assignment]
-            lambda _: FIXED_DATE.isoformat()
-        )
 
         logging_server.add_middleware(logging_middleware)
 
-        with caplog.at_level(logging.INFO):
-            async with Client(logging_server) as client:
-                with pytest.raises(Exception):
-                    await client.call_tool(
-                        "operation_with_error", {"should_fail": True}
-                    )
+        with caplog_for_fastmcp(caplog):
+            with caplog.at_level(logging.INFO):
+                async with Client(logging_server) as client:
+                    with pytest.raises(Exception):
+                        await client.call_tool(
+                            "operation_with_error", {"should_fail": True}
+                        )
 
-        assert remove_line_numbers(caplog.text) == snapshot("""\
-INFO     mcp.server.lowlevel.server:server.py:LINE_NUMBER Processing request of type CallToolRequest
-INFO     fastmcp.structured:logging.py:LINE_NUMBER Processing message: {"event": "request_start", "timestamp": "2023-01-01T00:00:00+00:00", "method": "tools/call", "type": "request", "source": "client"}
-ERROR    fastmcp.structured:logging.py:LINE_NUMBER Failed message: tools/call - Error calling tool 'operation_with_error': Operation failed intentionally
-
-""")
+        assert get_log_lines(caplog) == snapshot(
+            [
+                '{"event": "request_start", "method": "tools/call", "source": "client"}',
+                '{"event": "request_error", "method": "tools/call", "source": "client", "duration_ms": 0.02, "error": "Error calling tool \'operation_with_error\': Operation failed intentionally"}',
+            ]
+        )
 
     async def test_logging_middleware_with_different_operations(
         self, logging_server: FastMCP, caplog: pytest.LogCaptureFixture
@@ -444,7 +615,7 @@ ERROR    fastmcp.structured:logging.py:LINE_NUMBER Failed message: tools/call -
             )
         )
 
-        with caplog.at_level(logging.INFO):
+        with caplog_for_fastmcp(caplog):
             async with Client(logging_server) as client:
                 # Test different operation types
                 await client.call_tool("simple_operation", {"data": "test"})
@@ -452,16 +623,18 @@ ERROR    fastmcp.structured:logging.py:LINE_NUMBER Failed message: tools/call -
                 await client.get_prompt("test_prompt")
                 await client.list_resources()
 
-        log_text = caplog.text
-
-        # Should have logs for all different operation types
-        # Note: Different operations may have different method names
-        processing_count = log_text.count("Processing message:")
-        completion_count = log_text.count("Completed message:")
-
-        # Should have processed all 4 operations
-        assert processing_count == 4
-        assert completion_count == 4
+        assert get_log_lines(caplog) == snapshot(
+            [
+                "event=request_start method=tools/call source=client",
+                "event=request_success method=tools/call source=client duration_ms=0.02",
+                "event=request_start method=resources/read source=client",
+                "event=request_success method=resources/read source=client duration_ms=0.02",
+                "event=request_start method=prompts/get source=client",
+                "event=request_success method=prompts/get source=client duration_ms=0.02",
+                "event=request_start method=resources/list source=client",
+                "event=request_success method=resources/list source=client duration_ms=0.02",
+            ]
+        )
 
     async def test_logging_middleware_custom_configuration(
         self, logging_server: FastMCP
@@ -491,5 +664,7 @@ ERROR    fastmcp.structured:logging.py:LINE_NUMBER Failed message: tools/call -
 
         # Check that our custom logger captured the logs
         log_output = log_buffer.getvalue()
-        assert "Processing message:" in log_output
-        assert "payload=" in log_output
+        assert log_output == snapshot("""\
+event=request_start method=tools/call source=client payload={"_meta":null,"name":"simple_operation","arguments":{"data":"custom_test"}} payload_type=CallToolRequestParams
+event=request_success method=tools/call source=client duration_ms=0.02
+""")

tests/server/middleware/test_middleware.py

@@ -293,6 +293,17 @@ class TestMiddlewareHooks:
         result = list_prompts_calls[0].result
         assert isinstance(result, list)
 
+    async def test_initialize(
+        self, mcp_server: FastMCP, recording_middleware: RecordingMiddleware
+    ):
+        async with Client(mcp_server) as client:
+            await client.ping()
+
+        assert recording_middleware.assert_called(at_least=1)
+        assert recording_middleware.assert_called(hook="on_message", at_least=1)
+        assert recording_middleware.assert_called(hook="on_request", at_least=1)
+        assert recording_middleware.assert_called(hook="on_initialize", at_least=1)
+
     async def test_list_tools_filtering_middleware(self):
         """Test that middleware can filter tools."""
 

tests/server/middleware/test_rate_limiting.py

@@ -306,9 +306,10 @@ class TestRateLimitingMiddlewareIntegration:
 
     async def test_rate_limiting_blocks_rapid_requests(self, rate_limit_server):
         """Test that rate limiting blocks rapid successive requests."""
-        # Very restrictive rate limit (accounting for extra list_tools calls per tool call)
+        # Very restrictive rate limit (accounting for initialization and list_tools calls)
+        # Requests: 1 initialize + 1 list_tools + 4 call_tools = 6 total before limit
         rate_limit_server.add_middleware(
-            RateLimitingMiddleware(max_requests_per_second=10.0, burst_capacity=5)
+            RateLimitingMiddleware(max_requests_per_second=10.0, burst_capacity=6)
         )
 
         async with Client(rate_limit_server) as client:
@@ -356,7 +357,7 @@ class TestRateLimitingMiddlewareIntegration:
         """Test sliding window rate limiting implementation."""
         rate_limit_server.add_middleware(
             SlidingWindowRateLimitingMiddleware(
-                max_requests=5,  # Accounting for extra list_tools calls
+                max_requests=6,  # 1 init + 1 list_tools + 3 calls + 1 to fail
                 window_minutes=1,  # 1-minute window
             )
         )
@@ -374,7 +375,7 @@ class TestRateLimitingMiddlewareIntegration:
     async def test_rate_limiting_with_different_operations(self, rate_limit_server):
         """Test that rate limiting applies to all types of operations."""
         rate_limit_server.add_middleware(
-            RateLimitingMiddleware(max_requests_per_second=9.0, burst_capacity=4)
+            RateLimitingMiddleware(max_requests_per_second=9.0, burst_capacity=5)
         )
 
         async with Client(rate_limit_server) as client:
@@ -395,8 +396,8 @@ class TestRateLimitingMiddlewareIntegration:
 
         rate_limit_server.add_middleware(
             RateLimitingMiddleware(
-                max_requests_per_second=6.0,  # Accounting for extra list_tools calls
-                burst_capacity=3,
+                max_requests_per_second=6.0,  # Accounting for initialization and list_tools calls
+                burst_capacity=4,
                 get_client_id=get_client_id,
             )
         )
@@ -416,8 +417,8 @@ class TestRateLimitingMiddlewareIntegration:
         rate_limit_server.add_middleware(
             RateLimitingMiddleware(
                 max_requests_per_second=6.0,
-                burst_capacity=4,
-                global_limit=True,  # Accounting for extra list_tools calls
+                burst_capacity=5,  # 1 init + 2 list_tools + 2 calls before limit
+                global_limit=True,  # Accounting for initialization and list_tools calls
             )
         )
 
@@ -435,7 +436,7 @@ class TestRateLimitingMiddlewareIntegration:
         rate_limit_server.add_middleware(
             RateLimitingMiddleware(
                 max_requests_per_second=10.0,  # 10 per second = 1 every 100ms
-                burst_capacity=3,
+                burst_capacity=4,
             )
         )
 

tests/server/middleware/test_timing.py

@@ -11,6 +11,7 @@ from fastmcp import FastMCP
 from fastmcp.client import Client
 from fastmcp.server.middleware.middleware import MiddlewareContext
 from fastmcp.server.middleware.timing import DetailedTimingMiddleware, TimingMiddleware
+from fastmcp.utilities.tests import caplog_for_fastmcp
 
 
 @pytest.fixture
@@ -47,7 +48,7 @@ class TestTimingMiddleware:
         """Test timing successful requests."""
         middleware = TimingMiddleware()
 
-        with caplog.at_level(logging.INFO):
+        with caplog_for_fastmcp(caplog):
             result = await middleware.on_request(mock_context, mock_call_next)
 
         assert result == "test_result"
@@ -60,7 +61,7 @@ class TestTimingMiddleware:
         middleware = TimingMiddleware()
         mock_call_next = AsyncMock(side_effect=ValueError("test error"))
 
-        with caplog.at_level(logging.INFO):
+        with caplog_for_fastmcp(caplog):
             with pytest.raises(ValueError):
                 await middleware.on_request(mock_context, mock_call_next)
 
@@ -84,7 +85,7 @@ class TestDetailedTimingMiddleware:
         context.message.name = "test_tool"
         mock_call_next = AsyncMock(return_value="tool_result")
 
-        with caplog.at_level(logging.INFO):
+        with caplog_for_fastmcp(caplog):
             result = await middleware.on_call_tool(context, mock_call_next)
 
         assert result == "tool_result"
@@ -97,7 +98,7 @@ class TestDetailedTimingMiddleware:
         context.message.uri = "test://resource"
         mock_call_next = AsyncMock(return_value="resource_result")
 
-        with caplog.at_level(logging.INFO):
+        with caplog_for_fastmcp(caplog):
             result = await middleware.on_read_resource(context, mock_call_next)
 
         assert result == "resource_result"
@@ -110,7 +111,7 @@ class TestDetailedTimingMiddleware:
         context.message.name = "test_prompt"
         mock_call_next = AsyncMock(return_value="prompt_result")
 
-        with caplog.at_level(logging.INFO):
+        with caplog_for_fastmcp(caplog):
             result = await middleware.on_get_prompt(context, mock_call_next)
 
         assert result == "prompt_result"
@@ -122,7 +123,7 @@ class TestDetailedTimingMiddleware:
         context = MagicMock()
         mock_call_next = AsyncMock(return_value="tools_result")
 
-        with caplog.at_level(logging.INFO):
+        with caplog_for_fastmcp(caplog):
             result = await middleware.on_list_tools(context, mock_call_next)
 
         assert result == "tools_result"
@@ -135,7 +136,7 @@ class TestDetailedTimingMiddleware:
         context.message.name = "failing_tool"
         mock_call_next = AsyncMock(side_effect=RuntimeError("operation failed"))
 
-        with caplog.at_level(logging.INFO):
+        with caplog_for_fastmcp(caplog):
             with pytest.raises(RuntimeError):
                 await middleware.on_call_tool(context, mock_call_next)
 
@@ -155,15 +156,15 @@ def timing_server():
 
     @mcp.tool
     def short_task() -> str:
-        """A task that takes 0.1 seconds."""
-        time.sleep(0.1)
-        return "Done after 0.1s"
+        """A task that takes 0.01 seconds."""
+        time.sleep(0.01)
+        return "Done after 0.01 seconds"
 
     @mcp.tool
     def medium_task() -> str:
-        """A task that takes 0.15 seconds."""
-        time.sleep(0.15)
-        return "Done after 0.15s"
+        """A task that takes 0.02 seconds."""
+        time.sleep(0.02)
+        return "Done after 0.02 seconds"
 
     @mcp.tool
     def failing_task() -> str:
@@ -173,14 +174,14 @@ def timing_server():
     @mcp.resource("timer://test")
     def test_resource() -> str:
         """A resource that takes time to read."""
-        time.sleep(0.05)
-        return "Resource content after 0.05s"
+        time.sleep(0.005)
+        return "Resource content after 0.005 seconds"
 
     @mcp.prompt
     def test_prompt() -> str:
         """A prompt that takes time to generate."""
-        time.sleep(0.08)
-        return "Prompt content after 0.08s"
+        time.sleep(0.008)
+        return "Prompt content after 0.008 seconds"
 
     return mcp
 
@@ -194,7 +195,7 @@ class TestTimingMiddlewareIntegration:
         """Test that timing middleware accurately measures tool execution times."""
         timing_server.add_middleware(TimingMiddleware())
 
-        with caplog.at_level(logging.INFO):
+        with caplog_for_fastmcp(caplog):
             async with Client(timing_server) as client:
                 # Test instant task
                 await client.call_tool("instant_task")
@@ -225,7 +226,7 @@ class TestTimingMiddlewareIntegration:
         """Test that timing middleware measures time even for failed operations."""
         timing_server.add_middleware(TimingMiddleware())
 
-        with caplog.at_level(logging.INFO):
+        with caplog_for_fastmcp(caplog):
             async with Client(timing_server) as client:
                 # This should fail but still be timed
                 with pytest.raises(Exception):
@@ -241,7 +242,7 @@ class TestTimingMiddlewareIntegration:
         """Test that detailed timing middleware provides operation-specific timing."""
         timing_server.add_middleware(DetailedTimingMiddleware())
 
-        with caplog.at_level(logging.INFO):
+        with caplog_for_fastmcp(caplog):
             async with Client(timing_server) as client:
                 # Test tool call
                 await client.call_tool("short_task")
@@ -271,7 +272,7 @@ class TestTimingMiddlewareIntegration:
         """Test timing middleware with concurrent operations."""
         timing_server.add_middleware(TimingMiddleware())
 
-        with caplog.at_level(logging.INFO):
+        with caplog_for_fastmcp(caplog):
             async with Client(timing_server) as client:
                 # Run multiple operations concurrently
                 tasks = [
@@ -290,7 +291,7 @@ class TestTimingMiddlewareIntegration:
             len(timing_logs) >= 3
         )  # At least 3 tool calls, may have additional list_tools calls
 
-    async def test_timing_middleware_custom_logger(self, timing_server):
+    async def test_timing_middleware_custom_logger(self, timing_server, caplog):
         """Test timing middleware with custom logger configuration."""
         import io
         import logging

tests/server/openapi/test_openapi_path_parameters.py

@@ -163,7 +163,7 @@ async def test_integration_array_path_parameter(array_path_spec, mock_client):
     mcp = FastMCP.from_openapi(array_path_spec, client=mock_client)
 
     # Call the tool with a single value
-    await mcp._mcp_call_tool("test_operation", {"days": ["monday"]})
+    await mcp._call_tool_mcp("test_operation", {"days": ["monday"]})
 
     # Check the request was made correctly
     mock_client.request.assert_called_with(
@@ -177,7 +177,7 @@ async def test_integration_array_path_parameter(array_path_spec, mock_client):
     mock_client.request.reset_mock()
 
     # Call the tool with multiple values
-    await mcp._mcp_call_tool("test_operation", {"days": ["monday", "tuesday"]})
+    await mcp._call_tool_mcp("test_operation", {"days": ["monday", "tuesday"]})
 
     # Check the request was made correctly
     mock_client.request.assert_called_with(

tests/server/proxy/test_proxy_server.py

@@ -173,15 +173,15 @@ class TestTools:
 
     async def test_list_tools_same_as_original(self, fastmcp_server, proxy_server):
         assert (
-            await proxy_server._mcp_list_tools()
-            == await fastmcp_server._mcp_list_tools()
+            await proxy_server._list_tools_mcp()
+            == await fastmcp_server._list_tools_mcp()
         )
 
     async def test_call_tool_result_same_as_original(
         self, fastmcp_server: FastMCP, proxy_server: FastMCPProxy
     ):
-        result = await fastmcp_server._mcp_call_tool("greet", {"name": "Alice"})
-        proxy_result = await proxy_server._mcp_call_tool("greet", {"name": "Alice"})
+        result = await fastmcp_server._call_tool_mcp("greet", {"name": "Alice"})
+        proxy_result = await proxy_server._call_tool_mcp("greet", {"name": "Alice"})
 
         assert result == proxy_result
 
@@ -267,8 +267,8 @@ class TestResources:
 
     async def test_list_resources_same_as_original(self, fastmcp_server, proxy_server):
         assert (
-            await proxy_server._mcp_list_resources()
-            == await fastmcp_server._mcp_list_resources()
+            await proxy_server._list_resources_mcp()
+            == await fastmcp_server._list_resources_mcp()
         )
 
     async def test_read_resource(self, proxy_server: FastMCPProxy):
@@ -367,8 +367,8 @@ class TestResourceTemplates:
     async def test_list_resource_templates_same_as_original(
         self, fastmcp_server, proxy_server
     ):
-        result = await fastmcp_server._mcp_list_resource_templates()
-        proxy_result = await proxy_server._mcp_list_resource_templates()
+        result = await fastmcp_server._list_resource_templates_mcp()
+        proxy_result = await proxy_server._list_resource_templates_mcp()
         assert proxy_result == result
 
     @pytest.mark.parametrize("id", [1, 2, 3])

tests/server/test_file_server.py

@@ -74,7 +74,7 @@ def tools(mcp: FastMCP, test_dir: Path) -> FastMCP:
 
 
 async def test_list_resources(mcp: FastMCP):
-    resources = await mcp._mcp_list_resources()
+    resources = await mcp._list_resources_mcp()
     assert len(resources) == 4
 
     assert [str(r.uri) for r in resources] == [
@@ -86,7 +86,7 @@ async def test_list_resources(mcp: FastMCP):
 
 
 async def test_read_resource_dir(mcp: FastMCP):
-    res_iter = await mcp._mcp_read_resource("dir://test_dir")
+    res_iter = await mcp._read_resource_mcp("dir://test_dir")
     res_list = list(res_iter)
     assert len(res_list) == 1
     res = res_list[0]
@@ -102,7 +102,7 @@ async def test_read_resource_dir(mcp: FastMCP):
 
 
 async def test_read_resource_file(mcp: FastMCP):
-    res_iter = await mcp._mcp_read_resource("file://test_dir/example.py")
+    res_iter = await mcp._read_resource_mcp("file://test_dir/example.py")
     res_list = list(res_iter)
     assert len(res_list) == 1
     res = res_list[0]
@@ -110,17 +110,17 @@ async def test_read_resource_file(mcp: FastMCP):
 
 
 async def test_delete_file(mcp: FastMCP, test_dir: Path):
-    await mcp._mcp_call_tool(
+    await mcp._call_tool_mcp(
         "delete_file", arguments=dict(path=str(test_dir / "example.py"))
     )
     assert not (test_dir / "example.py").exists()
 
 
 async def test_delete_file_and_check_resources(mcp: FastMCP, test_dir: Path):
-    await mcp._mcp_call_tool(
+    await mcp._call_tool_mcp(
         "delete_file", arguments=dict(path=str(test_dir / "example.py"))
     )
-    res_iter = await mcp._mcp_read_resource("file://test_dir/example.py")
+    res_iter = await mcp._read_resource_mcp("file://test_dir/example.py")
     res_list = list(res_iter)
     assert len(res_list) == 1
     res = res_list[0]

tests/server/test_log_level.py

@@ -0,0 +1,88 @@
+"""Test log_level parameter support in FastMCP server."""
+
+import asyncio
+from unittest.mock import AsyncMock, patch
+
+from fastmcp import FastMCP
+
+
+class TestLogLevelParameter:
+    """Test that log_level parameter is properly accepted by run methods."""
+
+    async def test_run_stdio_accepts_log_level(self):
+        """Test that run_stdio_async accepts log_level parameter."""
+        server = FastMCP("TestServer")
+
+        # Mock the stdio_server to avoid actual stdio operations
+        with patch("fastmcp.server.server.stdio_server") as mock_stdio:
+            mock_stdio.return_value.__aenter__ = AsyncMock(
+                return_value=(AsyncMock(), AsyncMock())
+            )
+            mock_stdio.return_value.__aexit__ = AsyncMock()
+
+            # Mock the underlying MCP server run method
+            with patch.object(server._mcp_server, "run", new_callable=AsyncMock):
+                try:
+                    # This should accept the log_level parameter without error
+                    await asyncio.wait_for(
+                        server.run_stdio_async(log_level="DEBUG", show_banner=False),
+                        timeout=0.1,
+                    )
+                except asyncio.TimeoutError:
+                    pass  # Expected since we're mocking
+
+    async def test_run_http_accepts_log_level(self):
+        """Test that run_http_async accepts log_level parameter."""
+        server = FastMCP("TestServer")
+
+        # Mock uvicorn to avoid actual server start
+        with patch("fastmcp.server.server.uvicorn.Server") as mock_server_class:
+            mock_instance = mock_server_class.return_value
+            mock_instance.serve = AsyncMock()
+
+            # This should accept the log_level parameter without error
+            await server.run_http_async(
+                log_level="INFO", show_banner=False, host="127.0.0.1", port=8000
+            )
+
+            # Verify serve was called
+            mock_instance.serve.assert_called_once()
+
+    async def test_run_async_passes_log_level(self):
+        """Test that run_async passes log_level to transport methods."""
+        server = FastMCP("TestServer")
+
+        # Test stdio transport
+        with patch.object(
+            server, "run_stdio_async", new_callable=AsyncMock
+        ) as mock_stdio:
+            await server.run_async(transport="stdio", log_level="WARNING")
+            mock_stdio.assert_called_once_with(show_banner=True, log_level="WARNING")
+
+        # Test http transport
+        with patch.object(
+            server, "run_http_async", new_callable=AsyncMock
+        ) as mock_http:
+            await server.run_async(transport="http", log_level="ERROR")
+            mock_http.assert_called_once_with(
+                transport="http", show_banner=True, log_level="ERROR"
+            )
+
+    def test_sync_run_accepts_log_level(self):
+        """Test that the synchronous run method accepts log_level."""
+        server = FastMCP("TestServer")
+
+        with patch.object(server, "run_async", new_callable=AsyncMock):
+            # Mock anyio.run to avoid actual async execution
+            with patch("anyio.run") as mock_anyio_run:
+                server.run(transport="stdio", log_level="CRITICAL")
+
+                # Verify anyio.run was called
+                mock_anyio_run.assert_called_once()
+
+                # Get the function that was passed to anyio.run
+                called_func = mock_anyio_run.call_args[0][0]
+
+                # The function should be a partial that includes log_level
+                assert hasattr(called_func, "keywords")
+                assert called_func.keywords.get("log_level") == "CRITICAL"

tests/server/test_server.py

@@ -76,7 +76,7 @@ class TestTools:
         def fn(x: int) -> int:
             return x + 1
 
-        mcp_tools = await mcp._mcp_list_tools()
+        mcp_tools = await mcp._list_tools_mcp()
         assert len(mcp_tools) == 1
         assert mcp_tools[0].name == "fn"
 
@@ -89,7 +89,7 @@ class TestTools:
         def fn(x: int) -> int:
             return x + 1
 
-        mcp_tools = await mcp._mcp_list_tools()
+        mcp_tools = await mcp._list_tools_mcp()
         assert len(mcp_tools) == 1
         assert mcp_tools[0].name == "custom_name"
 
@@ -110,7 +110,7 @@ class TestTools:
         assert "adder" not in mcp_tools
 
         with pytest.raises(NotFoundError, match="Unknown tool: adder"):
-            await mcp._mcp_call_tool("adder", {"a": 1, "b": 2})
+            await mcp._call_tool_mcp("adder", {"a": 1, "b": 2})
 
     async def test_add_tool_at_init(self):
         def f(x: int) -> int:
@@ -136,7 +136,7 @@ class TestToolDecorator:
         mcp = FastMCP()
 
         with pytest.raises(NotFoundError, match="Unknown tool: add"):
-            await mcp._mcp_call_tool("add", {"x": 1, "y": 2})
+            await mcp._call_tool_mcp("add", {"x": 1, "y": 2})
 
     async def test_tool_decorator(self):
         mcp = FastMCP()
@@ -185,7 +185,7 @@ class TestToolDecorator:
         def add(x: int, y: int) -> int:
             return x + y
 
-        tools = await mcp._mcp_list_tools()
+        tools = await mcp._list_tools_mcp()
         assert len(tools) == 1
         tool = tools[0]
         assert tool.description == "Add two numbers"

tests/server/test_server_interactions.py

@@ -935,12 +935,13 @@ class TestToolOutputSchema:
             assert len(tools) == 1
 
             type_schema = TypeAdapter(annotation).json_schema()
+            # Remove title fields from the schema for comparison (title pruning is enabled)
+            type_schema = compress_schema(type_schema, prune_titles=True)
             # this line will fail until MCP adds output schemas!!
             assert tools[0].outputSchema == {
                 "type": "object",
-                "properties": {"result": {**type_schema, "title": "Result"}},
+                "properties": {"result": type_schema},
                 "required": ["result"],
-                "title": "_WrappedResult",
                 "x-fastmcp-wrap-result": True,
             }
 
@@ -958,7 +959,9 @@ class TestToolOutputSchema:
         async with Client(mcp) as client:
             tools = await client.list_tools()
 
-            type_schema = compress_schema(TypeAdapter(annotation).json_schema())
+            type_schema = compress_schema(
+                TypeAdapter(annotation).json_schema(), prune_titles=True
+            )
             assert len(tools) == 1
 
             # Normalize anyOf ordering for comparison since union type order
@@ -1071,9 +1074,8 @@ class TestToolOutputSchema:
             tool = next(t for t in tools if t.name == "primitive_tool")
             expected_schema = {
                 "type": "object",
-                "properties": {"result": {"type": "string", "title": "Result"}},
+                "properties": {"result": {"type": "string"}},
                 "required": ["result"],
-                "title": "_WrappedResult",
                 "x-fastmcp-wrap-result": True,
             }
             assert tool.outputSchema == expected_schema
@@ -1095,12 +1097,13 @@ class TestToolOutputSchema:
             # List tools and verify schema shows wrapped array
             tools = await client.list_tools()
             tool = next(t for t in tools if t.name == "complex_tool")
-            expected_inner_schema = TypeAdapter(list[dict[str, int]]).json_schema()
+            expected_inner_schema = compress_schema(
+                TypeAdapter(list[dict[str, int]]).json_schema(), prune_titles=True
+            )
             expected_schema = {
                 "type": "object",
-                "properties": {"result": {**expected_inner_schema, "title": "Result"}},
+                "properties": {"result": expected_inner_schema},
                 "required": ["result"],
-                "title": "_WrappedResult",
                 "x-fastmcp-wrap-result": True,
             }
             assert tool.outputSchema == expected_schema
@@ -1129,7 +1132,9 @@ class TestToolOutputSchema:
             # List tools and verify schema is object type (not wrapped)
             tools = await client.list_tools()
             tool = next(t for t in tools if t.name == "dataclass_tool")
-            expected_schema = compress_schema(TypeAdapter(User).json_schema())
+            expected_schema = compress_schema(
+                TypeAdapter(User).json_schema(), prune_titles=True
+            )
             assert tool.outputSchema == expected_schema
             assert (
                 tool.outputSchema and "x-fastmcp-wrap-result" not in tool.outputSchema
@@ -1743,7 +1748,7 @@ class TestResourceTemplates:
 
         with pytest.raises(
             ValueError,
-            match="Required function arguments .* must be a subset of the URI parameters",
+            match="Required function arguments .* must be a subset of the URI path parameters",
         ):
 
             @mcp.resource("resource://{name}/data")
@@ -1770,7 +1775,7 @@ class TestResourceTemplates:
 
         with pytest.raises(
             ValueError,
-            match="Required function arguments .* must be a subset of the URI parameters",
+            match="Required function arguments .* must be a subset of the URI path parameters",
         ):
 
             @mcp.resource("resource://{org}/{repo}/data")
@@ -1864,6 +1869,29 @@ class TestResourceTemplates:
             result = await client.read_resource(AnyUrl("resource://test/data"))
             assert result[0].text == "Template resource: test/data"  # type: ignore[attr-defined]
 
+    async def test_template_with_query_params(self):
+        """Test RFC 6570 query parameters in resource templates."""
+        mcp = FastMCP()
+
+        @mcp.resource("data://{id}{?format,limit}")
+        def get_data(id: str, format: str = "json", limit: int = 10) -> str:
+            return f"id={id}, format={format}, limit={limit}"
+
+        async with Client(mcp) as client:
+            # No query params - uses defaults
+            result = await client.read_resource(AnyUrl("data://123"))
+            assert result[0].text == "id=123, format=json, limit=10"  # type: ignore[attr-defined]
+
+            # One query param
+            result = await client.read_resource(AnyUrl("data://123?format=xml"))
+            assert result[0].text == "id=123, format=xml, limit=10"  # type: ignore[attr-defined]
+
+            # Multiple query params
+            result = await client.read_resource(
+                AnyUrl("data://123?format=csv&limit=50")
+            )
+            assert result[0].text == "id=123, format=csv, limit=50"  # type: ignore[attr-defined]
+
     async def test_templates_match_in_order_of_definition(self):
         """
         If a wildcard template is defined first, it will take priority over another

tests/server/test_tool_annotations.py

@@ -47,7 +47,7 @@ async def test_tool_annotations_in_mcp_protocol():
         return message
 
     # Check via MCP protocol
-    mcp_tools = await mcp._mcp_list_tools()
+    mcp_tools = await mcp._list_tools_mcp()
     assert len(mcp_tools) == 1
     assert mcp_tools[0].annotations is not None
     assert mcp_tools[0].annotations.title == "Echo Tool"

tests/server/test_tool_transformation.py

@@ -29,14 +29,14 @@ async def test_transformed_tool_filtering():
         """Echo back the message provided."""
         return message
 
-    tools = list(await mcp._list_tools())
+    tools = list(await mcp._list_tools_middleware())
     assert len(tools) == 0
 
     mcp.add_tool_transformation(
         "echo", ToolTransformConfig(name="echo_transformed", tags={"enabled_tools"})
     )
 
-    tools = list(await mcp._list_tools())
+    tools = list(await mcp._list_tools_middleware())
     assert len(tools) == 1