fix: correct Context type parameters in examples and tests

Context is Generic[LifespanContextT, RequestT], not Generic[ServerSessionT, ...].
The Context[ServerSession, None] pattern scattered across examples, tests, and
docs was passing ServerSession as LifespanContextT, which is semantically wrong
(the type checker only accepted it because LifespanContextT is unbounded).

Both TypeVars have PEP 696 defaults (dict[str, Any] and Any respectively), so
bare Context is the correct annotation when neither parameter needs to be
constrained.

This also removes the now-unused ServerSession imports from the affected files.

Author: maxisbey

README.v2.md

@@ -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}'")

examples/servers/everything-server/mcp_everything_server/server.py

@@ -13,7 +13,6 @@
 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")
 

examples/snippets/servers/elicitation.py

@@ -10,7 +10,6 @@
 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.

examples/snippets/servers/notifications.py

@@ -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}'")

examples/snippets/servers/sampling.py

@@ -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}"
 

examples/snippets/servers/tool_progress.py

@@ -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}")
 

tests/server/mcpserver/test_elicitation.py

@@ -8,7 +8,6 @@
 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 @@ class OptionalSchema(BaseModel):
         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 @@ class InvalidOptionalSchema(BaseModel):
         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 @@ class ValidMultiSelectSchema(BaseModel):
         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 @@ class OptionalMultiSelectSchema(BaseModel):
         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 @@ class DefaultsSchema(BaseModel):
         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 @@ class FavoriteColorSchema(BaseModel):
         )
 
     @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 @@ class FavoriteColorsSchema(BaseModel):
         )
 
     @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 @@ class LegacyColorSchema(BaseModel):
         )
 
     @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}"