Add log level support for stdio and HTTP transports (#1840)

AGENTS.md

@@ -20,24 +20,24 @@ uv run pytest                        # Run full test suite
 
 ## Repository Structure
 
-| Path             | Purpose                                                |
-| ---------------- | ------------------------------------------------------ |
-| `src/fastmcp/`   | Library source code (Python ≥ 3.10)                   |
-| `├─server/`    | Server implementation, `FastMCP`, auth, networking    |
-| `│  ├─auth/`   | Authentication providers (Bearer, JWT, WorkOS)        |
-| `│  └─middleware/` | Error handling, logging, rate limiting             |
-| `├─client/`    | High-level client SDK + transports                    |
-| `│  └─auth/`   | Client authentication (Bearer, OAuth)                 |
-| `├─tools/`     | Tool implementations + `ToolManager`                  |
-| `├─resources/` | Resources, templates + `ResourceManager`              |
-| `├─prompts/`   | Prompt templates + `PromptManager`                     |
-| `├─cli/`       | FastMCP CLI commands (`run`, `dev`, `install`)         |
-| `├─contrib/`   | Community contributions (bulk caller, mixins)         |
-| `├─experimental/` | Experimental features (new OpenAPI parser)         |
-| `└─utilities/` | Shared utilities (logging, JSON schema, HTTP)         |
-| `tests/`         | Comprehensive pytest suite with markers               |
-| `docs/`          | Mintlify documentation (published to gofastmcp.com)   |
-| `examples/`      | Runnable demo servers (echo, smart_home, atproto)     |
+| Path               | Purpose                                             |
+| ------------------ | --------------------------------------------------- |
+| `src/fastmcp/`     | Library source code (Python ≥ 3.10)                 |
+| `├─server/`        | Server implementation, `FastMCP`, auth, networking  |
+| `│  ├─auth/`       | Authentication providers (Bearer, JWT, WorkOS)      |
+| `│  └─middleware/` | Error handling, logging, rate limiting              |
+| `├─client/`        | High-level client SDK + transports                  |
+| `│  └─auth/`       | Client authentication (Bearer, OAuth)               |
+| `├─tools/`         | Tool implementations + `ToolManager`                |
+| `├─resources/`     | Resources, templates + `ResourceManager`            |
+| `├─prompts/`       | Prompt templates + `PromptManager`                  |
+| `├─cli/`           | FastMCP CLI commands (`run`, `dev`, `install`)      |
+| `├─contrib/`       | Community contributions (bulk caller, mixins)       |
+| `├─experimental/`  | Experimental features (new OpenAPI parser)          |
+| `└─utilities/`     | Shared utilities (logging, JSON schema, HTTP)       |
+| `tests/`           | Comprehensive pytest suite with markers             |
+| `docs/`            | Mintlify documentation (published to gofastmcp.com) |
+| `examples/`        | Runnable demo servers (echo, smart_home, atproto)   |
 
 ## Core MCP Objects
 
@@ -64,7 +64,7 @@ When modifying MCP functionality, changes typically need to be applied across al
 - Every test: atomic, self-contained, single functionality
 - Use parameterization for multiple examples of same functionality
 - Use separate tests for different functionality pieces
-- Put imports at the top of the file, not in the test body
+- **ALWAYS** Put imports at the top of the file, not in the test body
 - **NEVER** add `@pytest.mark.asyncio` to tests - `asyncio_mode = "auto"` is set globally
 - **ALWAYS** run pytest after significant changes
 

src/fastmcp/cli/run.py

@@ -184,8 +184,8 @@ async def run_command(
         kwargs["port"] = port
     if path:
         kwargs["path"] = path
-    # Note: log_level is not currently supported by run_async
-    # TODO: Add log_level support to server.run_async
+    if log_level:
+        kwargs["log_level"] = log_level
 
     if not show_banner:
         kwargs["show_banner"] = False

src/fastmcp/server/server.py

@@ -8,11 +8,7 @@ import re
 import secrets
 import warnings
 from collections.abc import AsyncIterator, Awaitable, Callable
-from contextlib import (
-    AbstractAsyncContextManager,
-    AsyncExitStack,
-    asynccontextmanager,
-)
+from contextlib import AbstractAsyncContextManager, AsyncExitStack, asynccontextmanager
 from dataclasses import dataclass
 from functools import partial
 from pathlib import Path
@@ -65,7 +61,7 @@ from fastmcp.tools.tool import FunctionTool, Tool, ToolResult
 from fastmcp.tools.tool_transform import ToolTransformConfig
 from fastmcp.utilities.cli import log_server_banner
 from fastmcp.utilities.components import FastMCPComponent
-from fastmcp.utilities.logging import get_logger
+from fastmcp.utilities.logging import get_logger, temporary_log_level
 from fastmcp.utilities.types import NotSet, NotSetT
 
 if TYPE_CHECKING:
@@ -1485,9 +1481,15 @@ class FastMCP(Generic[LifespanResultT]):
             meta=meta,
         )
 
-    async def run_stdio_async(self, show_banner: bool = True) -> None:
-        """Run the server using stdio transport."""
+    async def run_stdio_async(
+        self, show_banner: bool = True, log_level: str | None = None
+    ) -> None:
+        """Run the server using stdio transport.
 
+        Args:
+            show_banner: Whether to display the server banner
+            log_level: Log level for the server
+        """
         # Display server banner
         if show_banner:
             log_server_banner(
@@ -1495,15 +1497,16 @@ class FastMCP(Generic[LifespanResultT]):
                 transport="stdio",
             )
 
-        async with stdio_server() as (read_stream, write_stream):
-            logger.info(f"Starting MCP server {self.name!r} with transport 'stdio'")
-            await self._mcp_server.run(
-                read_stream,
-                write_stream,
-                self._mcp_server.create_initialization_options(
-                    NotificationOptions(tools_changed=True)
-                ),
-            )
+        with temporary_log_level(log_level):
+            async with stdio_server() as (read_stream, write_stream):
+                logger.info(f"Starting MCP server {self.name!r} with transport 'stdio'")
+                await self._mcp_server.run(
+                    read_stream,
+                    write_stream,
+                    self._mcp_server.create_initialization_options(
+                        NotificationOptions(tools_changed=True)
+                    ),
+                )
 
     async def run_http_async(
         self,
@@ -1529,7 +1532,6 @@ class FastMCP(Generic[LifespanResultT]):
             middleware: A list of middleware to apply to the app
             stateless_http: Whether to use stateless HTTP (defaults to settings.stateless_http)
         """
-
         host = host or self._deprecated_settings.host
         port = port or self._deprecated_settings.port
         default_log_level_to_use = (
@@ -1570,14 +1572,15 @@ class FastMCP(Generic[LifespanResultT]):
         if "log_config" not in config_kwargs and "log_level" not in config_kwargs:
             config_kwargs["log_level"] = default_log_level_to_use
 
-        config = uvicorn.Config(app, host=host, port=port, **config_kwargs)
-        server = uvicorn.Server(config)
-        path = app.state.path.lstrip("/")  # type: ignore
-        logger.info(
-            f"Starting MCP server {self.name!r} with transport {transport!r} on http://{host}:{port}/{path}"
-        )
+        with temporary_log_level(log_level):
+            config = uvicorn.Config(app, host=host, port=port, **config_kwargs)
+            server = uvicorn.Server(config)
+            path = app.state.path.lstrip("/")  # type: ignore
+            logger.info(
+                f"Starting MCP server {self.name!r} with transport {transport!r} on http://{host}:{port}/{path}"
+            )
 
-        await server.serve()
+            await server.serve()
 
     async def run_sse_async(
         self,
@@ -2126,10 +2129,8 @@ class FastMCP(Generic[LifespanResultT]):
             # - Connected clients: reuse existing session for all requests
             # - Disconnected clients: create fresh sessions per request for isolation
             if client.is_connected():
-                from fastmcp.utilities.logging import get_logger
-
-                logger = get_logger(__name__)
-                logger.info(
+                _proxy_logger = get_logger(__name__)
+                _proxy_logger.info(
                     "Proxy detected connected client - reusing existing session for all requests. "
                     "This may cause context mixing in concurrent scenarios."
                 )

src/fastmcp/utilities/logging.py

@@ -1,11 +1,14 @@
 """Logging utilities for FastMCP."""
 
+import contextlib
 import logging
-from typing import Any, Literal
+from typing import Any, Literal, cast
 
 from rich.console import Console
 from rich.logging import RichHandler
 
+import fastmcp
+
 
 def get_logger(name: str) -> logging.Logger:
     """Get a logger nested under FastMCP namespace.
@@ -22,7 +25,7 @@ def get_logger(name: str) -> logging.Logger:
 def configure_logging(
     level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] | int = "INFO",
     logger: logging.Logger | None = None,
-    enable_rich_tracebacks: bool = True,
+    enable_rich_tracebacks: bool | None = None,
     **rich_kwargs: Any,
 ) -> None:
     """
@@ -33,6 +36,13 @@ def configure_logging(
         level: the log level to use
         rich_kwargs: the parameters to use for creating RichHandler
     """
+    # Check if logging is disabled in settings
+    if not fastmcp.settings.log_enabled:
+        return
+
+    # Use settings default if not specified
+    if enable_rich_tracebacks is None:
+        enable_rich_tracebacks = fastmcp.settings.enable_rich_tracebacks
 
     if logger is None:
         logger = logging.getLogger("FastMCP")
@@ -56,3 +66,55 @@ def configure_logging(
 
     # Don't propagate to the root logger
     logger.propagate = False
+
+
+@contextlib.contextmanager
+def temporary_log_level(
+    level: str | None,
+    logger: logging.Logger | None = None,
+    enable_rich_tracebacks: bool | None = None,
+    **rich_kwargs: Any,
+):
+    """Context manager to temporarily set log level and restore it afterwards.
+
+    Args:
+        level: The temporary log level to set (e.g., "DEBUG", "INFO")
+        logger: Optional logger to configure (defaults to FastMCP logger)
+        enable_rich_tracebacks: Whether to enable rich tracebacks
+        **rich_kwargs: Additional parameters for RichHandler
+
+    Usage:
+        with temporary_log_level("DEBUG"):
+            # Code that runs with DEBUG logging
+            pass
+        # Original log level is restored here
+    """
+    if level:
+        # Get the original log level from settings
+        original_level = fastmcp.settings.log_level
+
+        # Configure with new level
+        # Cast to proper type for type checker
+        log_level_literal = cast(
+            Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
+            level.upper(),
+        )
+        configure_logging(
+            level=log_level_literal,
+            logger=logger,
+            enable_rich_tracebacks=enable_rich_tracebacks,
+            **rich_kwargs,
+        )
+        try:
+            yield
+        finally:
+            # Restore original configuration using configure_logging
+            # This will respect the log_enabled setting
+            configure_logging(
+                level=original_level,
+                logger=logger,
+                enable_rich_tracebacks=enable_rich_tracebacks,
+                **rich_kwargs,
+            )
+    else:
+        yield

src/fastmcp/utilities/mcp_server_config/v1/mcp_server_config.py

@@ -403,7 +403,8 @@ class MCPServerConfig(BaseModel):
                 run_args["port"] = self.deployment.port
             if self.deployment.path:
                 run_args["path"] = self.deployment.path
-            # Note: log_level not currently supported by run_async
+            if self.deployment.log_level:
+                run_args["log_level"] = self.deployment.log_level
 
         # Override with any provided kwargs
         run_args.update(kwargs)

tests/cli/test_run_config.py

@@ -206,6 +206,47 @@ def test_load_config_with_server_args(tmp_path):
     assert config.deployment.args == ["--debug", "--config", "custom.json"]
 
 
+def test_load_config_with_log_level(tmp_path):
+    """Test configuration with log_level setting."""
+    config_data = {
+        "source": {"path": "server.py"},
+        "deployment": {"log_level": "DEBUG"},
+    }
+
+    config_file = tmp_path / "fastmcp.json"
+    config_file.write_text(json.dumps(config_data))
+
+    # Create server file
+    server_file = tmp_path / "server.py"
+    server_file.write_text("# Server")
+
+    config = load_mcp_server_config(config_file)
+
+    assert config.deployment.log_level == "DEBUG"
+
+
+def test_load_config_with_various_log_levels(tmp_path):
+    """Test that all valid log levels are accepted."""
+    valid_levels = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
+
+    for level in valid_levels:
+        config_data = {
+            "source": {"path": "server.py"},
+            "deployment": {"log_level": level},
+        }
+
+        config_file = tmp_path / f"fastmcp_{level}.json"
+        config_file.write_text(json.dumps(config_data))
+
+        # Create server file
+        server_file = tmp_path / "server.py"
+        server_file.write_text("# Server")
+
+        config = load_mcp_server_config(config_file)
+
+        assert config.deployment.log_level == level
+
+
 def test_config_subset_independence(tmp_path):
     """Test that config subsets can be used independently."""
     config_data = {

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"