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

refactor: remove request_ctx ContextVar, thread Context explicitly (#2203)

Browse source 
Co-authored-by: Marcelo Trylesinski <marcelotryle@gmail.com>
This commit is contained in:
Max Isbey 2026-03-04 13:23:02 +00:00 • committed by GitHub
commit cc22bf5464
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
11 changed files with 484 additions and 413 deletions
Download patch file Download diff file Expand all files Collapse all files

9
src/mcp/server/lowlevel/server.py
View file

@ -36,7 +36,6 @@ handler callables by method string.
from __future__ import annotations
import contextvars
import logging
import warnings
from collections.abc import AsyncIterator, Awaitable, Callable
@ -74,8 +73,6 @@ logger = logging.getLogger(__name__)
LifespanResultT = TypeVar("LifespanResultT", default=Any)
request_ctx: contextvars.ContextVar[ServerRequestContext[Any]] = contextvars.ContextVar("request_ctx")
class NotificationOptions:
def __init__(self, prompts_changed: bool = False, resources_changed: bool = False, tools_changed: bool = False):
@ -474,11 +471,7 @@ class Server(Generic[LifespanResultT]):
close_sse_stream=close_sse_stream_cb,
close_standalone_sse_stream=close_standalone_sse_stream_cb,
)
token = request_ctx.set(ctx)
try:
response = await handler(ctx, req.params)
finally:
request_ctx.reset(token)
response = await handler(ctx, req.params)
except MCPError as err:
response = err.error
except anyio.get_cancelled_exc_class():

3
src/mcp/server/mcpserver/__init__.py
View file

@ -2,7 +2,8 @@
from mcp.types import Icon
from .server import Context, MCPServer
from .context import Context
from .server import MCPServer
from .utilities.types import Audio, Image
__all__ = ["MCPServer", "Context", "Image", "Audio", "Icon"]

280
src/mcp/server/mcpserver/context.py Normal file
View file

@ -0,0 +1,280 @@
from __future__ import annotations
from collections.abc import Iterable
from typing import TYPE_CHECKING, Any, Generic, Literal
from pydantic import AnyUrl, BaseModel
from mcp.server.context import LifespanContextT, RequestT, ServerRequestContext
from mcp.server.elicitation import (
ElicitationResult,
ElicitSchemaModelT,
UrlElicitationResult,
elicit_url,
elicit_with_validation,
)
from mcp.server.lowlevel.helper_types import ReadResourceContents
if TYPE_CHECKING:
from mcp.server.mcpserver.server import MCPServer
class Context(BaseModel, Generic[LifespanContextT, RequestT]):
"""Context object providing access to MCP capabilities.
This provides a cleaner interface to MCP's RequestContext functionality.
It gets injected into tool and resource functions that request it via type hints.
To use context in a tool function, add a parameter with the Context type annotation:
```python
@server.tool()
async def my_tool(x: int, ctx: Context) -> str:
# Log messages to the client
await ctx.info(f"Processing {x}")
await ctx.debug("Debug info")
await ctx.warning("Warning message")
await ctx.error("Error message")
# Report progress
await ctx.report_progress(50, 100)
# Access resources
data = await ctx.read_resource("resource://data")
# Get request info
request_id = ctx.request_id
client_id = ctx.client_id
return str(x)
```
The context parameter name can be anything as long as it's annotated with Context.
The context is optional - tools that don't need it can omit the parameter.
"""
_request_context: ServerRequestContext[LifespanContextT, RequestT] | None
_mcp_server: MCPServer | None
# TODO(maxisbey): Consider making request_context/mcp_server required, or refactor Context entirely.
def __init__(
self,
*,
request_context: ServerRequestContext[LifespanContextT, RequestT] | None = None,
mcp_server: MCPServer | None = None,
# TODO(Marcelo): We should drop this kwargs parameter.
**kwargs: Any,
):
super().__init__(**kwargs)
self._request_context = request_context
self._mcp_server = mcp_server
@property
def mcp_server(self) -> MCPServer:
"""Access to the MCPServer instance."""
if self._mcp_server is None: # pragma: no cover
raise ValueError("Context is not available outside of a request")
return self._mcp_server # pragma: no cover
@property
def request_context(self) -> ServerRequestContext[LifespanContextT, RequestT]:
"""Access to the underlying request context."""
if self._request_context is None: # pragma: no cover
raise ValueError("Context is not available outside of a request")
return self._request_context
async def report_progress(self, progress: float, total: float | None = None, message: str | None = None) -> None:
"""Report progress for the current operation.
Args:
progress: Current progress value (e.g., 24)
total: Optional total value (e.g., 100)
message: Optional message (e.g., "Starting render...")
"""
progress_token = self.request_context.meta.get("progress_token") if self.request_context.meta else None
if progress_token is None: # pragma: no cover
return
await self.request_context.session.send_progress_notification(
progress_token=progress_token,
progress=progress,
total=total,
message=message,
related_request_id=self.request_id,
)
async def read_resource(self, uri: str | AnyUrl) -> Iterable[ReadResourceContents]:
"""Read a resource by URI.
Args:
uri: Resource URI to read
Returns:
The resource content as either text or bytes
"""
assert self._mcp_server is not None, "Context is not available outside of a request"
return await self._mcp_server.read_resource(uri, self)
async def elicit(
self,
message: str,
schema: type[ElicitSchemaModelT],
) -> ElicitationResult[ElicitSchemaModelT]:
"""Elicit information from the client/user.
This method can be used to interactively ask for additional information from the
client within a tool's execution. The client might display the message to the
user and collect a response according to the provided schema. If the client
is an agent, it might decide how to handle the elicitation -- either by asking
the user or automatically generating a response.
Args:
message: Message to present to the user
schema: A Pydantic model class defining the expected response structure.
According to the specification, only primitive types are allowed.
Returns:
An ElicitationResult containing the action taken and the data if accepted
Note:
Check the result.action to determine if the user accepted, declined, or cancelled.
The result.data will only be populated if action is "accept" and validation succeeded.
"""
return await elicit_with_validation(
session=self.request_context.session,
message=message,
schema=schema,
related_request_id=self.request_id,
)
async def elicit_url(
self,
message: str,
url: str,
elicitation_id: str,
) -> UrlElicitationResult:
"""Request URL mode elicitation from the client.
This directs the user to an external URL for out-of-band interactions
that must not pass through the MCP client. Use this for:
- Collecting sensitive credentials (API keys, passwords)
- OAuth authorization flows with third-party services
- Payment and subscription flows
- Any interaction where data should not pass through the LLM context
The response indicates whether the user consented to navigate to the URL.
The actual interaction happens out-of-band. When the elicitation completes,
call `ctx.session.send_elicit_complete(elicitation_id)` to notify the client.
Args:
message: Human-readable explanation of why the interaction is needed
url: The URL the user should navigate to
elicitation_id: Unique identifier for tracking this elicitation
Returns:
UrlElicitationResult indicating accept, decline, or cancel
"""
return await elicit_url(
session=self.request_context.session,
message=message,
url=url,
elicitation_id=elicitation_id,
related_request_id=self.request_id,
)
async def log(
self,
level: Literal["debug", "info", "warning", "error"],
message: str,
*,
logger_name: str | None = None,
extra: dict[str, Any] | None = None,
) -> None:
"""Send a log message to the client.
Args:
level: Log level (debug, info, warning, error)
message: Log message
logger_name: Optional logger name
extra: Optional dictionary with additional structured data to include
"""
if extra:
log_data = {"message": message, **extra}
else:
log_data = message
await self.request_context.session.send_log_message(
level=level,
data=log_data,
logger=logger_name,
related_request_id=self.request_id,
)
@property
def client_id(self) -> str | None:
"""Get the client ID if available."""
return self.request_context.meta.get("client_id") if self.request_context.meta else None # pragma: no cover
@property
def request_id(self) -> str:
"""Get the unique ID for this request."""
return str(self.request_context.request_id)
@property
def session(self):
"""Access to the underlying session for advanced usage."""
return self.request_context.session
async def close_sse_stream(self) -> None:
"""Close the SSE stream to trigger client reconnection.
This method closes the HTTP connection for the current request, triggering
client reconnection. Events continue to be stored in the event store and will
be replayed when the client reconnects with Last-Event-ID.
Use this to implement polling behavior during long-running operations -
the client will reconnect after the retry interval specified in the priming event.
Note:
This is a no-op if not using StreamableHTTP transport with event_store.
The callback is only available when event_store is configured.
"""
if self._request_context and self._request_context.close_sse_stream: # pragma: no cover
await self._request_context.close_sse_stream()
async def close_standalone_sse_stream(self) -> None:
"""Close the standalone GET SSE stream to trigger client reconnection.
This method closes the HTTP connection for the standalone GET stream used
for unsolicited server-to-client notifications. The client SHOULD reconnect
with Last-Event-ID to resume receiving notifications.
Note:
This is a no-op if not using StreamableHTTP transport with event_store.
Currently, client reconnection for standalone GET streams is NOT
implemented - this is a known gap.
"""
if self._request_context and self._request_context.close_standalone_sse_stream: # pragma: no cover
await self._request_context.close_standalone_sse_stream()
# Convenience methods for common log levels
async def debug(self, message: str, *, logger_name: str | None = None, extra: dict[str, Any] | None = None) -> None:
"""Send a debug log message."""
await self.log("debug", message, logger_name=logger_name, extra=extra)
async def info(self, message: str, *, logger_name: str | None = None, extra: dict[str, Any] | None = None) -> None:
"""Send an info log message."""
await self.log("info", message, logger_name=logger_name, extra=extra)
async def warning(
self, message: str, *, logger_name: str | None = None, extra: dict[str, Any] | None = None
) -> None:
"""Send a warning log message."""
await self.log("warning", message, logger_name=logger_name, extra=extra)
async def error(self, message: str, *, logger_name: str | None = None, extra: dict[str, Any] | None = None) -> None:
"""Send an error log message."""
await self.log("error", message, logger_name=logger_name, extra=extra)

12
src/mcp/server/mcpserver/prompts/base.py
View file

@ -15,7 +15,7 @@ from mcp.types import ContentBlock, Icon, TextContent
if TYPE_CHECKING:
from mcp.server.context import LifespanContextT, RequestT
from mcp.server.mcpserver.server import Context
from mcp.server.mcpserver.context import Context
class Message(BaseModel):
@ -135,10 +135,14 @@ class Prompt(BaseModel):
async def render(
self,
arguments: dict[str, Any] | None = None,
context: Context[LifespanContextT, RequestT] | None = None,
arguments: dict[str, Any] | None,
context: Context[LifespanContextT, RequestT],
) -> list[Message]:
"""Render the prompt with arguments."""
"""Render the prompt with arguments.
Raises:
ValueError: If required arguments are missing, or if rendering fails.
"""
# Validate required arguments
if self.arguments:
required = {arg.name for arg in self.arguments if arg.required}

8
src/mcp/server/mcpserver/prompts/manager.py
View file

@ -9,7 +9,7 @@ from mcp.server.mcpserver.utilities.logging import get_logger
if TYPE_CHECKING:
from mcp.server.context import LifespanContextT, RequestT
from mcp.server.mcpserver.server import Context
from mcp.server.mcpserver.context import Context
logger = get_logger(__name__)
@ -48,12 +48,12 @@ class PromptManager:
async def render_prompt(
self,
name: str,
arguments: dict[str, Any] | None = None,
context: Context[LifespanContextT, RequestT] | None = None,
arguments: dict[str, Any] | None,
context: Context[LifespanContextT, RequestT],
) -> list[Message]:
"""Render a prompt by name with arguments."""
prompt = self.get_prompt(name)
if not prompt:
raise ValueError(f"Unknown prompt: {name}")
return await prompt.render(arguments, context=context)
return await prompt.render(arguments, context)

6
src/mcp/server/mcpserver/resources/resource_manager.py
View file

@ -14,7 +14,7 @@ from mcp.types import Annotations, Icon
if TYPE_CHECKING:
from mcp.server.context import LifespanContextT, RequestT
from mcp.server.mcpserver.server import Context
from mcp.server.mcpserver.context import Context
logger = get_logger(__name__)
@ -80,9 +80,7 @@ class ResourceManager:
self._templates[template.uri_template] = template
return template
async def get_resource(
self, uri: AnyUrl | str, context: Context[LifespanContextT, RequestT] | None = None
) -> Resource:
async def get_resource(self, uri: AnyUrl | str, context: Context[LifespanContextT, RequestT]) -> Resource:
"""Get resource by URI, checking concrete resources first, then templates."""
uri_str = str(uri)
logger.debug("Getting resource", extra={"uri": uri_str})

10
src/mcp/server/mcpserver/resources/templates.py
View file

@ -17,7 +17,7 @@ from mcp.types import Annotations, Icon
if TYPE_CHECKING:
from mcp.server.context import LifespanContextT, RequestT
from mcp.server.mcpserver.server import Context
from mcp.server.mcpserver.context import Context
class ResourceTemplate(BaseModel):
@ -99,9 +99,13 @@ class ResourceTemplate(BaseModel):
self,
uri: str,
params: dict[str, Any],
context: Context[LifespanContextT, RequestT] | None = None,
context: Context[LifespanContextT, RequestT],
) -> Resource:
"""Create a resource from the template with the given parameters."""
"""Create a resource from the template with the given parameters.
Raises:
ValueError: If creating the resource fails.
"""
try:
# Add context to params if needed
params = inject_context(self.fn, params, context, self.context_kwarg)

316
src/mcp/server/mcpserver/server.py
View file

@ -12,7 +12,6 @@ from typing import Any, Generic, Literal, TypeVar, overload
import anyio
import pydantic_core
from pydantic import BaseModel
from pydantic.networks import AnyUrl
from pydantic_settings import BaseSettings, SettingsConfigDict
from starlette.applications import Starlette
@ -27,12 +26,11 @@ from mcp.server.auth.middleware.auth_context import AuthContextMiddleware
from mcp.server.auth.middleware.bearer_auth import BearerAuthBackend, RequireAuthMiddleware
from mcp.server.auth.provider import OAuthAuthorizationServerProvider, ProviderTokenVerifier, TokenVerifier
from mcp.server.auth.settings import AuthSettings
from mcp.server.context import LifespanContextT, RequestT, ServerRequestContext
from mcp.server.elicitation import ElicitationResult, ElicitSchemaModelT, UrlElicitationResult, elicit_with_validation
from mcp.server.elicitation import elicit_url as _elicit_url
from mcp.server.context import ServerRequestContext
from mcp.server.lowlevel.helper_types import ReadResourceContents
from mcp.server.lowlevel.server import LifespanResultT, Server, request_ctx
from mcp.server.lowlevel.server import LifespanResultT, Server
from mcp.server.lowlevel.server import lifespan as default_lifespan
from mcp.server.mcpserver.context import Context
from mcp.server.mcpserver.exceptions import ResourceError
from mcp.server.mcpserver.prompts import Prompt, PromptManager
from mcp.server.mcpserver.resources import FunctionResource, Resource, ResourceManager
@ -300,8 +298,9 @@ class MCPServer(Generic[LifespanResultT]):
async def _handle_call_tool(
self, ctx: ServerRequestContext[LifespanResultT], params: CallToolRequestParams
) -> CallToolResult:
context = Context(request_context=ctx, mcp_server=self)
try:
result = await self.call_tool(params.name, params.arguments or {})
result = await self.call_tool(params.name, params.arguments or {}, context)
except MCPError:
raise
except Exception as e:
@ -332,7 +331,8 @@ class MCPServer(Generic[LifespanResultT]):
async def _handle_read_resource(
self, ctx: ServerRequestContext[LifespanResultT], params: ReadResourceRequestParams
) -> ReadResourceResult:
results = await self.read_resource(params.uri)
context = Context(request_context=ctx, mcp_server=self)
results = await self.read_resource(params.uri, context)
contents: list[TextResourceContents | BlobResourceContents] = []
for item in results:
if isinstance(item.content, bytes):
@ -368,7 +368,8 @@ class MCPServer(Generic[LifespanResultT]):
async def _handle_get_prompt(
self, ctx: ServerRequestContext[LifespanResultT], params: GetPromptRequestParams
) -> GetPromptResult:
return await self.get_prompt(params.name, params.arguments)
context = Context(request_context=ctx, mcp_server=self)
return await self.get_prompt(params.name, params.arguments, context)
async def list_tools(self) -> list[MCPTool]:
"""List all available tools."""
@ -387,22 +388,13 @@ class MCPServer(Generic[LifespanResultT]):
for info in tools
]
def get_context(self) -> Context[LifespanResultT, Request]:
"""Return a Context object.
Note that the context will only be valid during a request; outside a
request, most methods will error.
"""
try:
request_context = request_ctx.get()
except LookupError:
request_context = None
return Context(request_context=request_context, mcp_server=self)
async def call_tool(self, name: str, arguments: dict[str, Any]) -> Sequence[ContentBlock] | dict[str, Any]:
async def call_tool(
self, name: str, arguments: dict[str, Any], context: Context[LifespanResultT, Any] | None = None
) -> Sequence[ContentBlock] | dict[str, Any]:
"""Call a tool by name with arguments."""
context = self.get_context()
return await self._tool_manager.call_tool(name, arguments, context=context, convert_result=True)
if context is None:
context = Context(mcp_server=self)
return await self._tool_manager.call_tool(name, arguments, context, convert_result=True)
async def list_resources(self) -> list[MCPResource]:
"""List all available resources."""
@ -438,12 +430,14 @@ class MCPServer(Generic[LifespanResultT]):
for template in templates
]
async def read_resource(self, uri: AnyUrl | str) -> Iterable[ReadResourceContents]:
async def read_resource(
self, uri: AnyUrl | str, context: Context[LifespanResultT, Any] | None = None
) -> Iterable[ReadResourceContents]:
"""Read a resource by URI."""
context = self.get_context()
if context is None:
context = Context(mcp_server=self)
try:
resource = await self._resource_manager.get_resource(uri, context=context)
resource = await self._resource_manager.get_resource(uri, context)
except ValueError:
raise ResourceError(f"Unknown resource: {uri}")
@ -1087,14 +1081,18 @@ class MCPServer(Generic[LifespanResultT]):
for prompt in prompts
]
async def get_prompt(self, name: str, arguments: dict[str, Any] | None = None) -> GetPromptResult:
async def get_prompt(
self, name: str, arguments: dict[str, Any] | None = None, context: Context[LifespanResultT, Any] | None = None
) -> GetPromptResult:
"""Get a prompt by name with arguments."""
if context is None:
context = Context(mcp_server=self)
try:
prompt = self._prompt_manager.get_prompt(name)
if not prompt:
raise ValueError(f"Unknown prompt: {name}")
messages = await prompt.render(arguments, context=self.get_context())
messages = await prompt.render(arguments, context)
return GetPromptResult(
description=prompt.description,
@ -1103,263 +1101,3 @@ class MCPServer(Generic[LifespanResultT]):
except Exception as e:
logger.exception(f"Error getting prompt {name}")
raise ValueError(str(e))
class Context(BaseModel, Generic[LifespanContextT, RequestT]):
"""Context object providing access to MCP capabilities.
This provides a cleaner interface to MCP's RequestContext functionality.
It gets injected into tool and resource functions that request it via type hints.
To use context in a tool function, add a parameter with the Context type annotation:
```python
@server.tool()
async def my_tool(x: int, ctx: Context) -> str:
# Log messages to the client
await ctx.info(f"Processing {x}")
await ctx.debug("Debug info")
await ctx.warning("Warning message")
await ctx.error("Error message")
# Report progress
await ctx.report_progress(50, 100)
# Access resources
data = await ctx.read_resource("resource://data")
# Get request info
request_id = ctx.request_id
client_id = ctx.client_id
return str(x)
```
The context parameter name can be anything as long as it's annotated with Context.
The context is optional - tools that don't need it can omit the parameter.
"""
_request_context: ServerRequestContext[LifespanContextT, RequestT] | None
_mcp_server: MCPServer | None
def __init__(
self,
*,
request_context: ServerRequestContext[LifespanContextT, RequestT] | None = None,
mcp_server: MCPServer | None = None,
# TODO(Marcelo): We should drop this kwargs parameter.
**kwargs: Any,
):
super().__init__(**kwargs)
self._request_context = request_context
self._mcp_server = mcp_server
@property
def mcp_server(self) -> MCPServer:
"""Access to the MCPServer instance."""
if self._mcp_server is None: # pragma: no cover
raise ValueError("Context is not available outside of a request")
return self._mcp_server # pragma: no cover
@property
def request_context(self) -> ServerRequestContext[LifespanContextT, RequestT]:
"""Access to the underlying request context."""
if self._request_context is None: # pragma: no cover
raise ValueError("Context is not available outside of a request")
return self._request_context
async def report_progress(self, progress: float, total: float | None = None, message: str | None = None) -> None:
"""Report progress for the current operation.
Args:
progress: Current progress value (e.g., 24)
total: Optional total value (e.g., 100)
message: Optional message (e.g., "Starting render...")
"""
progress_token = self.request_context.meta.get("progress_token") if self.request_context.meta else None
if progress_token is None: # pragma: no cover
return
await self.request_context.session.send_progress_notification(
progress_token=progress_token,
progress=progress,
total=total,
message=message,
related_request_id=self.request_id,
)
async def read_resource(self, uri: str | AnyUrl) -> Iterable[ReadResourceContents]:
"""Read a resource by URI.
Args:
uri: Resource URI to read
Returns:
The resource content as either text or bytes
"""
assert self._mcp_server is not None, "Context is not available outside of a request"
return await self._mcp_server.read_resource(uri)
async def elicit(
self,
message: str,
schema: type[ElicitSchemaModelT],
) -> ElicitationResult[ElicitSchemaModelT]:
"""Elicit information from the client/user.
This method can be used to interactively ask for additional information from the
client within a tool's execution. The client might display the message to the
user and collect a response according to the provided schema. If the client
is an agent, it might decide how to handle the elicitation -- either by asking
the user or automatically generating a response.
Args:
message: Message to present to the user
schema: A Pydantic model class defining the expected response structure.
According to the specification, only primitive types are allowed.
Returns:
An ElicitationResult containing the action taken and the data if accepted
Note:
Check the result.action to determine if the user accepted, declined, or cancelled.
The result.data will only be populated if action is "accept" and validation succeeded.
"""
return await elicit_with_validation(
session=self.request_context.session,
message=message,
schema=schema,
related_request_id=self.request_id,
)
async def elicit_url(
self,
message: str,
url: str,
elicitation_id: str,
) -> UrlElicitationResult:
"""Request URL mode elicitation from the client.
This directs the user to an external URL for out-of-band interactions
that must not pass through the MCP client. Use this for:
- Collecting sensitive credentials (API keys, passwords)
- OAuth authorization flows with third-party services
- Payment and subscription flows
- Any interaction where data should not pass through the LLM context
The response indicates whether the user consented to navigate to the URL.
The actual interaction happens out-of-band. When the elicitation completes,
call `ctx.session.send_elicit_complete(elicitation_id)` to notify the client.
Args:
message: Human-readable explanation of why the interaction is needed
url: The URL the user should navigate to
elicitation_id: Unique identifier for tracking this elicitation
Returns:
UrlElicitationResult indicating accept, decline, or cancel
"""
return await _elicit_url(
session=self.request_context.session,
message=message,
url=url,
elicitation_id=elicitation_id,
related_request_id=self.request_id,
)
async def log(
self,
level: Literal["debug", "info", "warning", "error"],
message: str,
*,
logger_name: str | None = None,
extra: dict[str, Any] | None = None,
) -> None:
"""Send a log message to the client.
Args:
level: Log level (debug, info, warning, error)
message: Log message
logger_name: Optional logger name
extra: Optional dictionary with additional structured data to include
"""
if extra:
log_data = {"message": message, **extra}
else:
log_data = message
await self.request_context.session.send_log_message(
level=level,
data=log_data,
logger=logger_name,
related_request_id=self.request_id,
)
@property
def client_id(self) -> str | None:
"""Get the client ID if available."""
return self.request_context.meta.get("client_id") if self.request_context.meta else None # pragma: no cover
@property
def request_id(self) -> str:
"""Get the unique ID for this request."""
return str(self.request_context.request_id)
@property
def session(self):
"""Access to the underlying session for advanced usage."""
return self.request_context.session
async def close_sse_stream(self) -> None:
"""Close the SSE stream to trigger client reconnection.
This method closes the HTTP connection for the current request, triggering
client reconnection. Events continue to be stored in the event store and will
be replayed when the client reconnects with Last-Event-ID.
Use this to implement polling behavior during long-running operations -
the client will reconnect after the retry interval specified in the priming event.
Note:
This is a no-op if not using StreamableHTTP transport with event_store.
The callback is only available when event_store is configured.
"""
if self._request_context and self._request_context.close_sse_stream: # pragma: no cover
await self._request_context.close_sse_stream()
async def close_standalone_sse_stream(self) -> None:
"""Close the standalone GET SSE stream to trigger client reconnection.
This method closes the HTTP connection for the standalone GET stream used
for unsolicited server-to-client notifications. The client SHOULD reconnect
with Last-Event-ID to resume receiving notifications.
Note:
This is a no-op if not using StreamableHTTP transport with event_store.
Currently, client reconnection for standalone GET streams is NOT
implemented - this is a known gap.
"""
if self._request_context and self._request_context.close_standalone_sse_stream: # pragma: no cover
await self._request_context.close_standalone_sse_stream()
# Convenience methods for common log levels
async def debug(self, message: str, *, logger_name: str | None = None, extra: dict[str, Any] | None = None) -> None:
"""Send a debug log message."""
await self.log("debug", message, logger_name=logger_name, extra=extra)
async def info(self, message: str, *, logger_name: str | None = None, extra: dict[str, Any] | None = None) -> None:
"""Send an info log message."""
await self.log("info", message, logger_name=logger_name, extra=extra)
async def warning(
self, message: str, *, logger_name: str | None = None, extra: dict[str, Any] | None = None
) -> None:
"""Send a warning log message."""
await self.log("warning", message, logger_name=logger_name, extra=extra)
async def error(self, message: str, *, logger_name: str | None = None, extra: dict[str, Any] | None = None) -> None:
"""Send an error log message."""
await self.log("error", message, logger_name=logger_name, extra=extra)

10
src/mcp/server/mcpserver/tools/base.py
View file

@ -17,7 +17,7 @@ from mcp.types import Icon, ToolAnnotations
if TYPE_CHECKING:
from mcp.server.context import LifespanContextT, RequestT
from mcp.server.mcpserver.server import Context
from mcp.server.mcpserver.context import Context
class Tool(BaseModel):
@ -92,10 +92,14 @@ class Tool(BaseModel):
async def run(
self,
arguments: dict[str, Any],
context: Context[LifespanContextT, RequestT] | None = None,
context: Context[LifespanContextT, RequestT],
convert_result: bool = False,
) -> Any:
"""Run the tool with arguments."""
"""Run the tool with arguments.
Raises:
ToolError: If the tool function raises during execution.
"""
try:
result = await self.fn_metadata.call_fn_with_arg_validation(
self.fn,

6
src/mcp/server/mcpserver/tools/tool_manager.py
View file

@ -10,7 +10,7 @@ from mcp.types import Icon, ToolAnnotations
if TYPE_CHECKING:
from mcp.server.context import LifespanContextT, RequestT
from mcp.server.mcpserver.server import Context
from mcp.server.mcpserver.context import Context
logger = get_logger(__name__)
@ -81,7 +81,7 @@ class ToolManager:
self,
name: str,
arguments: dict[str, Any],
context: Context[LifespanContextT, RequestT] | None = None,
context: Context[LifespanContextT, RequestT],
convert_result: bool = False,
) -> Any:
"""Call a tool by name with arguments."""
@ -89,4 +89,4 @@ class ToolManager:
if not tool:
raise ToolError(f"Unknown tool: {name}")
return await tool.run(arguments, context=context, convert_result=convert_result)
return await tool.run(arguments, context, convert_result=convert_result)

4
src/mcp/server/mcpserver/utilities/context_injection.py
View file

@ -7,6 +7,8 @@ import typing
from collections.abc import Callable
from typing import Any
from mcp.server.mcpserver.context import Context
def find_context_parameter(fn: Callable[..., Any]) -> str | None:
"""Find the parameter that should receive the Context object.
@ -20,8 +22,6 @@ def find_context_parameter(fn: Callable[..., Any]) -> str | None:
Returns:
The name of the context parameter, or None if not found
"""
from mcp.server.mcpserver.server import Context
# Get type hints to properly resolve string annotations
try:
hints = typing.get_type_hints(fn)