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. The one exception is the lifespan example, which correctly uses
Context[AppContext] to type the lifespan context.

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

Author: maxisbey

README.md

@@ -229,7 +229,6 @@ from contextlib import asynccontextmanager
 from dataclasses import dataclass
 
 from mcp.server.fastmcp import Context, FastMCP
-from mcp.server.session import ServerSession
 
 
 # Mock database class for example
@@ -275,7 +274,7 @@ mcp = FastMCP("My App", lifespan=app_lifespan)
 
 # Access type-safe lifespan context in tools
 @mcp.tool()
-def query_db(ctx: Context[ServerSession, AppContext]) -> str:
+def query_db(ctx: Context[AppContext]) -> str:
     """Tool that uses initialized resources."""
     db = ctx.request_context.lifespan_context.db
     return db.query()
@@ -347,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.fastmcp import Context, FastMCP
-from mcp.server.session import ServerSession
 
 mcp = FastMCP(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}")
 
@@ -695,13 +693,12 @@ The Context object provides the following capabilities:
 <!-- snippet-source examples/snippets/servers/tool_progress.py -->
 ```python
 from mcp.server.fastmcp import Context, FastMCP
-from mcp.server.session import ServerSession
 
 mcp = FastMCP(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}")
 
@@ -828,7 +825,6 @@ import uuid
 from pydantic import BaseModel, Field
 
 from mcp.server.fastmcp import Context, FastMCP
-from mcp.server.session import ServerSession
 from mcp.shared.exceptions import UrlElicitationRequiredError
 from mcp.types import ElicitRequestURLParams
 
@@ -846,7 +842,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.
@@ -870,7 +866,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
@@ -894,7 +890,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.
@@ -935,14 +931,13 @@ Tools can interact with LLMs through sampling (generating text):
 <!-- snippet-source examples/snippets/servers/sampling.py -->
 ```python
 from mcp.server.fastmcp import Context, FastMCP
-from mcp.server.session import ServerSession
 from mcp.types import SamplingMessage, TextContent
 
 mcp = FastMCP(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}"
 
@@ -972,13 +967,12 @@ Tools can send logs and notifications through the context:
 <!-- snippet-source examples/snippets/servers/notifications.py -->
 ```python
 from mcp.server.fastmcp import Context, FastMCP
-from mcp.server.session import ServerSession
 
 mcp = FastMCP(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}'")

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