Max Isbey
1b74b06753
Cut comment and docstring volume roughly in half across src, tests, examples, and docs_src: removed comments that restate the adjacent code, leftover development narration, section banners, and self-evident Args/Returns blocks, and compressed the remaining docstrings to a Google-style summary line plus only the detail that earns its place. Kept (and tightened) the load-bearing content: Raises sections, deprecation and version-availability notes, spec/RFC/issue references, why-comments for non-obvious decisions, and all coverage pragmas. The generated mcp_types.v* wire modules are untouched.
102 lines
3.3 KiB
Python
102 lines
3.3 KiB
Python
from typing import Literal
|
|
|
|
import mcp_types as types
|
|
import pytest
|
|
from mcp_types import (
|
|
LoggingMessageNotificationParams,
|
|
TextContent,
|
|
)
|
|
|
|
from mcp import Client
|
|
from mcp.server.mcpserver import Context, MCPServer
|
|
from mcp.shared.session import RequestResponder
|
|
|
|
|
|
class LoggingCollector:
|
|
def __init__(self):
|
|
self.log_messages: list[LoggingMessageNotificationParams] = []
|
|
|
|
async def __call__(self, params: LoggingMessageNotificationParams) -> None:
|
|
self.log_messages.append(params)
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_logging_callback():
|
|
server = MCPServer("test")
|
|
logging_collector = LoggingCollector()
|
|
|
|
@server.tool("test_tool")
|
|
async def test_tool() -> bool:
|
|
return True
|
|
|
|
@server.tool("test_tool_with_log")
|
|
async def test_tool_with_log(
|
|
message: str, level: Literal["debug", "info", "warning", "error"], logger: str, ctx: Context
|
|
) -> bool:
|
|
"""Send a log notification to the client."""
|
|
await ctx.log(level=level, data=message, logger_name=logger) # pyright: ignore[reportDeprecated]
|
|
return True
|
|
|
|
@server.tool("test_tool_with_log_dict")
|
|
async def test_tool_with_log_dict(
|
|
level: Literal["debug", "info", "warning", "error"],
|
|
logger: str,
|
|
ctx: Context,
|
|
) -> bool:
|
|
"""Send a log notification with a dict payload."""
|
|
await ctx.log( # pyright: ignore[reportDeprecated]
|
|
level=level,
|
|
data={"message": "Test log message", "extra_string": "example", "extra_dict": {"a": 1, "b": 2, "c": 3}},
|
|
logger_name=logger,
|
|
)
|
|
return True
|
|
|
|
async def message_handler(
|
|
message: RequestResponder[types.ServerRequest, types.ClientResult] | types.ServerNotification | Exception,
|
|
) -> None:
|
|
if isinstance(message, Exception): # pragma: no cover
|
|
raise message
|
|
|
|
async with Client(
|
|
server,
|
|
logging_callback=logging_collector,
|
|
message_handler=message_handler,
|
|
mode="legacy",
|
|
) as client:
|
|
result = await client.call_tool("test_tool", {})
|
|
assert result.is_error is False
|
|
assert isinstance(result.content[0], TextContent)
|
|
assert result.content[0].text == "true"
|
|
|
|
log_result = await client.call_tool(
|
|
"test_tool_with_log",
|
|
{
|
|
"message": "Test log message",
|
|
"level": "info",
|
|
"logger": "test_logger",
|
|
},
|
|
)
|
|
log_result_with_dict = await client.call_tool(
|
|
"test_tool_with_log_dict",
|
|
{
|
|
"level": "info",
|
|
"logger": "test_logger",
|
|
},
|
|
)
|
|
assert log_result.is_error is False
|
|
assert log_result_with_dict.is_error is False
|
|
assert len(logging_collector.log_messages) == 2
|
|
log = logging_collector.log_messages[0]
|
|
assert log.level == "info"
|
|
assert log.logger == "test_logger"
|
|
assert log.data == "Test log message"
|
|
|
|
log_with_dict = logging_collector.log_messages[1]
|
|
assert log_with_dict.level == "info"
|
|
assert log_with_dict.logger == "test_logger"
|
|
assert log_with_dict.data == {
|
|
"message": "Test log message",
|
|
"extra_string": "example",
|
|
"extra_dict": {"a": 1, "b": 2, "c": 3},
|
|
}
|