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

fix: correct Context type parameters across examples and tests (#2256)

Browse source
This commit is contained in:
Max Isbey 2026-03-09 16:52:56 +00:00 • committed by GitHub
commit 31a38b5078
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
13 changed files with 59 additions and 75 deletions
Download patch file Download diff file Expand all files Collapse all files

19
README.v2.md
View file

@ -346,13 +346,12 @@ Tools can optionally receive a Context object by including a parameter with the
<!-- snippet-source examples/snippets/servers/tool_progress.py -->
```python
from mcp.server.mcpserver import Context, MCPServer
from mcp.server.session import ServerSession
mcp = MCPServer(name="Progress Example")
@mcp.tool()
async def long_running_task(task_name: str, ctx: Context[ServerSession, None], steps: int = 5) -> str:
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}")
@ -694,13 +693,12 @@ The Context object provides the following capabilities:
<!-- snippet-source examples/snippets/servers/tool_progress.py -->
```python
from mcp.server.mcpserver import Context, MCPServer
from mcp.server.session import ServerSession
mcp = MCPServer(name="Progress Example")
@mcp.tool()
async def long_running_task(task_name: str, ctx: Context[ServerSession, None], steps: int = 5) -> str:
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}")
@ -826,7 +824,6 @@ import uuid
from pydantic import BaseModel, Field
from mcp.server.mcpserver import Context, MCPServer
from mcp.server.session import ServerSession
from mcp.shared.exceptions import UrlElicitationRequiredError
from mcp.types import ElicitRequestURLParams
@ -844,7 +841,7 @@ class BookingPreferences(BaseModel):
@mcp.tool()
async def book_table(date: str, time: str, party_size: int, ctx: Context[ServerSession, None]) -> str:
async def book_table(date: str, time: str, party_size: int, ctx: Context) -> str:
"""Book a table with date availability check.
This demonstrates form mode elicitation for collecting non-sensitive user input.
@ -868,7 +865,7 @@ async def book_table(date: str, time: str, party_size: int, ctx: Context[ServerS
@mcp.tool()
async def secure_payment(amount: float, ctx: Context[ServerSession, None]) -> str:
async def secure_payment(amount: float, ctx: Context) -> str:
"""Process a secure payment requiring URL confirmation.
This demonstrates URL mode elicitation using ctx.elicit_url() for
@ -892,7 +889,7 @@ async def secure_payment(amount: float, ctx: Context[ServerSession, None]) -> st
@mcp.tool()
async def connect_service(service_name: str, ctx: Context[ServerSession, None]) -> str:
async def connect_service(service_name: str, ctx: Context) -> str:
"""Connect to a third-party service requiring OAuth authorization.
This demonstrates the "throw error" pattern using UrlElicitationRequiredError.
@ -933,14 +930,13 @@ Tools can interact with LLMs through sampling (generating text):
<!-- snippet-source examples/snippets/servers/sampling.py -->
```python
from mcp.server.mcpserver import Context, MCPServer
from mcp.server.session import ServerSession
from mcp.types import SamplingMessage, TextContent
mcp = MCPServer(name="Sampling Example")
@mcp.tool()
async def generate_poem(topic: str, ctx: Context[ServerSession, None]) -> str:
async def generate_poem(topic: str, ctx: Context) -> str:
"""Generate a poem using LLM sampling."""
prompt = f"Write a short poem about {topic}"
@ -970,13 +966,12 @@ Tools can send logs and notifications through the context:
<!-- snippet-source examples/snippets/servers/notifications.py -->
```python
from mcp.server.mcpserver import Context, MCPServer
from mcp.server.session import ServerSession
mcp = MCPServer(name="Notifications Example")
@mcp.tool()
async def process_data(data: str, ctx: Context[ServerSession, None]) -> str:
async def process_data(data: str, ctx: Context) -> str:
"""Process data with logging."""
# Different log levels
await ctx.debug(f"Debug: Processing '{data}'")

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

@ -13,7 +13,6 @@ import click
from mcp.server import ServerRequestContext
from mcp.server.mcpserver import Context, MCPServer
from mcp.server.mcpserver.prompts.base import UserMessage
from mcp.server.session import ServerSession
from mcp.server.streamable_http import EventCallback, EventMessage, EventStore
from mcp.types import (
AudioContent,
@ -142,7 +141,7 @@ def test_multiple_content_types() -> list[TextContent | ImageContent | EmbeddedR
@mcp.tool()
async def test_tool_with_logging(ctx: Context[ServerSession, None]) -> str:
async def test_tool_with_logging(ctx: Context) -> str:
"""Tests tool that emits log messages during execution"""
await ctx.info("Tool execution started")
await asyncio.sleep(0.05)
@ -155,7 +154,7 @@ async def test_tool_with_logging(ctx: Context[ServerSession, None]) -> str:
@mcp.tool()
async def test_tool_with_progress(ctx: Context[ServerSession, None]) -> str:
async def test_tool_with_progress(ctx: Context) -> str:
"""Tests tool that reports progress notifications"""
await ctx.report_progress(progress=0, total=100, message="Completed step 0 of 100")
await asyncio.sleep(0.05)
@ -173,7 +172,7 @@ async def test_tool_with_progress(ctx: Context[ServerSession, None]) -> str:
@mcp.tool()
async def test_sampling(prompt: str, ctx: Context[ServerSession, None]) -> str:
async def test_sampling(prompt: str, ctx: Context) -> str:
"""Tests server-initiated sampling (LLM completion request)"""
try:
# Request sampling from client
@ -198,7 +197,7 @@ class UserResponse(BaseModel):
@mcp.tool()
async def test_elicitation(message: str, ctx: Context[ServerSession, None]) -> str:
async def test_elicitation(message: str, ctx: Context) -> str:
"""Tests server-initiated elicitation (user input request)"""
try:
# Request user input from client
@ -230,7 +229,7 @@ class SEP1034DefaultsSchema(BaseModel):
@mcp.tool()
async def test_elicitation_sep1034_defaults(ctx: Context[ServerSession, None]) -> str:
async def test_elicitation_sep1034_defaults(ctx: Context) -> str:
"""Tests elicitation with default values for all primitive types (SEP-1034)"""
try:
# Request user input with defaults for all primitive types
@ -289,7 +288,7 @@ class EnumSchemasTestSchema(BaseModel):
@mcp.tool()
async def test_elicitation_sep1330_enums(ctx: Context[ServerSession, None]) -> str:
async def test_elicitation_sep1330_enums(ctx: Context) -> str:
"""Tests elicitation with enum schema variations per SEP-1330"""
try:
result = await ctx.elicit(
@ -313,7 +312,7 @@ def test_error_handling() -> str:
@mcp.tool()
async def test_reconnection(ctx: Context[ServerSession, None]) -> str:
async def test_reconnection(ctx: Context) -> str:
"""Tests SSE polling by closing stream mid-call (SEP-1699)"""
await ctx.info("Before disconnect")

7
examples/snippets/servers/elicitation.py
View file

@ -10,7 +10,6 @@ import uuid
from pydantic import BaseModel, Field
from mcp.server.mcpserver import Context, MCPServer
from mcp.server.session import ServerSession
from mcp.shared.exceptions import UrlElicitationRequiredError
from mcp.types import ElicitRequestURLParams
@ -28,7 +27,7 @@ class BookingPreferences(BaseModel):
@mcp.tool()
async def book_table(date: str, time: str, party_size: int, ctx: Context[ServerSession, None]) -> str:
async def book_table(date: str, time: str, party_size: int, ctx: Context) -> str:
"""Book a table with date availability check.
This demonstrates form mode elicitation for collecting non-sensitive user input.
@ -52,7 +51,7 @@ async def book_table(date: str, time: str, party_size: int, ctx: Context[ServerS
@mcp.tool()
async def secure_payment(amount: float, ctx: Context[ServerSession, None]) -> str:
async def secure_payment(amount: float, ctx: Context) -> str:
"""Process a secure payment requiring URL confirmation.
This demonstrates URL mode elicitation using ctx.elicit_url() for
@ -76,7 +75,7 @@ async def secure_payment(amount: float, ctx: Context[ServerSession, None]) -> st
@mcp.tool()
async def connect_service(service_name: str, ctx: Context[ServerSession, None]) -> str:
async def connect_service(service_name: str, ctx: Context) -> str:
"""Connect to a third-party service requiring OAuth authorization.
This demonstrates the "throw error" pattern using UrlElicitationRequiredError.

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

@ -1,11 +1,10 @@
from mcp.server.mcpserver import Context, MCPServer
from mcp.server.session import ServerSession
mcp = MCPServer(name="Notifications Example")
@mcp.tool()
async def process_data(data: str, ctx: Context[ServerSession, None]) -> str:
async def process_data(data: str, ctx: Context) -> str:
"""Process data with logging."""
# Different log levels
await ctx.debug(f"Debug: Processing '{data}'")

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

@ -1,12 +1,11 @@
from mcp.server.mcpserver import Context, MCPServer
from mcp.server.session import ServerSession
from mcp.types import SamplingMessage, TextContent
mcp = MCPServer(name="Sampling Example")
@mcp.tool()
async def generate_poem(topic: str, ctx: Context[ServerSession, None]) -> str:
async def generate_poem(topic: str, ctx: Context) -> str:
"""Generate a poem using LLM sampling."""
prompt = f"Write a short poem about {topic}"

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

@ -1,11 +1,10 @@
from mcp.server.mcpserver import Context, MCPServer
from mcp.server.session import ServerSession
mcp = MCPServer(name="Progress Example")
@mcp.tool()
async def long_running_task(task_name: str, ctx: Context[ServerSession, None], steps: int = 5) -> str:
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}")

2
tests/client/test_list_roots_callback.py
View file

@ -25,7 +25,7 @@ async def test_list_roots_callback():
return callback_return
@server.tool("test_list_roots")
async def test_list_roots(context: Context[None], message: str):
async def test_list_roots(context: Context, message: str):
roots = await context.session.list_roots()
assert roots == callback_return
return True

21
tests/server/mcpserver/test_elicitation.py
View file

@ -8,7 +8,6 @@ from pydantic import BaseModel, Field
from mcp import Client, types
from mcp.client.session import ClientSession, ElicitationFnT
from mcp.server.mcpserver import Context, MCPServer
from mcp.server.session import ServerSession
from mcp.shared._context import RequestContext
from mcp.types import ElicitRequestParams, ElicitResult, TextContent
@ -22,7 +21,7 @@ def create_ask_user_tool(mcp: MCPServer):
"""Create a standard ask_user tool that handles all elicitation responses."""
@mcp.tool(description="A tool that uses elicitation")
async def ask_user(prompt: str, ctx: Context[ServerSession, None]) -> str:
async def ask_user(prompt: str, ctx: Context) -> str:
result = await ctx.elicit(message=f"Tool wants to ask: {prompt}", schema=AnswerSchema)
if result.action == "accept" and result.data:
@ -97,7 +96,7 @@ async def test_elicitation_schema_validation():
def create_validation_tool(name: str, schema_class: type[BaseModel]):
@mcp.tool(name=name, description=f"Tool testing {name}")
async def tool(ctx: Context[ServerSession, None]) -> str:
async def tool(ctx: Context) -> str:
try:
await ctx.elicit(message="This should fail validation", schema=schema_class)
return "Should not reach here" # pragma: no cover
@ -147,7 +146,7 @@ async def test_elicitation_with_optional_fields():
subscribe: bool | None = Field(default=False, description="Subscribe to newsletter?")
@mcp.tool(description="Tool with optional fields")
async def optional_tool(ctx: Context[ServerSession, None]) -> str:
async def optional_tool(ctx: Context) -> str:
result = await ctx.elicit(message="Please provide your information", schema=OptionalSchema)
if result.action == "accept" and result.data:
@ -188,7 +187,7 @@ async def test_elicitation_with_optional_fields():
optional_list: list[int] | None = Field(default=None, description="Invalid optional list")
@mcp.tool(description="Tool with invalid optional field")
async def invalid_optional_tool(ctx: Context[ServerSession, None]) -> str:
async def invalid_optional_tool(ctx: Context) -> str:
try:
await ctx.elicit(message="This should fail", schema=InvalidOptionalSchema)
return "Should not reach here" # pragma: no cover
@ -214,7 +213,7 @@ async def test_elicitation_with_optional_fields():
tags: list[str] = Field(description="Tags")
@mcp.tool(description="Tool with valid list[str] field")
async def valid_multiselect_tool(ctx: Context[ServerSession, None]) -> str:
async def valid_multiselect_tool(ctx: Context) -> str:
result = await ctx.elicit(message="Please provide tags", schema=ValidMultiSelectSchema)
if result.action == "accept" and result.data:
return f"Name: {result.data.name}, Tags: {', '.join(result.data.tags)}"
@ -233,7 +232,7 @@ async def test_elicitation_with_optional_fields():
tags: list[str] | None = Field(default=None, description="Optional tags")
@mcp.tool(description="Tool with optional list[str] field")
async def optional_multiselect_tool(ctx: Context[ServerSession, None]) -> str:
async def optional_multiselect_tool(ctx: Context) -> str:
result = await ctx.elicit(message="Please provide optional tags", schema=OptionalMultiSelectSchema)
if result.action == "accept" and result.data:
tags_str = ", ".join(result.data.tags) if result.data.tags else "none"
@ -262,7 +261,7 @@ async def test_elicitation_with_default_values():
email: str = Field(description="Email address (required)")
@mcp.tool(description="Tool with default values")
async def defaults_tool(ctx: Context[ServerSession, None]) -> str:
async def defaults_tool(ctx: Context) -> str:
result = await ctx.elicit(message="Please provide your information", schema=DefaultsSchema)
if result.action == "accept" and result.data:
@ -327,7 +326,7 @@ async def test_elicitation_with_enum_titles():
)
@mcp.tool(description="Single color selection")
async def select_favorite_color(ctx: Context[ServerSession, None]) -> str:
async def select_favorite_color(ctx: Context) -> str:
result = await ctx.elicit(message="Select your favorite color", schema=FavoriteColorSchema)
if result.action == "accept" and result.data:
return f"User: {result.data.user_name}, Favorite: {result.data.favorite_color}"
@ -351,7 +350,7 @@ async def test_elicitation_with_enum_titles():
)
@mcp.tool(description="Multiple color selection")
async def select_favorite_colors(ctx: Context[ServerSession, None]) -> str:
async def select_favorite_colors(ctx: Context) -> str:
result = await ctx.elicit(message="Select your favorite colors", schema=FavoriteColorsSchema)
if result.action == "accept" and result.data:
return f"User: {result.data.user_name}, Colors: {', '.join(result.data.favorite_colors)}"
@ -366,7 +365,7 @@ async def test_elicitation_with_enum_titles():
)
@mcp.tool(description="Legacy enum format")
async def select_color_legacy(ctx: Context[ServerSession, None]) -> str:
async def select_color_legacy(ctx: Context) -> str:
result = await ctx.elicit(message="Select a color (legacy format)", schema=LegacyColorSchema)
if result.action == "accept" and result.data:
return f"User: {result.data.user_name}, Color: {result.data.color}"

17
tests/server/mcpserver/test_server.py
View file

@ -17,7 +17,6 @@ from mcp.server.mcpserver.exceptions import ToolError
from mcp.server.mcpserver.prompts.base import Message, UserMessage
from mcp.server.mcpserver.resources import FileResource, FunctionResource
from mcp.server.mcpserver.utilities.types import Audio, Image
from mcp.server.session import ServerSession
from mcp.server.transport_security import TransportSecuritySettings
from mcp.shared.exceptions import MCPError
from mcp.types import (
@ -1003,7 +1002,7 @@ class TestContextInjection:
"""Test that context parameters are properly detected."""
mcp = MCPServer()
def tool_with_context(x: int, ctx: Context[ServerSession, None]) -> str: # pragma: no cover
def tool_with_context(x: int, ctx: Context) -> str: # pragma: no cover
return f"Request {ctx.request_id}: {x}"
tool = mcp._tool_manager.add_tool(tool_with_context)
@ -1013,7 +1012,7 @@ class TestContextInjection:
"""Test that context is properly injected into tool calls."""
mcp = MCPServer()
def tool_with_context(x: int, ctx: Context[ServerSession, None]) -> str:
def tool_with_context(x: int, ctx: Context) -> str:
assert ctx.request_id is not None
return f"Request {ctx.request_id}: {x}"
@ -1030,7 +1029,7 @@ class TestContextInjection:
"""Test that context works in async functions."""
mcp = MCPServer()
async def async_tool(x: int, ctx: Context[ServerSession, None]) -> str:
async def async_tool(x: int, ctx: Context) -> str:
assert ctx.request_id is not None
return f"Async request {ctx.request_id}: {x}"
@ -1047,7 +1046,7 @@ class TestContextInjection:
"""Test that context logging methods work."""
mcp = MCPServer()
async def logging_tool(msg: str, ctx: Context[ServerSession, None]) -> str:
async def logging_tool(msg: str, ctx: Context) -> str:
await ctx.debug("Debug message")
await ctx.info("Info message")
await ctx.warning("Warning message")
@ -1094,7 +1093,7 @@ class TestContextInjection:
return "resource data"
@mcp.tool()
async def tool_with_resource(ctx: Context[ServerSession, None]) -> str:
async def tool_with_resource(ctx: Context) -> str:
r_iter = await ctx.read_resource("test://data")
r_list = list(r_iter)
assert len(r_list) == 1
@ -1113,7 +1112,7 @@ class TestContextInjection:
mcp = MCPServer()
@mcp.resource("resource://context/{name}")
def resource_with_context(name: str, ctx: Context[ServerSession, None]) -> str:
def resource_with_context(name: str, ctx: Context) -> str:
"""Resource that receives context."""
assert ctx is not None
return f"Resource {name} - context injected"
@ -1166,7 +1165,7 @@ class TestContextInjection:
mcp = MCPServer()
@mcp.resource("resource://custom/{id}")
def resource_custom_ctx(id: str, my_ctx: Context[ServerSession, None]) -> str:
def resource_custom_ctx(id: str, my_ctx: Context) -> str:
"""Resource with custom context parameter name."""
assert my_ctx is not None
return f"Resource {id} with context"
@ -1194,7 +1193,7 @@ class TestContextInjection:
mcp = MCPServer()
@mcp.prompt("prompt_with_ctx")
def prompt_with_context(text: str, ctx: Context[ServerSession, None]) -> str:
def prompt_with_context(text: str, ctx: Context) -> str:
"""Prompt that expects context."""
assert ctx is not None
return f"Prompt '{text}' - context injected"

11
tests/server/mcpserver/test_tool_manager.py
View file

@ -11,7 +11,6 @@ from mcp.server.mcpserver import Context, MCPServer
from mcp.server.mcpserver.exceptions import ToolError
from mcp.server.mcpserver.tools import Tool, ToolManager
from mcp.server.mcpserver.utilities.func_metadata import ArgModelBase, FuncMetadata
from mcp.server.session import ServerSessionT
from mcp.types import TextContent, ToolAnnotations
@ -319,7 +318,7 @@ class TestCallTools:
class TestToolSchema:
@pytest.mark.anyio
async def test_context_arg_excluded_from_schema(self):
def something(a: int, ctx: Context[ServerSessionT, None]) -> int: # pragma: no cover
def something(a: int, ctx: Context) -> int: # pragma: no cover
return a
manager = ToolManager()
@ -336,7 +335,7 @@ class TestContextHandling:
"""Test that context parameters are properly detected in
Tool.from_function()."""
def tool_with_context(x: int, ctx: Context[ServerSessionT, None]) -> str: # pragma: no cover
def tool_with_context(x: int, ctx: Context) -> str: # pragma: no cover
return str(x)
manager = ToolManager()
@ -359,7 +358,7 @@ class TestContextHandling:
async def test_context_injection(self):
"""Test that context is properly injected during tool execution."""
def tool_with_context(x: int, ctx: Context[ServerSessionT, None]) -> str:
def tool_with_context(x: int, ctx: Context) -> str:
assert isinstance(ctx, Context)
return str(x)
@ -373,7 +372,7 @@ class TestContextHandling:
async def test_context_injection_async(self):
"""Test that context is properly injected in async tools."""
async def async_tool(x: int, ctx: Context[ServerSessionT, None]) -> str:
async def async_tool(x: int, ctx: Context) -> str:
assert isinstance(ctx, Context)
return str(x)
@ -387,7 +386,7 @@ class TestContextHandling:
async def test_context_error_handling(self):
"""Test error handling when context injection fails."""
def tool_with_context(x: int, ctx: Context[ServerSessionT, None]) -> str:
def tool_with_context(x: int, ctx: Context) -> str:
raise ValueError("Test error")
manager = ToolManager()

23
tests/server/mcpserver/test_url_elicitation.py
View file

@ -8,7 +8,6 @@ from mcp import Client, types
from mcp.client.session import ClientSession
from mcp.server.elicitation import CancelledElicitation, DeclinedElicitation, elicit_url
from mcp.server.mcpserver import Context, MCPServer
from mcp.server.session import ServerSession
from mcp.shared._context import RequestContext
from mcp.types import ElicitRequestParams, ElicitResult, TextContent
@ -19,7 +18,7 @@ async def test_url_elicitation_accept():
mcp = MCPServer(name="URLElicitationServer")
@mcp.tool(description="A tool that uses URL elicitation")
async def request_api_key(ctx: Context[ServerSession, None]) -> str:
async def request_api_key(ctx: Context) -> str:
result = await ctx.session.elicit_url(
message="Please provide your API key to continue.",
url="https://example.com/api_key_setup",
@ -49,7 +48,7 @@ async def test_url_elicitation_decline():
mcp = MCPServer(name="URLElicitationDeclineServer")
@mcp.tool(description="A tool that uses URL elicitation")
async def oauth_flow(ctx: Context[ServerSession, None]) -> str:
async def oauth_flow(ctx: Context) -> str:
result = await ctx.session.elicit_url(
message="Authorize access to your files.",
url="https://example.com/oauth/authorize",
@ -75,7 +74,7 @@ async def test_url_elicitation_cancel():
mcp = MCPServer(name="URLElicitationCancelServer")
@mcp.tool(description="A tool that uses URL elicitation")
async def payment_flow(ctx: Context[ServerSession, None]) -> str:
async def payment_flow(ctx: Context) -> str:
result = await ctx.session.elicit_url(
message="Complete payment to proceed.",
url="https://example.com/payment",
@ -101,7 +100,7 @@ async def test_url_elicitation_helper_function():
mcp = MCPServer(name="URLElicitationHelperServer")
@mcp.tool(description="Tool using elicit_url helper")
async def setup_credentials(ctx: Context[ServerSession, None]) -> str:
async def setup_credentials(ctx: Context) -> str:
result = await elicit_url(
session=ctx.session,
message="Set up your credentials",
@ -127,7 +126,7 @@ async def test_url_no_content_in_response():
mcp = MCPServer(name="URLContentCheckServer")
@mcp.tool(description="Check URL response format")
async def check_url_response(ctx: Context[ServerSession, None]) -> str:
async def check_url_response(ctx: Context) -> str:
result = await ctx.session.elicit_url(
message="Test message",
url="https://example.com/test",
@ -164,7 +163,7 @@ async def test_form_mode_still_works():
name: str = Field(description="Your name")
@mcp.tool(description="Test form mode")
async def ask_name(ctx: Context[ServerSession, None]) -> str:
async def ask_name(ctx: Context) -> str:
result = await ctx.elicit(message="What is your name?", schema=NameSchema)
# Test only checks accept path with data
assert result.action == "accept"
@ -195,7 +194,7 @@ async def test_elicit_complete_notification():
notification_sent = False
@mcp.tool(description="Tool that sends completion notification")
async def trigger_elicitation(ctx: Context[ServerSession, None]) -> str:
async def trigger_elicitation(ctx: Context) -> str:
nonlocal notification_sent
# Simulate an async operation (e.g., user completing auth in browser)
@ -238,7 +237,7 @@ async def test_elicit_url_typed_results():
mcp = MCPServer(name="TypedResultsServer")
@mcp.tool(description="Test declined result")
async def test_decline(ctx: Context[ServerSession, None]) -> str:
async def test_decline(ctx: Context) -> str:
result = await elicit_url(
session=ctx.session,
message="Test decline",
@ -251,7 +250,7 @@ async def test_elicit_url_typed_results():
return "Not declined" # pragma: no cover
@mcp.tool(description="Test cancelled result")
async def test_cancel(ctx: Context[ServerSession, None]) -> str:
async def test_cancel(ctx: Context) -> str:
result = await elicit_url(
session=ctx.session,
message="Test cancel",
@ -293,7 +292,7 @@ async def test_deprecated_elicit_method():
email: str = Field(description="Email address")
@mcp.tool(description="Test deprecated elicit method")
async def use_deprecated_elicit(ctx: Context[ServerSession, None]) -> str:
async def use_deprecated_elicit(ctx: Context) -> str:
# Use the deprecated elicit() method which should call elicit_form()
result = await ctx.session.elicit(
message="Enter your email",
@ -323,7 +322,7 @@ async def test_ctx_elicit_url_convenience_method():
mcp = MCPServer(name="CtxElicitUrlServer")
@mcp.tool(description="A tool that uses ctx.elicit_url() directly")
async def direct_elicit_url(ctx: Context[ServerSession, None]) -> str:
async def direct_elicit_url(ctx: Context) -> str:
# Use ctx.elicit_url() directly instead of ctx.session.elicit_url()
result = await ctx.elicit_url(
message="Test the convenience method",

7
tests/server/mcpserver/test_url_elicitation_error_throw.py
View file

@ -5,7 +5,6 @@ from inline_snapshot import snapshot
from mcp import Client, ErrorData, types
from mcp.server.mcpserver import Context, MCPServer
from mcp.server.session import ServerSession
from mcp.shared.exceptions import MCPError, UrlElicitationRequiredError
@ -15,7 +14,7 @@ async def test_url_elicitation_error_thrown_from_tool():
mcp = MCPServer(name="UrlElicitationErrorServer")
@mcp.tool(description="A tool that raises UrlElicitationRequiredError")
async def connect_service(service_name: str, ctx: Context[ServerSession, None]) -> str:
async def connect_service(service_name: str, ctx: Context) -> str:
# This tool cannot proceed without authorization
raise UrlElicitationRequiredError(
[
@ -56,7 +55,7 @@ async def test_url_elicitation_error_from_error():
mcp = MCPServer(name="UrlElicitationErrorServer")
@mcp.tool(description="A tool that raises UrlElicitationRequiredError with multiple elicitations")
async def multi_auth(ctx: Context[ServerSession, None]) -> str:
async def multi_auth(ctx: Context) -> str:
raise UrlElicitationRequiredError(
[
types.ElicitRequestURLParams(
@ -97,7 +96,7 @@ async def test_normal_exceptions_still_return_error_result():
mcp = MCPServer(name="NormalErrorServer")
@mcp.tool(description="A tool that raises a normal exception")
async def failing_tool(ctx: Context[ServerSession, None]) -> str:
async def failing_tool(ctx: Context) -> str:
raise ValueError("Something went wrong")
async with Client(mcp) as client:

3
tests/server/test_lifespan.py
View file

@ -11,7 +11,6 @@ from mcp.server import ServerRequestContext
from mcp.server.lowlevel.server import NotificationOptions, Server
from mcp.server.mcpserver import Context, MCPServer
from mcp.server.models import InitializationOptions
from mcp.server.session import ServerSession
from mcp.shared.message import SessionMessage
from mcp.types import (
CallToolRequestParams,
@ -143,7 +142,7 @@ async def test_mcpserver_server_lifespan():
# Add a tool that checks lifespan context
@server.tool()
def check_lifespan(ctx: Context[ServerSession, None]) -> bool:
def check_lifespan(ctx: Context) -> bool:
"""Tool that checks lifespan context."""
assert isinstance(ctx.request_context.lifespan_context, dict)
assert ctx.request_context.lifespan_context["started"]