forked from modelcontextprotocol/python-sdk
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathtest_logging.py
More file actions
192 lines (147 loc) · 8.27 KB
/
Copy pathtest_logging.py
File metadata and controls
192 lines (147 loc) · 8.27 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
"""Logging interactions against the low-level Server, driven through the public Client API.
Notification ordering: await-free callbacks finish in arrival order, and passing
``related_request_id`` keeps each notification on the originating request's POST stream over
streamable HTTP, so plain-list collection is deterministic on every transport leg.
"""
import mcp_types as types
import pytest
from inline_snapshot import snapshot
from mcp_types import (
INVALID_PARAMS,
LOG_LEVEL_META_KEY,
CallToolResult,
EmptyResult,
LoggingMessageNotificationParams,
TextContent,
)
from mcp import MCPError
from mcp.server import Server, ServerRequestContext
from tests._stamp import Unstamp
from tests.interaction._connect import Connect
from tests.interaction._requirements import requirement
pytestmark = pytest.mark.anyio
ALL_LEVELS: tuple[types.LoggingLevel, ...] = (
"debug",
"info",
"notice",
"warning",
"error",
"critical",
"alert",
"emergency",
)
@requirement("logging:set-level")
async def test_set_logging_level_reaches_handler(connect: Connect) -> None:
"""The level requested by the client is delivered to the server's handler verbatim."""
async def set_logging_level(ctx: ServerRequestContext, params: types.SetLevelRequestParams) -> EmptyResult:
assert params.level == "warning"
return EmptyResult()
server = Server("logger", on_set_logging_level=set_logging_level) # pyright: ignore[reportDeprecated]
async with connect(server) as client:
result = await client.set_logging_level("warning") # pyright: ignore[reportDeprecated]
assert result == snapshot(EmptyResult())
@requirement("logging:message:fields")
@requirement("tools:call:logging-mid-execution")
async def test_log_messages_reach_logging_callback_in_order(connect: Connect, unstamped: Unstamp) -> None:
"""Log messages sent during a tool call arrive at the logging callback, in order, before the call returns.
The two messages pin the full notification shape: severity, optional logger name, and both
string and structured data payloads.
"""
received: list[LoggingMessageNotificationParams] = []
async def collect(params: LoggingMessageNotificationParams) -> None:
received.append(params)
async def list_tools(
ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
) -> types.ListToolsResult:
return types.ListToolsResult(tools=[types.Tool(name="chatty", input_schema={"type": "object"})])
async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult:
assert params.name == "chatty"
await ctx.session.send_log_message( # pyright: ignore[reportDeprecated]
level="info", data="starting up", logger="app.lifecycle", related_request_id=ctx.request_id
)
await ctx.session.send_log_message( # pyright: ignore[reportDeprecated]
level="error", data={"code": 502, "retryable": True}, related_request_id=ctx.request_id
)
return CallToolResult(content=[TextContent(text="done")])
async def set_logging_level(ctx: ServerRequestContext, params: types.SetLevelRequestParams) -> EmptyResult:
"""Registered so the logging capability is advertised; the client never sets a level."""
raise NotImplementedError
server = Server( # pyright: ignore[reportDeprecated]
"logger", on_list_tools=list_tools, on_call_tool=call_tool, on_set_logging_level=set_logging_level
)
async with connect(server, logging_callback=collect, log_level="debug") as client:
result = await client.call_tool("chatty", {})
assert unstamped(result) == snapshot(CallToolResult(content=[TextContent(text="done")]))
assert received == snapshot(
[
LoggingMessageNotificationParams(level="info", logger="app.lifecycle", data="starting up"),
LoggingMessageNotificationParams(level="error", data={"code": 502, "retryable": True}),
]
)
@requirement("logging:message:all-levels")
async def test_log_messages_at_every_severity_level(connect: Connect) -> None:
"""Each of the eight RFC 5424 severity levels is deliverable as a log message notification."""
received: list[LoggingMessageNotificationParams] = []
async def collect(params: LoggingMessageNotificationParams) -> None:
received.append(params)
async def list_tools(
ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
) -> types.ListToolsResult:
return types.ListToolsResult(tools=[types.Tool(name="siren", input_schema={"type": "object"})])
async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult:
assert params.name == "siren"
for level in ALL_LEVELS:
await ctx.session.send_log_message( # pyright: ignore[reportDeprecated]
level=level, data=f"a {level} message", related_request_id=ctx.request_id
)
return CallToolResult(content=[TextContent(text="logged")])
async def set_logging_level(ctx: ServerRequestContext, params: types.SetLevelRequestParams) -> EmptyResult:
"""Registered so the logging capability is advertised; the client never sets a level."""
raise NotImplementedError
server = Server( # pyright: ignore[reportDeprecated]
"logger", on_list_tools=list_tools, on_call_tool=call_tool, on_set_logging_level=set_logging_level
)
async with connect(server, logging_callback=collect, log_level="debug") as client:
await client.call_tool("siren", {})
assert [params.level for params in received] == list(ALL_LEVELS)
def _siren_server() -> Server:
"""A server whose `siren` tool logs one message at each of the eight severity levels.
The messages are sent without `related_request_id`: on 2026-07-28+ log delivery is
request-scoped by construction, so they still ride the requesting stream on every leg.
"""
async def list_tools(
ctx: ServerRequestContext, params: types.PaginatedRequestParams | None
) -> types.ListToolsResult:
return types.ListToolsResult(tools=[types.Tool(name="siren", input_schema={"type": "object"})])
async def call_tool(ctx: ServerRequestContext, params: types.CallToolRequestParams) -> CallToolResult:
assert params.name == "siren"
for level in ALL_LEVELS:
await ctx.session.send_log_message(level=level, data=f"a {level} message") # pyright: ignore[reportDeprecated]
return CallToolResult(content=[TextContent(text="logged")])
return Server("logger", on_list_tools=list_tools, on_call_tool=call_tool)
@requirement("logging:per-request:opt-in")
@requirement("logging:per-request:threshold")
async def test_log_delivery_follows_the_per_request_log_level(connect: Connect) -> None:
"""Without io.modelcontextprotocol/logLevel in _meta a request gets no log messages;
with it, only entries at or above the requested level are delivered, in order.
The handler emits at every severity in both phases: the un-opted request receives
nothing (the log calls are dropped, not delivered on some other stream), and the request
opting in at `warning` receives warning and above.
"""
received: list[types.LoggingLevel] = []
async def collect(params: LoggingMessageNotificationParams) -> None:
received.append(params.level)
async with connect(_siren_server(), logging_callback=collect) as client:
result = await client.call_tool("siren", {})
assert isinstance(result.content[0], TextContent) and result.content[0].text == "logged"
assert received == []
async with connect(_siren_server(), logging_callback=collect, log_level="warning") as client:
await client.call_tool("siren", {})
assert received == ["warning", "error", "critical", "alert", "emergency"]
@requirement("logging:per-request:invalid-level")
async def test_a_request_with_an_unrecognized_log_level_is_rejected(connect: Connect) -> None:
"""A request whose _meta names an unrecognized log level is rejected with -32602 before the handler runs."""
async with connect(_siren_server()) as client:
with pytest.raises(MCPError) as exc_info:
await client.call_tool("siren", {}, meta={LOG_LEVEL_META_KEY: "verbose"})
assert exc_info.value.error.code == INVALID_PARAMS