Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 16 additions & 4 deletions src/mcp/cli/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,7 @@
import subprocess
import sys
from pathlib import Path
from typing import Annotated, Any
from typing import Annotated, Any, Literal

from mcp.server import MCPServer
from mcp.server import Server as LowLevelServer
Expand All @@ -19,7 +19,7 @@

try:
from mcp.cli import claude
from mcp.server.mcpserver.utilities.logging import get_logger
from mcp.server.mcpserver.utilities.logging import configure_logging, get_logger
except ImportError: # pragma: no cover
print("Error: mcp.server is not installed or not in PYTHONPATH")
sys.exit(1)
Expand All @@ -31,6 +31,8 @@

logger = get_logger("cli")

LogLevel = Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]

app = typer.Typer(
name="mcp",
help="MCP development tools",
Expand Down Expand Up @@ -116,6 +118,14 @@ def _parse_file_path(file_spec: str) -> tuple[Path, str | None]:
return file_path, server_object


def _get_server_log_level(server: Any) -> LogLevel:
settings = getattr(server, "settings", None)
log_level = getattr(settings, "log_level", "INFO")
if log_level in ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"):
return log_level
return "INFO"


def _import_server(file: Path, server_object: str | None = None): # pragma: no cover
"""Import an MCP server from a file.

Expand Down Expand Up @@ -314,7 +324,7 @@ def run(
help="Transport protocol to use (stdio or sse)",
),
] = None,
) -> None: # pragma: no cover
) -> None:
"""Run an MCP server.

The server can be specified in two ways:
Expand All @@ -340,14 +350,16 @@ def run(
# Import and get server object
server = _import_server(file, server_object)

configure_logging(_get_server_log_level(server))

# Run the server
kwargs = {}
if transport:
kwargs["transport"] = transport

server.run(**kwargs)

except Exception:
except Exception: # pragma: no cover
logger.exception(
"Failed to run server",
extra={
Expand Down
5 changes: 1 addition & 4 deletions src/mcp/server/mcpserver/server.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,7 @@
from mcp.server.mcpserver.resources import FunctionResource, Resource, ResourceManager
from mcp.server.mcpserver.tools import Tool, ToolManager
from mcp.server.mcpserver.utilities.context_injection import find_context_parameter
from mcp.server.mcpserver.utilities.logging import configure_logging, get_logger
from mcp.server.mcpserver.utilities.logging import get_logger
from mcp.server.sse import SseServerTransport
from mcp.server.stdio import stdio_server
from mcp.server.streamable_http import EventStore
Expand Down Expand Up @@ -203,9 +203,6 @@ def __init__(
self._token_verifier = ProviderTokenVerifier(auth_server_provider)
self._custom_starlette_routes: list[Route] = []

# Configure logging
configure_logging(self.settings.log_level)

@property
def name(self) -> str:
return self._lowlevel_server.name
Expand Down
65 changes: 64 additions & 1 deletion tests/cli/test_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,13 @@

import pytest

from mcp.cli.cli import _build_uv_command, _get_npx_command, _parse_file_path # type: ignore[reportPrivateUsage]
from mcp.cli.cli import ( # type: ignore[reportPrivateUsage]
_build_uv_command,
_get_npx_command,
_get_server_log_level,
_parse_file_path,
run,
)


@pytest.mark.parametrize(
Expand Down Expand Up @@ -38,6 +44,63 @@ def test_parse_file_exit_on_dir(tmp_path: Path):
_parse_file_path(str(dir_path))


@pytest.mark.parametrize(
("transport", "expected_kwargs"),
[
(None, {}),
("stdio", {"transport": "stdio"}),
],
)
def test_run_configures_logging_at_cli_entrypoint(
monkeypatch: pytest.MonkeyPatch,
tmp_path: Path,
transport: str | None,
expected_kwargs: dict[str, str],
):
"""The CLI owns process logging setup; importing or constructing the server does not."""
file = tmp_path / "server.py"
file.write_text("server = object()")
calls: list[tuple[str, Any]] = []

class Settings:
log_level = "DEBUG"

class ImportedServer:
settings = Settings()

def run(self, **kwargs: Any) -> None:
calls.append(("run", kwargs))

def fake_import_server(*_args: Any) -> ImportedServer:
return ImportedServer()

def fake_configure_logging(level: str) -> None:
calls.append(("logging", level))

monkeypatch.setattr("mcp.cli.cli._import_server", fake_import_server)
monkeypatch.setattr(
"mcp.cli.cli.configure_logging",
fake_configure_logging,
)

run(f"{file}:server", transport=transport)

assert calls == [
("logging", "DEBUG"),
("run", expected_kwargs),
]


def test_get_server_log_level_defaults_invalid_setting():
class Settings:
log_level = "VERBOSE"

class Server:
settings = Settings()

assert _get_server_log_level(Server()) == "INFO"


def test_build_uv_command_minimal():
"""Should emit core command when no extras specified."""
cmd = _build_uv_command("foo.py")
Expand Down
9 changes: 9 additions & 0 deletions tests/server/mcpserver/test_server.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import base64
import logging
from pathlib import Path
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
Expand Down Expand Up @@ -46,6 +47,14 @@


class TestServer:
def test_create_server_does_not_configure_application_logging(self, monkeypatch: pytest.MonkeyPatch):
basic_config = MagicMock()
monkeypatch.setattr(logging, "basicConfig", basic_config)

MCPServer("test")

basic_config.assert_not_called()

async def test_create_server(self):
mcp = MCPServer(
title="MCPServer Server",
Expand Down
34 changes: 34 additions & 0 deletions tests/server/mcpserver/utilities/test_logging.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,34 @@
import logging
from typing import Any

import pytest

from mcp.server.mcpserver.utilities.logging import configure_logging, get_logger


def test_get_logger_returns_named_logger():
logger = get_logger("mcp.test")

assert logger is logging.getLogger("mcp.test")


def test_configure_logging_uses_rich_handler(monkeypatch: pytest.MonkeyPatch):
calls: list[dict[str, Any]] = []

def fake_basic_config(**kwargs: Any) -> None:
calls.append(kwargs)

monkeypatch.setattr(logging, "basicConfig", fake_basic_config)

configure_logging("WARNING")

handlers = calls[0]["handlers"]
assert calls == [
{
"level": "WARNING",
"format": "%(message)s",
"handlers": handlers,
}
]
assert len(handlers) == 1
assert isinstance(handlers[0], logging.Handler)
Loading