modelcontextprotocol/python-sdk
Watch
1
0
Fork
You've already forked python-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Deprecate roots, sampling, and logging methods per SEP-2577 (#2926)

Browse source
This commit is contained in:
Marcelo Trylesinski 2026-06-20 17:25:41 +01:00 • committed by GitHub
commit 4573e4ac33
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
7 changed files with 206 additions and 123 deletions
Download patch file Download diff file Expand all files Collapse all files

12
examples/servers/everything-server/mcp_everything_server/server.py
View file

@ -143,13 +143,13 @@ def test_multiple_content_types() -> list[TextContent | ImageContent | EmbeddedR
@mcp.tool()
async def test_tool_with_logging(ctx: Context) -> str:
"""Tests tool that emits log messages during execution"""
await ctx.info("Tool execution started")
await ctx.info("Tool execution started") # pyright: ignore[reportDeprecated]
await asyncio.sleep(0.05)
await ctx.info("Tool processing data")
await ctx.info("Tool processing data") # pyright: ignore[reportDeprecated]
await asyncio.sleep(0.05)
await ctx.info("Tool execution completed")
await ctx.info("Tool execution completed") # pyright: ignore[reportDeprecated]
return "Tool with logging executed successfully"
@ -176,7 +176,7 @@ async def test_sampling(prompt: str, ctx: Context) -> str:
"""Tests server-initiated sampling (LLM completion request)"""
try:
# Request sampling from client
result = await ctx.session.create_message(
result = await ctx.session.create_message( # pyright: ignore[reportDeprecated]
messages=[SamplingMessage(role="user", content=TextContent(type="text", text=prompt))],
max_tokens=100,
)
@ -314,13 +314,13 @@ def test_error_handling() -> str:
@mcp.tool()
async def test_reconnection(ctx: Context) -> str:
"""Tests SSE polling by closing stream mid-call (SEP-1699)"""
await ctx.info("Before disconnect")
await ctx.info("Before disconnect") # pyright: ignore[reportDeprecated]
await ctx.close_sse_stream()
await asyncio.sleep(0.2) # Wait for client to reconnect
await ctx.info("After reconnect")
await ctx.info("After reconnect") # pyright: ignore[reportDeprecated]
return "Reconnection test completed"

2
examples/servers/simple-streamablehttp-stateless/mcp_simple_streamablehttp_stateless/server.py
View file

@ -49,7 +49,7 @@ async def handle_call_tool(ctx: ServerRequestContext, params: types.CallToolRequ
# Send the specified number of notifications with the given interval
for i in range(count):
await ctx.session.send_log_message(
await ctx.session.send_log_message( # pyright: ignore[reportDeprecated]
level="info",
data=f"Notification {i + 1}/{count} from caller: {caller}",
logger="notification_stream",

2
examples/servers/simple-streamablehttp/mcp_simple_streamablehttp/server.py
View file

@ -54,7 +54,7 @@ async def handle_call_tool(ctx: ServerRequestContext, params: types.CallToolRequ
for i in range(count):
# Include more detailed message for resumability demonstration
notification_msg = f"[{i + 1}/{count}] Event from '{caller}' - Use Last-Event-ID to resume if disconnected"
await ctx.session.send_log_message(
await ctx.session.send_log_message( # pyright: ignore[reportDeprecated]
level="info",
data=notification_msg,
logger="notification_stream",

6
examples/servers/sse-polling-demo/mcp_sse_polling_demo/server.py
View file

@ -74,7 +74,7 @@ async def handle_call_tool(ctx: ServerRequestContext, params: types.CallToolRequ
content=[types.TextContent(type="text", text="Error: checkpoint_every must be between 1 and 20")]
)
await ctx.session.send_log_message(
await ctx.session.send_log_message( # pyright: ignore[reportDeprecated]
level="info",
data=f"Starting batch processing of {items} items...",
logger="process_batch",
@ -86,7 +86,7 @@ async def handle_call_tool(ctx: ServerRequestContext, params: types.CallToolRequ
await anyio.sleep(0.5)
# Report progress
await ctx.session.send_log_message(
await ctx.session.send_log_message( # pyright: ignore[reportDeprecated]
level="info",
data=f"[{i}/{items}] Processing item {i}",
logger="process_batch",
@ -95,7 +95,7 @@ async def handle_call_tool(ctx: ServerRequestContext, params: types.CallToolRequ
# Checkpoint: close stream to trigger client reconnect
if i % checkpoint_every == 0 and i < items:
await ctx.session.send_log_message(
await ctx.session.send_log_message( # pyright: ignore[reportDeprecated]
level="info",
data=f"Checkpoint at item {i} - closing SSE stream for polling",
logger="process_batch",

8
examples/snippets/servers/notifications.py
View file

@ -7,10 +7,10 @@ mcp = MCPServer(name="Notifications Example")
async def process_data(data: str, ctx: Context) -> str:
"""Process data with logging."""
# Different log levels
await ctx.debug(f"Debug: Processing '{data}'")
await ctx.info("Info: Starting processing")
await ctx.warning("Warning: This is experimental")
await ctx.error("Error: (This is just a demo)")
await ctx.debug(f"Debug: Processing '{data}'") # pyright: ignore[reportDeprecated]
await ctx.info("Info: Starting processing") # pyright: ignore[reportDeprecated]
await ctx.warning("Warning: This is experimental") # pyright: ignore[reportDeprecated]
await ctx.error("Error: (This is just a demo)") # pyright: ignore[reportDeprecated]
# Notify about resource changes
await ctx.session.send_resource_list_changed()

2
examples/snippets/servers/sampling.py
View file

@ -9,7 +9,7 @@ async def generate_poem(topic: str, ctx: Context) -> str:
"""Generate a poem using LLM sampling."""
prompt = f"Write a short poem about {topic}"
result = await ctx.session.create_message(
result = await ctx.session.create_message( # pyright: ignore[reportDeprecated]
messages=[
SamplingMessage(
role="user",

4
examples/snippets/servers/tool_progress.py
View file

@ -6,7 +6,7 @@ mcp = MCPServer(name="Progress Example")
@mcp.tool()
async def long_running_task(task_name: str, ctx: Context, steps: int = 5) -> str:
"""Execute a task with progress updates."""
await ctx.info(f"Starting: {task_name}")
await ctx.info(f"Starting: {task_name}") # pyright: ignore[reportDeprecated]
for i in range(steps):
progress = (i + 1) / steps
@ -15,6 +15,6 @@ async def long_running_task(task_name: str, ctx: Context, steps: int = 5) -> str
total=1.0,
message=f"Step {i + 1}/{steps}",
)
await ctx.debug(f"Completed step {i + 1}")
await ctx.debug(f"Completed step {i + 1}") # pyright: ignore[reportDeprecated]
return f"Task '{task_name}' completed"