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

docs: fix docstrings across public API surface (#2095)

Browse source
This commit is contained in:
Jonathan Hefner 2026-02-18 23:45:59 -06:00 • committed by GitHub
commit c0328540c9
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
16 changed files with 177 additions and 163 deletions
Download patch file Download diff file Expand all files Collapse all files

2
src/mcp/server/auth/handlers/revoke.py
View file

@ -15,7 +15,7 @@ from mcp.server.auth.provider import AccessToken, OAuthAuthorizationServerProvid
class RevocationRequest(BaseModel):
"""# See https://datatracker.ietf.org/doc/html/rfc7009#section-2.1"""
"""See https://datatracker.ietf.org/doc/html/rfc7009#section-2.1"""
token: str
token_type_hint: Literal["access_token", "refresh_token"] | None = None

8
src/mcp/server/auth/middleware/client_auth.py
View file

@ -19,15 +19,17 @@ class AuthenticationError(Exception):
class ClientAuthenticator:
"""ClientAuthenticator is a callable which validates requests from a client
application, used to verify /token calls.
If, during registration, the client requested to be issued a secret, the
authenticator asserts that /token calls must be authenticated with
that same token.
that same secret.
NOTE: clients can opt for no authentication during registration, in which case this
logic is skipped.
"""
def __init__(self, provider: OAuthAuthorizationServerProvider[Any, Any, Any]):
"""Initialize the dependency.
"""Initialize the authenticator.
Args:
provider: Provider to look up client information
@ -83,7 +85,7 @@ class ClientAuthenticator:
elif client.token_endpoint_auth_method == "client_secret_post":
raw_form_data = form_data.get("client_secret")
# form_data.get() can return a UploadFile or None, so we need to check if it's a string
# form_data.get() can return an UploadFile or None, so we need to check if it's a string
if isinstance(raw_form_data, str):
request_client_secret = str(raw_form_data)

17
src/mcp/server/auth/provider.py
View file

@ -131,8 +131,9 @@ class OAuthAuthorizationServerProvider(Protocol, Generic[AuthorizationCodeT, Ref
"""
async def authorize(self, client: OAuthClientInformationFull, params: AuthorizationParams) -> str:
"""Called as part of the /authorize endpoint, and returns a URL that the client
"""Handle the /authorize endpoint and return a URL that the client
will be redirected to.
Many MCP implementations will redirect to a third-party provider to perform
a second OAuth exchange with that provider. In this sort of setup, the client
has an OAuth connection with the MCP server, and the MCP server has an OAuth
@ -151,7 +152,7 @@ class OAuthAuthorizationServerProvider(Protocol, Generic[AuthorizationCodeT, Ref
| |
+------------+
Implementations will need to define another handler on the MCP server return
Implementations will need to define another handler on the MCP server's return
flow to perform the second redirect, and generate and store an authorization
code as part of completing the OAuth authorization step.
@ -182,7 +183,7 @@ class OAuthAuthorizationServerProvider(Protocol, Generic[AuthorizationCodeT, Ref
authorization_code: The authorization code to get the challenge for.
Returns:
The AuthorizationCode, or None if not found
The AuthorizationCode, or None if not found.
"""
...
@ -199,7 +200,7 @@ class OAuthAuthorizationServerProvider(Protocol, Generic[AuthorizationCodeT, Ref
The OAuth token, containing access and refresh tokens.
Raises:
TokenError: If the request is invalid
TokenError: If the request is invalid.
"""
...
@ -234,18 +235,18 @@ class OAuthAuthorizationServerProvider(Protocol, Generic[AuthorizationCodeT, Ref
The OAuth token, containing access and refresh tokens.
Raises:
TokenError: If the request is invalid
TokenError: If the request is invalid.
"""
...
async def load_access_token(self, token: str) -> AccessTokenT | None:
"""Loads an access token by its token.
"""Loads an access token by its token string.
Args:
token: The access token to verify.
Returns:
The AuthInfo, or None if the token is invalid.
The access token, or None if the token is invalid.
"""
async def revoke_token(
@ -261,7 +262,7 @@ class OAuthAuthorizationServerProvider(Protocol, Generic[AuthorizationCodeT, Ref
provided.
Args:
token: the token to revoke
token: The token to revoke.
"""

6
src/mcp/server/auth/routes.py
View file

@ -25,10 +25,10 @@ def validate_issuer_url(url: AnyHttpUrl):
"""Validate that the issuer URL meets OAuth 2.0 requirements.
Args:
url: The issuer URL to validate
url: The issuer URL to validate.
Raises:
ValueError: If the issuer URL is invalid
ValueError: If the issuer URL is invalid.
"""
# RFC 8414 requires HTTPS, but we allow loopback/localhost HTTP for testing
@ -213,6 +213,8 @@ def create_protected_resource_routes(
resource_url: The URL of this resource server
authorization_servers: List of authorization servers that can issue tokens
scopes_supported: Optional list of scopes supported by this resource
resource_name: Optional human-readable name for this resource
resource_documentation: Optional URL to documentation for this resource
Returns:
List of Starlette routes for protected resource metadata

4
src/mcp/server/elicitation.py
View file

@ -112,8 +112,8 @@ async def elicit_with_validation(
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. Or in case a
client is an agent, it might decide how to handle the elicitation -- either by asking
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.
For sensitive data like credentials or OAuth flows, use elicit_url() instead.

8
src/mcp/server/experimental/request_context.py
View file

@ -62,8 +62,8 @@ class Experimental:
"""Validate that the request is compatible with the tool's task execution mode.
Per MCP spec:
- "required": Clients MUST invoke as task. Server returns -32601 if not.
- "forbidden" (or None): Clients MUST NOT invoke as task. Server returns -32601 if they do.
- "required": Clients MUST invoke as a task. Server returns -32601 if not.
- "forbidden" (or None): Clients MUST NOT invoke as a task. Server returns -32601 if they do.
- "optional": Either is acceptable.
Args:
@ -111,7 +111,7 @@ class Experimental:
"""Check if this client can use a tool with the given task mode.
Useful for filtering tool lists or providing warnings.
Returns False if tool requires "required" but client doesn't support tasks.
Returns False if the tool's task mode is "required" but the client doesn't support tasks.
Args:
tool_task_mode: The tool's execution.taskSupport value
@ -164,7 +164,7 @@ class Experimental:
async def work(task: ServerTaskContext) -> CallToolResult:
result = await task.elicit(
message="Are you sure?",
requestedSchema={"type": "object", ...}
requested_schema={"type": "object", ...}
)
confirmed = result.content.get("confirm", False)
return CallToolResult(content=[TextContent(text="Done" if confirmed else "Cancelled")])

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

@ -88,9 +88,6 @@ class NotificationOptions:
async def lifespan(_: Server[LifespanResultT]) -> AsyncIterator[dict[str, Any]]:
"""Default lifespan context manager that does nothing.
Args:
server: The server instance this lifespan is managing
Returns:
An empty context object
"""

2
src/mcp/server/mcpserver/resources/types.py
View file

@ -109,7 +109,7 @@ class FunctionResource(Resource):
class FileResource(Resource):
"""A resource that reads from a file.
Set is_binary=True to read file as binary data instead of text.
Set is_binary=True to read the file as binary data instead of text.
"""
path: Path = Field(description="Path to the file")

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

@ -108,7 +108,7 @@ class Settings(BaseSettings, Generic[LifespanResultT]):
warn_on_duplicate_prompts: bool
lifespan: Callable[[MCPServer[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]] | None
"""A async context manager that will be called when the server is started."""
"""An async context manager that will be called when the server is started."""
auth: AuthSettings | None
@ -388,8 +388,10 @@ class MCPServer(Generic[LifespanResultT]):
]
def get_context(self) -> Context[LifespanResultT, Request]:
"""Returns a Context object. Note that the context will only be valid
during a request; outside a request, most methods will error.
"""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()
@ -475,6 +477,8 @@ class MCPServer(Generic[LifespanResultT]):
title: Optional human-readable title for the tool
description: Optional description of what the tool does
annotations: Optional ToolAnnotations providing additional tool information
icons: Optional list of icons for the tool
meta: Optional metadata dictionary for the tool
structured_output: Controls whether the tool's output is structured or unstructured
- If None, auto-detects based on the function's return type annotation
- If True, creates a structured tool (return type annotation permitting)
@ -523,6 +527,8 @@ class MCPServer(Generic[LifespanResultT]):
title: Optional human-readable title for the tool
description: Optional description of what the tool does
annotations: Optional ToolAnnotations providing additional tool information
icons: Optional list of icons for the tool
meta: Optional metadata dictionary for the tool
structured_output: Controls whether the tool's output is structured or unstructured
- If None, auto-detects based on the function's return type annotation
- If True, creates a structured tool (return type annotation permitting)
@ -534,8 +540,8 @@ class MCPServer(Generic[LifespanResultT]):
return str(x)
@server.tool()
def tool_with_context(x: int, ctx: Context) -> str:
ctx.info(f"Processing {x}")
async def tool_with_context(x: int, ctx: Context) -> str:
await ctx.info(f"Processing {x}")
return str(x)
@server.tool()
@ -636,6 +642,8 @@ class MCPServer(Generic[LifespanResultT]):
title: Optional human-readable title for the resource
description: Optional description of the resource
mime_type: Optional MIME type for the resource
icons: Optional list of icons for the resource
annotations: Optional annotations for the resource
meta: Optional metadata dictionary for the resource
Example:
@ -644,7 +652,7 @@ class MCPServer(Generic[LifespanResultT]):
return "Hello, world!"
@server.resource("resource://my-resource")
async get_data() -> str:
async def get_data() -> str:
data = await fetch_data()
return f"Hello, world! {data}"
@ -736,6 +744,7 @@ class MCPServer(Generic[LifespanResultT]):
name: Optional name for the prompt (defaults to function name)
title: Optional human-readable title for the prompt
description: Optional description of what the prompt does
icons: Optional list of icons for the prompt
Example:
@server.prompt()
@ -1092,18 +1101,18 @@ class Context(BaseModel, Generic[LifespanContextT, RequestT]):
```python
@server.tool()
def my_tool(x: int, ctx: Context) -> str:
async def my_tool(x: int, ctx: Context) -> str:
# Log messages to the client
ctx.info(f"Processing {x}")
ctx.debug("Debug info")
ctx.warning("Warning message")
ctx.error("Error message")
await ctx.info(f"Processing {x}")
await ctx.debug("Debug info")
await ctx.warning("Warning message")
await ctx.error("Error message")
# Report progress
ctx.report_progress(50, 100)
await ctx.report_progress(50, 100)
# Access resources
data = ctx.read_resource("resource://data")
data = await ctx.read_resource("resource://data")
# Get request info
request_id = ctx.request_id
@ -1149,9 +1158,9 @@ class Context(BaseModel, Generic[LifespanContextT, RequestT]):
"""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: 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
@ -1187,15 +1196,14 @@ class Context(BaseModel, Generic[LifespanContextT, RequestT]):
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. Or in case a
client is an agent, it might decide how to handle the elicitation -- either by asking
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:
schema: A Pydantic model class defining the expected response structure, according to the specification,
only primitive types are allowed.
message: Optional message to present to the user. If not provided, will use
a default message based on the schema
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
@ -1229,7 +1237,7 @@ class Context(BaseModel, Generic[LifespanContextT, RequestT]):
The response indicates whether the user consented to navigate to the URL.
The actual interaction happens out-of-band. When the elicitation completes,
call `self.session.send_elicit_complete(elicitation_id)` to notify the client.
call `ctx.session.send_elicit_complete(elicitation_id)` to notify the client.
Args:
message: Human-readable explanation of why the interaction is needed
@ -1299,7 +1307,7 @@ class Context(BaseModel, Generic[LifespanContextT, RequestT]):
be replayed when the client reconnects with Last-Event-ID.
Use this to implement polling behavior during long-running operations -
client will reconnect after the retry interval specified in the priming event.
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.

30
src/mcp/server/mcpserver/utilities/func_metadata.py
View file

@ -46,7 +46,7 @@ class ArgModelBase(BaseModel):
def model_dump_one_level(self) -> dict[str, Any]:
"""Return a dict of the model's fields, one level deep.
That is, sub-models etc are not dumped - they are kept as pydantic models.
That is, sub-models etc are not dumped - they are kept as Pydantic models.
"""
kwargs: dict[str, Any] = {}
for field_name, field_info in self.__class__.model_fields.items():
@ -89,8 +89,7 @@ class FuncMetadata(BaseModel):
return await anyio.to_thread.run_sync(functools.partial(fn, **arguments_parsed_dict))
def convert_result(self, result: Any) -> Any:
"""Convert the result of a function call to the appropriate format for
the lowlevel server tool call handler:
"""Convert a function call result to the format for the lowlevel tool call handler.
- If output_model is None, return the unstructured content directly.
- If output_model is not None, convert the result to structured output format
@ -126,11 +125,11 @@ class FuncMetadata(BaseModel):
def pre_parse_json(self, data: dict[str, Any]) -> dict[str, Any]:
"""Pre-parse data from JSON.
Return a dict with same keys as input but with values parsed from JSON
Return a dict with the same keys as input but with values parsed from JSON
if appropriate.
This is to handle cases like `["a", "b", "c"]` being passed in as JSON inside
a string rather than an actual list. Claude desktop is prone to this - in fact
a string rather than an actual list. Claude Desktop is prone to this - in fact
it seems incapable of NOT doing this. For sub-models, it tends to pass
dicts (JSON objects) as JSON strings, which can be pre-parsed here.
"""
@ -173,8 +172,7 @@ def func_metadata(
skip_names: Sequence[str] = (),
structured_output: bool | None = None,
) -> FuncMetadata:
"""Given a function, return metadata including a pydantic model representing its
signature.
"""Given a function, return metadata including a Pydantic model representing its signature.
The use case for this is
```
@ -183,11 +181,11 @@ def func_metadata(
return func(**validated_args.model_dump_one_level())
```
**critically** it also provides pre-parse helper to attempt to parse things from
**critically** it also provides a pre-parse helper to attempt to parse things from
JSON.
Args:
func: The function to convert to a pydantic model
func: The function to convert to a Pydantic model
skip_names: A list of parameter names to skip. These will not be included in
the model.
structured_output: Controls whether the tool's output is structured or unstructured
@ -195,8 +193,8 @@ def func_metadata(
- If True, creates a structured tool (return type annotation permitting)
- If False, unconditionally creates an unstructured tool
If structured, creates a Pydantic model for the function's result based on its annotation.
Supports various return types:
If structured, creates a Pydantic model for the function's result based on its annotation.
Supports various return types:
- BaseModel subclasses (used directly)
- Primitive types (str, int, float, bool, bytes, None) - wrapped in a
model with a 'result' field
@ -206,9 +204,9 @@ def func_metadata(
Returns:
A FuncMetadata object containing:
- arg_model: A pydantic model representing the function's arguments
- output_model: A pydantic model for the return type if output is structured
- output_conversion: Records how function output should be converted before returning.
- arg_model: A Pydantic model representing the function's arguments
- output_model: A Pydantic model for the return type if the output is structured
- wrap_output: Whether the function result needs to be wrapped in `{"result": ...}` for structured output.
"""
try:
sig = inspect.signature(func, eval_str=True)
@ -296,7 +294,7 @@ def func_metadata(
] # pragma: no cover
else:
# We only had `Annotated[CallToolResult, ReturnType]`, treat the original annotation
# as beging `ReturnType`:
# as being `ReturnType`:
original_annotation = return_type_expr
else:
return FuncMetadata(arg_model=arguments_model)
@ -355,7 +353,7 @@ def _try_create_model_and_schema(
if origin is dict:
args = get_args(type_expr)
if len(args) == 2 and args[0] is str:
# TODO: should we use the original annotation? We are loosing any potential `Annotated`
# TODO: should we use the original annotation? We are losing any potential `Annotated`
# metadata for Pydantic here:
model = _create_dict_model(func_name, type_expr)
else:

6
src/mcp/server/mcpserver/utilities/logging.py
View file

@ -8,10 +8,10 @@ def get_logger(name: str) -> logging.Logger:
"""Get a logger nested under MCP namespace.
Args:
name: the name of the logger
name: The name of the logger.
Returns:
a configured logger instance
A configured logger instance.
"""
return logging.getLogger(name)
@ -22,7 +22,7 @@ def configure_logging(
"""Configure logging for MCP.
Args:
level: the log level to use
level: The log level to use.
"""
handlers: list[logging.Handler] = []
try:

2
src/mcp/server/models.py
View file

@ -1,4 +1,4 @@
"""This module provides simpler types to use with the server for managing prompts
"""This module provides simplified types to use with the server for managing prompts
and tools.
"""

28
src/mcp/server/session.py
View file

@ -363,12 +363,12 @@ class ServerSession(
"""Send a form mode elicitation/create request.
Args:
message: The message to present to the user
requested_schema: Schema defining the expected response structure
related_request_id: Optional ID of the request that triggered this elicitation
message: The message to present to the user.
requested_schema: Schema defining the expected response structure.
related_request_id: Optional ID of the request that triggered this elicitation.
Returns:
The client's response
The client's response.
Note:
This method is deprecated in favor of elicit_form(). It remains for
@ -385,12 +385,12 @@ class ServerSession(
"""Send a form mode elicitation/create request.
Args:
message: The message to present to the user
requested_schema: Schema defining the expected response structure
related_request_id: Optional ID of the request that triggered this elicitation
message: The message to present to the user.
requested_schema: Schema defining the expected response structure.
related_request_id: Optional ID of the request that triggered this elicitation.
Returns:
The client's response with form data
The client's response with form data.
Raises:
StatelessModeNotSupported: If called in stateless HTTP mode.
@ -421,13 +421,13 @@ class ServerSession(
like OAuth flows, credential collection, or payment processing.
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
related_request_id: Optional ID of the request that triggered this elicitation
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.
related_request_id: Optional ID of the request that triggered this elicitation.
Returns:
The client's response indicating acceptance, decline, or cancellation
The client's response indicating acceptance, decline, or cancellation.
Raises:
StatelessModeNotSupported: If called in stateless HTTP mode.
@ -499,7 +499,7 @@ class ServerSession(
Args:
elicitation_id: The unique identifier of the completed elicitation
related_request_id: Optional ID of the request that triggered this
related_request_id: Optional ID of the request that triggered this notification
"""
await self.send_notification(
types.ElicitCompleteNotification(

8
src/mcp/server/sse.py
View file

@ -29,8 +29,8 @@ Example usage:
uvicorn.run(starlette_app, host="127.0.0.1", port=port)
```
Note: The handle_sse function must return a Response to avoid a "TypeError: 'NoneType'
object is not callable" error when client disconnects. The example above returns
Note: The handle_sse function must return a Response to avoid a
"TypeError: 'NoneType' object is not callable" error when client disconnects. The example above returns
an empty Response() after the SSE connection ends to fix this.
See SseServerTransport class documentation for more details.
@ -61,8 +61,8 @@ logger = logging.getLogger(__name__)
class SseServerTransport:
"""SSE server transport for MCP. This class provides _two_ ASGI applications,
suitable to be used with a framework like Starlette and a server like Hypercorn:
"""SSE server transport for MCP. This class provides two ASGI applications,
suitable for use with a framework like Starlette and a server like Hypercorn:
1. connect_sse() is an ASGI application which receives incoming GET requests,
and sets up a new SSE stream to send server messages to the client.

15
src/mcp/server/streamable_http.py
View file

@ -89,7 +89,7 @@ class EventStore(ABC):
message: The JSON-RPC message to store, or None for priming events
Returns:
The generated event ID for the stored event
The generated event ID for the stored event.
"""
pass # pragma: no cover
@ -106,7 +106,7 @@ class EventStore(ABC):
send_callback: A callback function to send events to the client
Returns:
The stream ID of the replayed events
The stream ID of the replayed events, or None if no events were found.
"""
pass # pragma: no cover
@ -185,7 +185,7 @@ class StreamableHTTPServerTransport:
be replayed when the client reconnects with Last-Event-ID.
Use this to implement polling behavior during long-running operations -
client will reconnect after the retry interval specified in the priming event.
the client will reconnect after the retry interval specified in the priming event.
Args:
request_id: The request ID whose SSE stream should be closed.
@ -213,7 +213,7 @@ class StreamableHTTPServerTransport:
with Last-Event-ID to resume receiving notifications.
Use this to implement polling behavior for the notification stream -
client will reconnect after the retry interval specified in the priming event.
the client will reconnect after the retry interval specified in the priming event.
Note:
This is a no-op if there is no active standalone SSE stream.
@ -316,7 +316,7 @@ class StreamableHTTPServerTransport:
status_code: HTTPStatus = HTTPStatus.OK,
headers: dict[str, str] | None = None,
) -> Response:
"""Create a JSON response from a JSONRPCMessage"""
"""Create a JSON response from a JSONRPCMessage."""
response_headers = {"Content-Type": CONTENT_TYPE_JSON}
if headers: # pragma: lax no cover
response_headers.update(headers)
@ -362,7 +362,7 @@ class StreamableHTTPServerTransport:
self._request_streams.pop(request_id, None)
async def handle_request(self, scope: Scope, receive: Receive, send: Send) -> None:
"""Application entry point that handles all HTTP requests"""
"""Application entry point that handles all HTTP requests."""
request = Request(scope, receive)
# Validate request headers for DNS rebinding protection
@ -536,7 +536,7 @@ class StreamableHTTPServerTransport:
if isinstance(event_message.message, JSONRPCResponse | JSONRPCError):
response_message = event_message.message
break
# For notifications and request, keep waiting
# For notifications and requests, keep waiting
else: # pragma: no cover
logger.debug(f"received: {event_message.message.method}")
@ -860,6 +860,7 @@ class StreamableHTTPServerTransport:
async def _replay_events(self, last_event_id: str, request: Request, send: Send) -> None: # pragma: no cover
"""Replays events that would have been sent after the specified event ID.
Only used when resumability is enabled.
"""
event_store = self._event_store

4
src/mcp/server/websocket.py
View file

@ -12,8 +12,8 @@ from mcp.shared.message import SessionMessage
@asynccontextmanager # pragma: no cover
async def websocket_server(scope: Scope, receive: Receive, send: Send):
"""WebSocket server transport for MCP. This is an ASGI application, suitable to be
used with a framework like Starlette and a server like Hypercorn.
"""WebSocket server transport for MCP. This is an ASGI application, suitable for use
with a framework like Starlette and a server like Hypercorn.
"""
websocket = WebSocket(scope, receive, send)