Merge branch 'main' into claude/issue-2300-20251029-1512

.github/workflows/marvin.yml

@@ -74,5 +74,6 @@ jobs:
               "model": "claude-sonnet-4-5-20250929",
               "env": {
                 "GH_TOKEN": "${{ steps.marvin-token.outputs.token }}"
-              }
+              },
+              "customInstructions": "When you complete work on an issue: (1) You MUST create a pull request using the mcp__github__create_pull_request tool instead of posting a link, and (2) You MUST add the 'marvin-pr' label to the original issue using mcp__github__update_issue. Even if PR creation fails and you post a link instead, you MUST still add the 'marvin-pr' label. Follow the PR message guidelines in CLAUDE.md."
             }

docs/clients/auth/oauth.mdx

@@ -126,3 +126,7 @@ async with Client("https://fastmcp.cloud/mcp", auth=oauth) as client:
 ```
 
 You can use any `AsyncKeyValue`-compatible backend from the [key-value library](https://github.com/strawgate/py-key-value) including Redis, DynamoDB, and more. Wrap your storage in `FernetEncryptionWrapper` for encryption.
+
+<Note>
+When selecting a storage backend, review the [py-key-value documentation](https://github.com/strawgate/py-key-value) to understand the maturity level and limitations of your chosen backend. Some backends may be in preview or have constraints that affect production suitability.
+</Note>

docs/servers/auth/token-verification.mdx

@@ -210,6 +210,67 @@ Static token verification stores tokens as plain text and should never be used i
 </Warning>
 
 
+### Debug/Custom Token Verification
+
+The `DebugTokenVerifier` provides maximum flexibility for testing and special cases where standard token verification isn't applicable. It delegates validation to a user-provided callable, making it useful for prototyping, testing scenarios, or handling opaque tokens without introspection endpoints.
+
+```python
+from fastmcp import FastMCP
+from fastmcp.server.auth.providers.debug import DebugTokenVerifier
+
+# Accept all tokens (useful for rapid development)
+verifier = DebugTokenVerifier()
+
+mcp = FastMCP(name="Development Server", auth=verifier)
+```
+
+By default, `DebugTokenVerifier` accepts any non-empty token as valid. This eliminates authentication barriers during early development, allowing you to focus on core functionality before adding security.
+
+For more controlled testing, provide custom validation logic:
+
+```python
+from fastmcp.server.auth.providers.debug import DebugTokenVerifier
+
+# Synchronous validation - check token prefix
+verifier = DebugTokenVerifier(
+    validate=lambda token: token.startswith("dev-"),
+    client_id="development-client",
+    scopes=["read", "write"]
+)
+
+mcp = FastMCP(name="Development Server", auth=verifier)
+```
+
+The validation callable can also be async, enabling database lookups or external service calls:
+
+```python
+from fastmcp.server.auth.providers.debug import DebugTokenVerifier
+
+# Asynchronous validation - check against cache
+async def validate_token(token: str) -> bool:
+    # Check if token exists in Redis, database, etc.
+    return await redis.exists(f"valid_tokens:{token}")
+
+verifier = DebugTokenVerifier(
+    validate=validate_token,
+    client_id="api-client",
+    scopes=["api:access"]
+)
+
+mcp = FastMCP(name="Custom API", auth=verifier)
+```
+
+**Use Cases:**
+
+- **Testing**: Accept any token during integration tests without setting up token infrastructure
+- **Prototyping**: Quickly validate concepts without authentication complexity
+- **Opaque tokens without introspection**: When you have tokens from an IDP that provides no introspection endpoint, and you're willing to accept tokens without validation (validation happens later at the upstream service)
+- **Custom token formats**: Implement validation for non-standard token formats or legacy systems
+
+<Warning>
+`DebugTokenVerifier` bypasses standard security checks. Only use in controlled environments (development, testing) or when you fully understand the security implications. For production, use proper JWT or introspection-based verification.
+</Warning>
+
 ### Test Token Generation
 
 Test token generation helps when you need to test JWT verification without setting up complete identity infrastructure. FastMCP includes utilities for generating test key pairs and signed tokens.

docs/servers/icons.mdx

@@ -115,6 +115,7 @@ For small icons or when you want to embed the icon directly, use data URIs:
 
 ```python
 from mcp.types import Icon
+from fastmcp.utilities.types import Image
 
 # SVG icon as data URI
 svg_icon = Icon(
@@ -126,4 +127,13 @@ svg_icon = Icon(
 def my_tool() -> str:
     """A tool with an embedded SVG icon."""
     return "result"
+
+# Generating a data URI from a local image file.
+img = Image(path="./assets/brand/favicon.png")
+icon = Icon(src=img.to_data_uri())
+
+@mcp.tool(icons=[icon])
+def file_icon_tool() -> str:
+    """A tool with an icon generated from a local file."""
+    return "result"
 ```

docs/servers/storage-backends.mdx

@@ -143,6 +143,10 @@ The py-key-value-aio library includes additional implementations for various sto
 
 For configuration details on these backends, consult the [py-key-value-aio documentation](https://github.com/strawgate/py-key-value).
 
+<Warning>
+Before using these backends in production, review the [py-key-value documentation](https://github.com/strawgate/py-key-value) to understand the maturity level and limitations of your chosen backend. Some backends may be in preview or have specific constraints that make them unsuitable for production use.
+</Warning>
+
 ## Use Cases in FastMCP
 
 ### Server-Side OAuth Token Storage

pyproject.toml

@@ -16,6 +16,7 @@ dependencies = [
     "pydantic[email]>=2.11.7",
     "pyperclip>=1.9.0",
     "py-key-value-aio[disk,keyring,memory]>=0.2.8,<0.3.0",
+    "uvicorn>=0.35",
     "websockets>=15.0.1",
     "jsonschema-path>=0.3.4",
 ]
@@ -102,6 +103,11 @@ fallback-version = "0.0.0"
 [tool.pytest.ini_options]
 asyncio_mode = "auto"
 # filterwarnings = ["error::DeprecationWarning"]
+filterwarnings = [
+    # Suppress OAuth in-memory token storage warnings in tests
+    # Tests intentionally use ephemeral storage; this warning is for end users
+    "ignore:Using in-memory token storage:UserWarning",
+]
 timeout = 5
 env = [
     "FASTMCP_TEST_MODE=1",

src/fastmcp/resources/resource_manager.py

@@ -236,7 +236,7 @@ class ResourceManager:
         # Then check templates (local and mounted) only if not found in concrete resources
         templates = await self.get_resource_templates()
         for template_key in templates:
-            if match_uri_template(uri_str, template_key):
+            if match_uri_template(uri_str, template_key) is not None:
                 return True
 
         return False
@@ -262,7 +262,7 @@ class ResourceManager:
         templates = await self.get_resource_templates()
         for storage_key, template in templates.items():
             # Try to match against the storage key (which might be a custom key)
-            if params := match_uri_template(uri_str, storage_key):
+            if (params := match_uri_template(uri_str, storage_key)) is not None:
                 try:
                     return await template.create_resource(
                         uri_str,
@@ -318,7 +318,7 @@ class ResourceManager:
 
         # 1b. Check local templates if not found in concrete resources
         for key, template in self._templates.items():
-            if params := match_uri_template(uri_str, key):
+            if (params := match_uri_template(uri_str, key)) is not None:
                 try:
                     resource = await template.create_resource(uri_str, params=params)
                     return await resource.read()

src/fastmcp/server/auth/__init__.py

@@ -5,6 +5,7 @@ from .auth import (
     AccessToken,
     AuthProvider,
 )
+from .providers.debug import DebugTokenVerifier
 from .providers.jwt import JWTVerifier, StaticTokenVerifier
 from .oauth_proxy import OAuthProxy
 from .oidc_proxy import OIDCProxy
@@ -13,6 +14,7 @@ from .oidc_proxy import OIDCProxy
 __all__ = [
     "AccessToken",
     "AuthProvider",
+    "DebugTokenVerifier",
     "JWTVerifier",
     "OAuthProvider",
     "OAuthProxy",

src/fastmcp/server/auth/providers/debug.py

@@ -0,0 +1,114 @@
+"""Debug token verifier for testing and special cases.
+
+This module provides a flexible token verifier that delegates validation
+to a custom callable. Useful for testing, development, or scenarios where
+standard verification isn't possible (like opaque tokens without introspection).
+
+Example:
+    ```python
+    from fastmcp import FastMCP
+    from fastmcp.server.auth.providers.debug import DebugTokenVerifier
+
+    # Accept all tokens (default - useful for testing)
+    auth = DebugTokenVerifier()
+
+    # Custom sync validation logic
+    auth = DebugTokenVerifier(validate=lambda token: token.startswith("valid-"))
+
+    # Custom async validation logic
+    async def check_cache(token: str) -> bool:
+        return await redis.exists(f"token:{token}")
+
+    auth = DebugTokenVerifier(validate=check_cache)
+
+    mcp = FastMCP("My Server", auth=auth)
+    ```
+"""
+
+from __future__ import annotations
+
+import inspect
+from collections.abc import Awaitable, Callable
+
+from fastmcp.server.auth import TokenVerifier
+from fastmcp.server.auth.auth import AccessToken
+from fastmcp.utilities.logging import get_logger
+
+logger = get_logger(__name__)
+
+
+class DebugTokenVerifier(TokenVerifier):
+    """Token verifier with custom validation logic.
+
+    This verifier delegates token validation to a user-provided callable.
+    By default, it accepts all non-empty tokens (useful for testing).
+
+    Use cases:
+    - Testing: Accept any token without real verification
+    - Development: Custom validation logic for prototyping
+    - Opaque tokens: When you have tokens with no introspection endpoint
+
+    WARNING: This bypasses standard security checks. Only use in controlled
+    environments or when you understand the security implications.
+    """
+
+    def __init__(
+        self,
+        validate: Callable[[str], bool]
+        | Callable[[str], Awaitable[bool]] = lambda token: True,
+        client_id: str = "debug-client",
+        scopes: list[str] | None = None,
+        required_scopes: list[str] | None = None,
+    ):
+        """Initialize the debug token verifier.
+
+        Args:
+            validate: Callable that takes a token string and returns True if valid.
+                Can be sync or async. Default accepts all tokens.
+            client_id: Client ID to assign to validated tokens
+            scopes: Scopes to assign to validated tokens
+            required_scopes: Required scopes (inherited from TokenVerifier base class)
+        """
+        super().__init__(required_scopes=required_scopes)
+        self.validate = validate
+        self.client_id = client_id
+        self.scopes = scopes or []
+
+    async def verify_token(self, token: str) -> AccessToken | None:
+        """Verify token using custom validation logic.
+
+        Args:
+            token: The token string to validate
+
+        Returns:
+            AccessToken if validation succeeds, None otherwise
+        """
+        # Reject empty tokens
+        if not token or not token.strip():
+            logger.debug("Rejecting empty token")
+            return None
+
+        try:
+            # Call validation function and await if result is awaitable
+            result = self.validate(token)
+            if inspect.isawaitable(result):
+                is_valid = await result
+            else:
+                is_valid = result
+
+            if not is_valid:
+                logger.debug("Token validation failed: callable returned False")
+                return None
+
+            # Return valid AccessToken
+            return AccessToken(
+                token=token,
+                client_id=self.client_id,
+                scopes=self.scopes,
+                expires_at=None,  # No expiration
+                claims={"token": token},  # Store original token in claims
+            )
+
+        except Exception as e:
+            logger.debug("Token validation error: %s", e, exc_info=True)
+            return None

src/fastmcp/utilities/types.py

@@ -190,34 +190,33 @@ class Image:
         if path is not None and data is not None:
             raise ValueError("Only one of path or data can be provided")
 
-        self.path = Path(os.path.expandvars(str(path))).expanduser() if path else None
+        self.path = self._get_expanded_path(path)
         self.data = data
         self._format = format
         self._mime_type = self._get_mime_type()
         self.annotations = annotations
 
+    @staticmethod
+    def _get_expanded_path(path: str | Path | None) -> Path | None:
+        """Expand environment variables and user home in path."""
+        return Path(os.path.expandvars(str(path))).expanduser() if path else None
+
     def _get_mime_type(self) -> str:
         """Get MIME type from format or guess from file extension."""
         if self._format:
             return f"image/{self._format.lower()}"
 
         if self.path:
-            suffix = self.path.suffix.lower()
-            return {
-                ".png": "image/png",
-                ".jpg": "image/jpeg",
-                ".jpeg": "image/jpeg",
-                ".gif": "image/gif",
-                ".webp": "image/webp",
-            }.get(suffix, "application/octet-stream")
+            # Workaround for WEBP in Py3.10
+            mimetypes.add_type("image/webp", ".webp")
+            resp = mimetypes.guess_type(self.path, strict=False)
+            if resp and resp[0] is not None:
+                return resp[0]
+            return "application/octet-stream"
         return "image/png"  # default for raw binary data
 
-    def to_image_content(
-        self,
-        mime_type: str | None = None,
-        annotations: Annotations | None = None,
-    ) -> mcp.types.ImageContent:
-        """Convert to MCP ImageContent."""
+    def _get_data(self) -> str:
+        """Get raw image data as base64-encoded string."""
         if self.path:
             with open(self.path, "rb") as f:
                 data = base64.b64encode(f.read()).decode()
@@ -225,6 +224,15 @@ class Image:
             data = base64.b64encode(self.data).decode()
         else:
             raise ValueError("No image data available")
+        return data
+
+    def to_image_content(
+        self,
+        mime_type: str | None = None,
+        annotations: Annotations | None = None,
+    ) -> mcp.types.ImageContent:
+        """Convert to MCP ImageContent."""
+        data = self._get_data()
 
         return mcp.types.ImageContent(
             type="image",
@@ -233,6 +241,11 @@ class Image:
             annotations=annotations or self.annotations,
         )
 
+    def to_data_uri(self, mime_type: str | None = None) -> str:
+        """Get image as a data URI."""
+        data = self._get_data()
+        return f"data:{mime_type or self._mime_type};base64,{data}"
+
 
 class Audio:
     """Helper class for returning audio from tools."""

tests/client/test_sse.py

@@ -94,10 +94,13 @@ async def nested_sse_server():
     from starlette.applications import Starlette
     from starlette.routing import Mount
 
+    from fastmcp.server.http import create_sse_app
     from fastmcp.utilities.http import find_available_port
 
     server = create_test_server()
-    sse_app = server.sse_app(path="/mcp/sse/", message_path="/mcp/messages")
+    sse_app = create_sse_app(
+        server=server, message_path="/mcp/messages", sse_path="/mcp/sse/"
+    )
 
     # Nest the app under multiple mounts to test URL resolution
     inner = Starlette(routes=[Mount("/nest-inner", app=sse_app)])

tests/resources/test_resource_manager.py

@@ -567,6 +567,112 @@ class TestCustomResourceKeys:
             await manager.get_resource("greet://world")
 
 
+class TestQueryOnlyTemplates:
+    """Test resource templates with only query parameters (no path params)."""
+
+    async def test_template_with_only_query_params_no_query_string(self):
+        """Test that templates with only query params work without query string.
+
+        Regression test for bug where empty parameter dict {} was treated as falsy,
+        causing templates with only query parameters to fail when no query string
+        was provided in the URI.
+        """
+        manager = ResourceManager()
+
+        def get_config(format: str = "json") -> str:
+            return f"Config in {format} format"
+
+        template = ResourceTemplate.from_function(
+            fn=get_config,
+            uri_template="data://config{?format}",
+            name="config",
+        )
+        manager.add_template(template)
+
+        # Should work without query param (uses default)
+        resource = await manager.get_resource("data://config")
+        content = await resource.read()
+        assert content == "Config in json format"
+
+        # Should also work via read_resource
+        content = await manager.read_resource("data://config")
+        assert content == "Config in json format"
+
+    async def test_template_with_only_query_params_with_query_string(self):
+        """Test that templates with only query params work with query string."""
+        manager = ResourceManager()
+
+        def get_config(format: str = "json") -> str:
+            return f"Config in {format} format"
+
+        template = ResourceTemplate.from_function(
+            fn=get_config,
+            uri_template="data://config{?format}",
+            name="config",
+        )
+        manager.add_template(template)
+
+        # Should work with query param (overrides default)
+        resource = await manager.get_resource("data://config?format=xml")
+        content = await resource.read()
+        assert content == "Config in xml format"
+
+        # Should also work via read_resource
+        content = await manager.read_resource("data://config?format=xml")
+        assert content == "Config in xml format"
+
+    async def test_template_with_only_multiple_query_params(self):
+        """Test template with only multiple query parameters."""
+        manager = ResourceManager()
+
+        def get_data(format: str = "json", limit: int = 10) -> str:
+            return f"Data in {format} (limit: {limit})"
+
+        template = ResourceTemplate.from_function(
+            fn=get_data,
+            uri_template="data://items{?format,limit}",
+            name="items",
+        )
+        manager.add_template(template)
+
+        # No query params - use all defaults
+        content = await manager.read_resource("data://items")
+        assert content == "Data in json (limit: 10)"
+
+        # Partial query params
+        content = await manager.read_resource("data://items?format=xml")
+        assert content == "Data in xml (limit: 10)"
+
+        # All query params
+        content = await manager.read_resource("data://items?format=xml&limit=20")
+        assert content == "Data in xml (limit: 20)"
+
+    async def test_has_resource_with_query_only_template(self):
+        """Test that has_resource() works with query-only templates.
+
+        Regression test for bug where empty parameter dict {} was treated as falsy,
+        causing has_resource() to return False for query-only templates when no
+        query string was provided.
+        """
+        manager = ResourceManager()
+
+        def get_config(format: str = "json") -> str:
+            return f"Config in {format} format"
+
+        template = ResourceTemplate.from_function(
+            fn=get_config,
+            uri_template="data://config{?format}",
+            name="config",
+        )
+        manager.add_template(template)
+
+        # Should find resource without query param (uses default)
+        assert await manager.has_resource("data://config")
+
+        # Should also find resource with query param
+        assert await manager.has_resource("data://config?format=xml")
+
+
 class TestResourceErrorHandling:
     """Test error handling in the ResourceManager."""
 

tests/server/auth/test_debug_verifier.py

@@ -0,0 +1,169 @@
+"""Unit tests for DebugTokenVerifier."""
+
+import re
+
+from fastmcp.server.auth.providers.debug import DebugTokenVerifier
+
+
+class TestDebugTokenVerifier:
+    """Test DebugTokenVerifier initialization and validation."""
+
+    def test_init_defaults(self):
+        """Test initialization with default parameters."""
+        verifier = DebugTokenVerifier()
+
+        assert verifier.client_id == "debug-client"
+        assert verifier.scopes == []
+        assert verifier.required_scopes == []
+        assert callable(verifier.validate)
+
+    def test_init_custom_parameters(self):
+        """Test initialization with custom parameters."""
+        verifier = DebugTokenVerifier(
+            validate=lambda t: t.startswith("valid-"),
+            client_id="custom-client",
+            scopes=["read", "write"],
+            required_scopes=["admin"],
+        )
+
+        assert verifier.client_id == "custom-client"
+        assert verifier.scopes == ["read", "write"]
+        assert verifier.required_scopes == ["admin"]
+
+    async def test_verify_token_default_accepts_all(self):
+        """Test that default verifier accepts all non-empty tokens."""
+        verifier = DebugTokenVerifier()
+
+        result = await verifier.verify_token("any-token")
+
+        assert result is not None
+        assert result.token == "any-token"
+        assert result.client_id == "debug-client"
+        assert result.scopes == []
+        assert result.expires_at is None
+        assert result.claims == {"token": "any-token"}
+
+    async def test_verify_token_rejects_empty(self):
+        """Test that empty tokens are rejected even with default verifier."""
+        verifier = DebugTokenVerifier()
+
+        # Empty string
+        assert await verifier.verify_token("") is None
+
+        # Whitespace only
+        assert await verifier.verify_token("   ") is None
+
+    async def test_verify_token_sync_callable_success(self):
+        """Test token verification with custom sync callable that passes."""
+        verifier = DebugTokenVerifier(
+            validate=lambda t: t.startswith("valid-"),
+            client_id="test-client",
+            scopes=["read"],
+        )
+
+        result = await verifier.verify_token("valid-token-123")
+
+        assert result is not None
+        assert result.token == "valid-token-123"
+        assert result.client_id == "test-client"
+        assert result.scopes == ["read"]
+        assert result.expires_at is None
+        assert result.claims == {"token": "valid-token-123"}
+
+    async def test_verify_token_sync_callable_failure(self):
+        """Test token verification with custom sync callable that fails."""
+        verifier = DebugTokenVerifier(validate=lambda t: t.startswith("valid-"))
+
+        result = await verifier.verify_token("invalid-token")
+
+        assert result is None
+
+    async def test_verify_token_async_callable_success(self):
+        """Test token verification with custom async callable that passes."""
+
+        async def async_validator(token: str) -> bool:
+            # Simulate async operation (e.g., database check)
+            return token in {"token1", "token2", "token3"}
+
+        verifier = DebugTokenVerifier(
+            validate=async_validator,
+            client_id="async-client",
+            scopes=["admin"],
+        )
+
+        result = await verifier.verify_token("token2")
+
+        assert result is not None
+        assert result.token == "token2"
+        assert result.client_id == "async-client"
+        assert result.scopes == ["admin"]
+
+    async def test_verify_token_async_callable_failure(self):
+        """Test token verification with custom async callable that fails."""
+
+        async def async_validator(token: str) -> bool:
+            return token in {"token1", "token2", "token3"}
+
+        verifier = DebugTokenVerifier(validate=async_validator)
+
+        result = await verifier.verify_token("token99")
+
+        assert result is None
+
+    async def test_verify_token_callable_exception(self):
+        """Test that exceptions in validate callable are handled gracefully."""
+
+        def failing_validator(token: str) -> bool:
+            raise ValueError("Something went wrong")
+
+        verifier = DebugTokenVerifier(validate=failing_validator)
+
+        result = await verifier.verify_token("any-token")
+
+        assert result is None
+
+    async def test_verify_token_async_callable_exception(self):
+        """Test that exceptions in async validate callable are handled gracefully."""
+
+        async def failing_async_validator(token: str) -> bool:
+            raise ValueError("Async validation failed")
+
+        verifier = DebugTokenVerifier(validate=failing_async_validator)
+
+        result = await verifier.verify_token("any-token")
+
+        assert result is None
+
+    async def test_verify_token_whitelist_pattern(self):
+        """Test using verifier with a whitelist of allowed tokens."""
+        allowed_tokens = {"secret-token-1", "secret-token-2", "admin-token"}
+
+        verifier = DebugTokenVerifier(validate=lambda t: t in allowed_tokens)
+
+        # Allowed tokens
+        assert await verifier.verify_token("secret-token-1") is not None
+        assert await verifier.verify_token("admin-token") is not None
+
+        # Disallowed tokens
+        assert await verifier.verify_token("unknown-token") is None
+        assert await verifier.verify_token("hacker-token") is None
+
+    async def test_verify_token_pattern_matching(self):
+        """Test using verifier with regex-like pattern matching."""
+
+        pattern = re.compile(r"^[A-Z]{3}-\d{4}-[a-z]{2}$")
+
+        verifier = DebugTokenVerifier(
+            validate=lambda t: bool(pattern.match(t)),
+            client_id="pattern-client",
+        )
+
+        # Valid patterns
+        result = await verifier.verify_token("ABC-1234-xy")
+        assert result is not None
+        assert result.client_id == "pattern-client"
+
+        # Invalid patterns
+        assert await verifier.verify_token("abc-1234-xy") is None  # Wrong case
+        assert await verifier.verify_token("ABC-123-xy") is None  # Wrong digits
+        assert await verifier.verify_token("ABC-1234-xyz") is None  # Too many chars

tests/test_mcp_config.py

@@ -636,6 +636,7 @@ async def test_canonical_multi_client_with_transforms(tmp_path: Path):
         assert "test_1_transformed_add" not in tools_by_name
 
 
+@pytest.mark.flaky(retries=3)
 async def test_multi_client_transform_with_filtering(tmp_path: Path):
     """
     Tests that tag-based filtering works when using a transforming MCPConfig.

tests/utilities/test_types.py

@@ -177,22 +177,23 @@ class TestImage:
         ):
             Image(path="test.png", data=b"test")
 
-    def test_get_mime_type_from_path(self, tmp_path):
+    @pytest.mark.parametrize(
+        "extension,mime_type",
+        [
+            (".png", "image/png"),
+            (".jpg", "image/jpeg"),
+            (".jpeg", "image/jpeg"),
+            (".gif", "image/gif"),
+            (".webp", "image/webp"),
+            (".unknown", "application/octet-stream"),
+        ],
+    )
+    def test_get_mime_type_from_path(self, tmp_path, extension, mime_type):
         """Test MIME type detection from file extension."""
-        extensions = {
-            ".png": "image/png",
-            ".jpg": "image/jpeg",
-            ".jpeg": "image/jpeg",
-            ".gif": "image/gif",
-            ".webp": "image/webp",
-            ".unknown": "application/octet-stream",
-        }
-
-        for ext, mime in extensions.items():
-            path = tmp_path / f"test{ext}"
-            path.write_bytes(b"fake image data")
-            img = Image(path=path)
-            assert img._mime_type == mime
+        path = tmp_path / f"test{extension}"
+        path.write_bytes(b"fake image data")
+        img = Image(path=path)
+        assert img._mime_type == mime_type
 
     def test_to_image_content(self, tmp_path, monkeypatch):
         """Test conversion to ImageContent."""
@@ -227,6 +228,27 @@ class TestImage:
         with pytest.raises(ValueError, match="No image data available"):
             img.to_image_content()
 
+    @pytest.mark.parametrize(
+        "mime_type,fname,expected_mime",
+        [
+            (None, "test.png", "image/png"),
+            ("image/jpeg", "test.unknown", "image/jpeg"),
+        ],
+    )
+    def test_to_data_uri(self, tmp_path, mime_type, fname, expected_mime):
+        """Test conversion to data URI."""
+        img_path = tmp_path / fname
+        test_data = b"fake image data"
+        img_path.write_bytes(test_data)
+
+        img = Image(path=img_path)
+        data_uri = img.to_data_uri(mime_type=mime_type)
+
+        expected_data_uri = (
+            f"data:{expected_mime};base64,{base64.b64encode(test_data).decode()}"
+        )
+        assert data_uri == expected_data_uri
+
 
 class TestAudio:
     def test_audio_initialization_with_path(self):

uv.lock

@@ -593,6 +593,7 @@ dependencies = [
     { name = "pyperclip" },
     { name = "python-dotenv" },
     { name = "rich" },
+    { name = "uvicorn" },
     { name = "websockets" },
 ]
 
@@ -644,6 +645,7 @@ requires-dist = [
     { name = "pyperclip", specifier = ">=1.9.0" },
     { name = "python-dotenv", specifier = ">=1.1.0" },
     { name = "rich", specifier = ">=13.9.4" },
+    { name = "uvicorn", specifier = ">=0.35" },
     { name = "websockets", specifier = ">=15.0.1" },
 ]
 provides-extras = ["openai"]