Index

Server

Bases: Generic[LifespanResultT]

Source code in src/mcp/server/lowlevel/server.py

class Server(Generic[LifespanResultT]):
    def __init__(
        self,
        name: str,
        *,
        version: str | None = None,
        title: str | None = None,
        description: str | None = None,
        instructions: str | None = None,
        website_url: str | None = None,
        icons: list[types.Icon] | None = None,
        lifespan: Callable[
            [Server[LifespanResultT]],
            AbstractAsyncContextManager[LifespanResultT],
        ] = lifespan,
        # Request handlers
        on_list_tools: Callable[
            [ServerRequestContext[LifespanResultT], types.PaginatedRequestParams | None],
            Awaitable[types.ListToolsResult],
        ]
        | None = None,
        on_call_tool: Callable[
            [ServerRequestContext[LifespanResultT], types.CallToolRequestParams],
            Awaitable[types.CallToolResult | types.CreateTaskResult],
        ]
        | None = None,
        on_list_resources: Callable[
            [ServerRequestContext[LifespanResultT], types.PaginatedRequestParams | None],
            Awaitable[types.ListResourcesResult],
        ]
        | None = None,
        on_list_resource_templates: Callable[
            [ServerRequestContext[LifespanResultT], types.PaginatedRequestParams | None],
            Awaitable[types.ListResourceTemplatesResult],
        ]
        | None = None,
        on_read_resource: Callable[
            [ServerRequestContext[LifespanResultT], types.ReadResourceRequestParams],
            Awaitable[types.ReadResourceResult],
        ]
        | None = None,
        on_subscribe_resource: Callable[
            [ServerRequestContext[LifespanResultT], types.SubscribeRequestParams],
            Awaitable[types.EmptyResult],
        ]
        | None = None,
        on_unsubscribe_resource: Callable[
            [ServerRequestContext[LifespanResultT], types.UnsubscribeRequestParams],
            Awaitable[types.EmptyResult],
        ]
        | None = None,
        on_list_prompts: Callable[
            [ServerRequestContext[LifespanResultT], types.PaginatedRequestParams | None],
            Awaitable[types.ListPromptsResult],
        ]
        | None = None,
        on_get_prompt: Callable[
            [ServerRequestContext[LifespanResultT], types.GetPromptRequestParams],
            Awaitable[types.GetPromptResult],
        ]
        | None = None,
        on_completion: Callable[
            [ServerRequestContext[LifespanResultT], types.CompleteRequestParams],
            Awaitable[types.CompleteResult],
        ]
        | None = None,
        on_set_logging_level: Callable[
            [ServerRequestContext[LifespanResultT], types.SetLevelRequestParams],
            Awaitable[types.EmptyResult],
        ]
        | None = None,
        on_ping: Callable[
            [ServerRequestContext[LifespanResultT], types.RequestParams | None],
            Awaitable[types.EmptyResult],
        ] = _ping_handler,
        # Notification handlers
        on_roots_list_changed: Callable[
            [ServerRequestContext[LifespanResultT], types.NotificationParams | None],
            Awaitable[None],
        ]
        | None = None,
        on_progress: Callable[
            [ServerRequestContext[LifespanResultT], types.ProgressNotificationParams],
            Awaitable[None],
        ]
        | None = None,
    ):
        self.name = name
        self.version = version
        self.title = title
        self.description = description
        self.instructions = instructions
        self.website_url = website_url
        self.icons = icons
        self.lifespan = lifespan
        self._request_handlers: dict[str, Callable[[ServerRequestContext[LifespanResultT], Any], Awaitable[Any]]] = {}
        self._notification_handlers: dict[
            str, Callable[[ServerRequestContext[LifespanResultT], Any], Awaitable[None]]
        ] = {}
        self._experimental_handlers: ExperimentalHandlers[LifespanResultT] | None = None
        self._session_manager: StreamableHTTPSessionManager | None = None
        logger.debug("Initializing server %r", name)

        # Populate internal handler dicts from on_* kwargs
        self._request_handlers.update(
            {
                method: handler
                for method, handler in {
                    "ping": on_ping,
                    "prompts/list": on_list_prompts,
                    "prompts/get": on_get_prompt,
                    "resources/list": on_list_resources,
                    "resources/templates/list": on_list_resource_templates,
                    "resources/read": on_read_resource,
                    "resources/subscribe": on_subscribe_resource,
                    "resources/unsubscribe": on_unsubscribe_resource,
                    "tools/list": on_list_tools,
                    "tools/call": on_call_tool,
                    "logging/setLevel": on_set_logging_level,
                    "completion/complete": on_completion,
                }.items()
                if handler is not None
            }
        )

        self._notification_handlers.update(
            {
                method: handler
                for method, handler in {
                    "notifications/roots/list_changed": on_roots_list_changed,
                    "notifications/progress": on_progress,
                }.items()
                if handler is not None
            }
        )

    def _add_request_handler(
        self,
        method: str,
        handler: Callable[[ServerRequestContext[LifespanResultT], Any], Awaitable[Any]],
    ) -> None:
        """Add a request handler, silently replacing any existing handler for the same method."""
        self._request_handlers[method] = handler

    def _has_handler(self, method: str) -> bool:
        """Check if a handler is registered for the given method."""
        return method in self._request_handlers or method in self._notification_handlers

    # TODO: Rethink capabilities API. Currently capabilities are derived from registered
    # handlers but require NotificationOptions to be passed externally for list_changed
    # flags, and experimental_capabilities as a separate dict. Consider deriving capabilities
    # entirely from server state (e.g. constructor params for list_changed) instead of
    # requiring callers to assemble them at create_initialization_options() time.
    def create_initialization_options(
        self,
        notification_options: NotificationOptions | None = None,
        experimental_capabilities: dict[str, dict[str, Any]] | None = None,
    ) -> InitializationOptions:
        """Create initialization options from this server instance."""

        def pkg_version(package: str) -> str:
            try:
                return importlib_version(package)
            except Exception:  # pragma: no cover
                pass

            return "unknown"  # pragma: no cover

        return InitializationOptions(
            server_name=self.name,
            server_version=self.version if self.version else pkg_version("mcp"),
            title=self.title,
            description=self.description,
            capabilities=self.get_capabilities(
                notification_options or NotificationOptions(),
                experimental_capabilities or {},
            ),
            instructions=self.instructions,
            website_url=self.website_url,
            icons=self.icons,
        )

    def get_capabilities(
        self,
        notification_options: NotificationOptions,
        experimental_capabilities: dict[str, dict[str, Any]],
    ) -> types.ServerCapabilities:
        """Convert existing handlers to a ServerCapabilities object."""
        prompts_capability = None
        resources_capability = None
        tools_capability = None
        logging_capability = None
        completions_capability = None

        # Set prompt capabilities if handler exists
        if "prompts/list" in self._request_handlers:
            prompts_capability = types.PromptsCapability(list_changed=notification_options.prompts_changed)

        # Set resource capabilities if handler exists
        if "resources/list" in self._request_handlers:
            resources_capability = types.ResourcesCapability(
                subscribe="resources/subscribe" in self._request_handlers,
                list_changed=notification_options.resources_changed,
            )

        # Set tool capabilities if handler exists
        if "tools/list" in self._request_handlers:
            tools_capability = types.ToolsCapability(list_changed=notification_options.tools_changed)

        # Set logging capabilities if handler exists
        if "logging/setLevel" in self._request_handlers:
            logging_capability = types.LoggingCapability()

        # Set completions capabilities if handler exists
        if "completion/complete" in self._request_handlers:
            completions_capability = types.CompletionsCapability()

        capabilities = types.ServerCapabilities(
            prompts=prompts_capability,
            resources=resources_capability,
            tools=tools_capability,
            logging=logging_capability,
            experimental=experimental_capabilities,
            completions=completions_capability,
        )
        if self._experimental_handlers:
            self._experimental_handlers.update_capabilities(capabilities)
        return capabilities

    @property
    def experimental(self) -> ExperimentalHandlers[LifespanResultT]:
        """Experimental APIs for tasks and other features.

        WARNING: These APIs are experimental and may change without notice.
        """

        # We create this inline so we only add these capabilities _if_ they're actually used
        if self._experimental_handlers is None:
            self._experimental_handlers = ExperimentalHandlers(
                add_request_handler=self._add_request_handler,
                has_handler=self._has_handler,
            )
        return self._experimental_handlers

    @property
    def session_manager(self) -> StreamableHTTPSessionManager:
        """Get the StreamableHTTP session manager.

        Raises:
            RuntimeError: If called before streamable_http_app() has been called.
        """
        if self._session_manager is None:  # pragma: no cover
            raise RuntimeError(
                "Session manager can only be accessed after calling streamable_http_app(). "
                "The session manager is created lazily to avoid unnecessary initialization."
            )
        return self._session_manager  # pragma: no cover

    async def run(
        self,
        read_stream: ReadStream[SessionMessage | Exception],
        write_stream: WriteStream[SessionMessage],
        initialization_options: InitializationOptions,
        # When False, exceptions are returned as messages to the client.
        # When True, exceptions are raised, which will cause the server to shut down
        # but also make tracing exceptions much easier during testing and when using
        # in-process servers.
        raise_exceptions: bool = False,
        # When True, the server is stateless and
        # clients can perform initialization with any node. The client must still follow
        # the initialization lifecycle, but can do so with any available node
        # rather than requiring initialization for each connection.
        stateless: bool = False,
    ):
        async with AsyncExitStack() as stack:
            lifespan_context = await stack.enter_async_context(self.lifespan(self))
            session = await stack.enter_async_context(
                ServerSession(
                    read_stream,
                    write_stream,
                    initialization_options,
                    stateless=stateless,
                )
            )

            # Configure task support for this session if enabled
            task_support = self._experimental_handlers.task_support if self._experimental_handlers else None
            if task_support is not None:
                task_support.configure_session(session)
                await stack.enter_async_context(task_support.run())

            async with anyio.create_task_group() as tg:
                try:
                    async for message in session.incoming_messages:
                        logger.debug("Received message: %s", message)

                        if isinstance(message, RequestResponder) and message.context is not None:
                            context = message.context
                        else:
                            context = contextvars.copy_context()

                        context.run(
                            tg.start_soon,
                            self._handle_message,
                            message,
                            session,
                            lifespan_context,
                            raise_exceptions,
                        )
                finally:
                    # Transport closed: cancel in-flight handlers. Without this the
                    # TG join waits for them, and when they eventually try to
                    # respond they hit a closed write stream (the session's
                    # _receive_loop closed it when the read stream ended).
                    tg.cancel_scope.cancel()

    async def _handle_message(
        self,
        message: RequestResponder[types.ClientRequest, types.ServerResult] | types.ClientNotification | Exception,
        session: ServerSession,
        lifespan_context: LifespanResultT,
        raise_exceptions: bool = False,
    ):
        with warnings.catch_warnings(record=True) as w:
            match message:
                case RequestResponder() as responder:
                    with responder:
                        await self._handle_request(
                            message, responder.request, session, lifespan_context, raise_exceptions
                        )
                case Exception():
                    logger.error(f"Received exception from stream: {message}")
                    if raise_exceptions:
                        raise message
                case _:
                    await self._handle_notification(message, session, lifespan_context)

            for warning in w:  # pragma: lax no cover
                logger.info("Warning: %s: %s", warning.category.__name__, warning.message)

    async def _handle_request(
        self,
        message: RequestResponder[types.ClientRequest, types.ServerResult],
        req: types.ClientRequest,
        session: ServerSession,
        lifespan_context: LifespanResultT,
        raise_exceptions: bool,
    ):
        logger.info("Processing request of type %s", type(req).__name__)

        target = getattr(req.params, "name", None) if req.params else None
        span_name = f"MCP handle {req.method} {target}" if target else f"MCP handle {req.method}"

        # Extract W3C trace context from _meta (SEP-414).
        meta = cast(dict[str, Any] | None, getattr(req.params, "meta", None)) if req.params else None
        parent_context = extract_trace_context(meta) if meta is not None else None

        with otel_span(
            span_name,
            kind=SpanKind.SERVER,
            attributes={"mcp.method.name": req.method, "jsonrpc.request.id": message.request_id},
            context=parent_context,
        ) as span:
            if handler := self._request_handlers.get(req.method):
                logger.debug("Dispatching request of type %s", type(req).__name__)

                try:
                    # Extract request context and close_sse_stream from message metadata
                    request_data = None
                    close_sse_stream_cb = None
                    close_standalone_sse_stream_cb = None
                    if message.message_metadata is not None and isinstance(
                        message.message_metadata, ServerMessageMetadata
                    ):
                        request_data = message.message_metadata.request_context
                        close_sse_stream_cb = message.message_metadata.close_sse_stream
                        close_standalone_sse_stream_cb = message.message_metadata.close_standalone_sse_stream

                    client_capabilities = session.client_params.capabilities if session.client_params else None
                    task_support = self._experimental_handlers.task_support if self._experimental_handlers else None
                    # Get task metadata from request params if present
                    task_metadata = None
                    if hasattr(req, "params") and req.params is not None:  # pragma: no branch
                        task_metadata = getattr(req.params, "task", None)
                    ctx = ServerRequestContext(
                        request_id=message.request_id,
                        meta=message.request_meta,
                        session=session,
                        lifespan_context=lifespan_context,
                        experimental=Experimental(
                            task_metadata=task_metadata,
                            _client_capabilities=client_capabilities,
                            _session=session,
                            _task_support=task_support,
                        ),
                        request=request_data,
                        close_sse_stream=close_sse_stream_cb,
                        close_standalone_sse_stream=close_standalone_sse_stream_cb,
                    )
                    response = await handler(ctx, req.params)
                except MCPError as err:
                    response = err.error
                except anyio.get_cancelled_exc_class():
                    if message.cancelled:
                        # Client sent CancelledNotification; responder.cancel() already
                        # sent an error response, so skip the duplicate.
                        logger.info("Request %s cancelled - duplicate response suppressed", message.request_id)
                        return
                    # Transport-close cancellation from the TG in run(); re-raise so the
                    # TG swallows its own cancellation.
                    raise
                except Exception as err:
                    if raise_exceptions:  # pragma: no cover
                        raise err
                    response = types.ErrorData(code=0, message=str(err))
            else:  # pragma: no cover
                response = types.ErrorData(code=types.METHOD_NOT_FOUND, message="Method not found")

            if isinstance(response, types.ErrorData) and span is not None:
                span.set_status(StatusCode.ERROR, response.message)

            try:
                await message.respond(response)
            except (anyio.BrokenResourceError, anyio.ClosedResourceError):
                # Transport closed between handler unblocking and respond. Happens
                # when _receive_loop's finally wakes a handler blocked on
                # send_request: the handler runs to respond() before run()'s TG
                # cancel fires, but after the write stream closed. Closed if our
                # end closed (_receive_loop's async-with exit); Broken if the peer
                # end closed first (streamable_http terminate()).
                logger.debug("Response for %s dropped - transport closed", message.request_id)
                return

            logger.debug("Response sent")

    async def _handle_notification(
        self,
        notify: types.ClientNotification,
        session: ServerSession,
        lifespan_context: LifespanResultT,
    ) -> None:
        if handler := self._notification_handlers.get(notify.method):
            logger.debug("Dispatching notification of type %s", type(notify).__name__)

            try:
                client_capabilities = session.client_params.capabilities if session.client_params else None
                task_support = self._experimental_handlers.task_support if self._experimental_handlers else None
                ctx = ServerRequestContext(
                    session=session,
                    lifespan_context=lifespan_context,
                    experimental=Experimental(
                        task_metadata=None,
                        _client_capabilities=client_capabilities,
                        _session=session,
                        _task_support=task_support,
                    ),
                )
                await handler(ctx, notify.params)
            except Exception:  # pragma: no cover
                logger.exception("Uncaught exception in notification handler")

    def streamable_http_app(
        self,
        *,
        streamable_http_path: str = "/mcp",
        json_response: bool = False,
        stateless_http: bool = False,
        event_store: EventStore | None = None,
        retry_interval: int | None = None,
        transport_security: TransportSecuritySettings | None = None,
        host: str = "127.0.0.1",
        auth: AuthSettings | None = None,
        token_verifier: TokenVerifier | None = None,
        auth_server_provider: OAuthAuthorizationServerProvider[Any, Any, Any] | None = None,
        custom_starlette_routes: list[Route] | None = None,
        debug: bool = False,
    ) -> Starlette:
        """Return an instance of the StreamableHTTP server app."""
        # Auto-enable DNS rebinding protection for localhost (IPv4 and IPv6)
        if transport_security is None and host in ("127.0.0.1", "localhost", "::1"):
            transport_security = TransportSecuritySettings(
                enable_dns_rebinding_protection=True,
                allowed_hosts=["127.0.0.1:*", "localhost:*", "[::1]:*"],
                allowed_origins=["http://127.0.0.1:*", "http://localhost:*", "http://[::1]:*"],
            )

        session_manager = StreamableHTTPSessionManager(
            app=self,
            event_store=event_store,
            retry_interval=retry_interval,
            json_response=json_response,
            stateless=stateless_http,
            security_settings=transport_security,
        )
        self._session_manager = session_manager

        # Create the ASGI handler
        streamable_http_app = StreamableHTTPASGIApp(session_manager)

        # Create routes
        routes: list[Route | Mount] = []
        middleware: list[Middleware] = []
        required_scopes: list[str] = []

        # Set up auth if configured
        if auth:  # pragma: no cover
            required_scopes = auth.required_scopes or []

            # Add auth middleware if token verifier is available
            if token_verifier:
                middleware = [
                    Middleware(
                        AuthenticationMiddleware,
                        backend=BearerAuthBackend(token_verifier),
                    ),
                    Middleware(AuthContextMiddleware),
                ]

            # Add auth endpoints if auth server provider is configured
            if auth_server_provider:
                routes.extend(
                    create_auth_routes(
                        provider=auth_server_provider,
                        issuer_url=auth.issuer_url,
                        service_documentation_url=auth.service_documentation_url,
                        client_registration_options=auth.client_registration_options,
                        revocation_options=auth.revocation_options,
                    )
                )

        # Set up routes with or without auth
        if token_verifier:  # pragma: no cover
            # Determine resource metadata URL
            resource_metadata_url = None
            if auth and auth.resource_server_url:
                # Build compliant metadata URL for WWW-Authenticate header
                resource_metadata_url = build_resource_metadata_url(auth.resource_server_url)

            routes.append(
                Route(
                    streamable_http_path,
                    endpoint=RequireAuthMiddleware(streamable_http_app, required_scopes, resource_metadata_url),
                )
            )
        else:
            # Auth is disabled, no wrapper needed
            routes.append(
                Route(
                    streamable_http_path,
                    endpoint=streamable_http_app,
                )
            )

        # Add protected resource metadata endpoint if configured as RS
        if auth and auth.resource_server_url:  # pragma: no cover
            routes.extend(
                create_protected_resource_routes(
                    resource_url=auth.resource_server_url,
                    authorization_servers=[auth.issuer_url],
                    scopes_supported=auth.required_scopes,
                )
            )

        if custom_starlette_routes:  # pragma: no cover
            routes.extend(custom_starlette_routes)

        return Starlette(
            debug=debug,
            routes=routes,
            middleware=middleware,
            lifespan=lambda app: session_manager.run(),
        )

create_initialization_options

create_initialization_options(
    notification_options: NotificationOptions | None = None,
    experimental_capabilities: (
        dict[str, dict[str, Any]] | None
    ) = None,
) -> InitializationOptions

Create initialization options from this server instance.

Source code in src/mcp/server/lowlevel/server.py

def create_initialization_options(
    self,
    notification_options: NotificationOptions | None = None,
    experimental_capabilities: dict[str, dict[str, Any]] | None = None,
) -> InitializationOptions:
    """Create initialization options from this server instance."""

    def pkg_version(package: str) -> str:
        try:
            return importlib_version(package)
        except Exception:  # pragma: no cover
            pass

        return "unknown"  # pragma: no cover

    return InitializationOptions(
        server_name=self.name,
        server_version=self.version if self.version else pkg_version("mcp"),
        title=self.title,
        description=self.description,
        capabilities=self.get_capabilities(
            notification_options or NotificationOptions(),
            experimental_capabilities or {},
        ),
        instructions=self.instructions,
        website_url=self.website_url,
        icons=self.icons,
    )

get_capabilities

get_capabilities(
    notification_options: NotificationOptions,
    experimental_capabilities: dict[str, dict[str, Any]],
) -> ServerCapabilities

Convert existing handlers to a ServerCapabilities object.

Source code in src/mcp/server/lowlevel/server.py

def get_capabilities(
    self,
    notification_options: NotificationOptions,
    experimental_capabilities: dict[str, dict[str, Any]],
) -> types.ServerCapabilities:
    """Convert existing handlers to a ServerCapabilities object."""
    prompts_capability = None
    resources_capability = None
    tools_capability = None
    logging_capability = None
    completions_capability = None

    # Set prompt capabilities if handler exists
    if "prompts/list" in self._request_handlers:
        prompts_capability = types.PromptsCapability(list_changed=notification_options.prompts_changed)

    # Set resource capabilities if handler exists
    if "resources/list" in self._request_handlers:
        resources_capability = types.ResourcesCapability(
            subscribe="resources/subscribe" in self._request_handlers,
            list_changed=notification_options.resources_changed,
        )

    # Set tool capabilities if handler exists
    if "tools/list" in self._request_handlers:
        tools_capability = types.ToolsCapability(list_changed=notification_options.tools_changed)

    # Set logging capabilities if handler exists
    if "logging/setLevel" in self._request_handlers:
        logging_capability = types.LoggingCapability()

    # Set completions capabilities if handler exists
    if "completion/complete" in self._request_handlers:
        completions_capability = types.CompletionsCapability()

    capabilities = types.ServerCapabilities(
        prompts=prompts_capability,
        resources=resources_capability,
        tools=tools_capability,
        logging=logging_capability,
        experimental=experimental_capabilities,
        completions=completions_capability,
    )
    if self._experimental_handlers:
        self._experimental_handlers.update_capabilities(capabilities)
    return capabilities

experimental property

experimental: ExperimentalHandlers[LifespanResultT]

Experimental APIs for tasks and other features.

WARNING: These APIs are experimental and may change without notice.

session_manager property

session_manager: StreamableHTTPSessionManager

Get the StreamableHTTP session manager.

Raises:

Type	Description
RuntimeError	If called before streamable_http_app() has been called.

streamable_http_app

streamable_http_app(
    *,
    streamable_http_path: str = "/mcp",
    json_response: bool = False,
    stateless_http: bool = False,
    event_store: EventStore | None = None,
    retry_interval: int | None = None,
    transport_security: (
        TransportSecuritySettings | None
    ) = None,
    host: str = "127.0.0.1",
    auth: AuthSettings | None = None,
    token_verifier: TokenVerifier | None = None,
    auth_server_provider: (
        OAuthAuthorizationServerProvider[Any, Any, Any]
        | None
    ) = None,
    custom_starlette_routes: list[Route] | None = None,
    debug: bool = False
) -> Starlette

Return an instance of the StreamableHTTP server app.

Source code in src/mcp/server/lowlevel/server.py

def streamable_http_app(
    self,
    *,
    streamable_http_path: str = "/mcp",
    json_response: bool = False,
    stateless_http: bool = False,
    event_store: EventStore | None = None,
    retry_interval: int | None = None,
    transport_security: TransportSecuritySettings | None = None,
    host: str = "127.0.0.1",
    auth: AuthSettings | None = None,
    token_verifier: TokenVerifier | None = None,
    auth_server_provider: OAuthAuthorizationServerProvider[Any, Any, Any] | None = None,
    custom_starlette_routes: list[Route] | None = None,
    debug: bool = False,
) -> Starlette:
    """Return an instance of the StreamableHTTP server app."""
    # Auto-enable DNS rebinding protection for localhost (IPv4 and IPv6)
    if transport_security is None and host in ("127.0.0.1", "localhost", "::1"):
        transport_security = TransportSecuritySettings(
            enable_dns_rebinding_protection=True,
            allowed_hosts=["127.0.0.1:*", "localhost:*", "[::1]:*"],
            allowed_origins=["http://127.0.0.1:*", "http://localhost:*", "http://[::1]:*"],
        )

    session_manager = StreamableHTTPSessionManager(
        app=self,
        event_store=event_store,
        retry_interval=retry_interval,
        json_response=json_response,
        stateless=stateless_http,
        security_settings=transport_security,
    )
    self._session_manager = session_manager

    # Create the ASGI handler
    streamable_http_app = StreamableHTTPASGIApp(session_manager)

    # Create routes
    routes: list[Route | Mount] = []
    middleware: list[Middleware] = []
    required_scopes: list[str] = []

    # Set up auth if configured
    if auth:  # pragma: no cover
        required_scopes = auth.required_scopes or []

        # Add auth middleware if token verifier is available
        if token_verifier:
            middleware = [
                Middleware(
                    AuthenticationMiddleware,
                    backend=BearerAuthBackend(token_verifier),
                ),
                Middleware(AuthContextMiddleware),
            ]

        # Add auth endpoints if auth server provider is configured
        if auth_server_provider:
            routes.extend(
                create_auth_routes(
                    provider=auth_server_provider,
                    issuer_url=auth.issuer_url,
                    service_documentation_url=auth.service_documentation_url,
                    client_registration_options=auth.client_registration_options,
                    revocation_options=auth.revocation_options,
                )
            )

    # Set up routes with or without auth
    if token_verifier:  # pragma: no cover
        # Determine resource metadata URL
        resource_metadata_url = None
        if auth and auth.resource_server_url:
            # Build compliant metadata URL for WWW-Authenticate header
            resource_metadata_url = build_resource_metadata_url(auth.resource_server_url)

        routes.append(
            Route(
                streamable_http_path,
                endpoint=RequireAuthMiddleware(streamable_http_app, required_scopes, resource_metadata_url),
            )
        )
    else:
        # Auth is disabled, no wrapper needed
        routes.append(
            Route(
                streamable_http_path,
                endpoint=streamable_http_app,
            )
        )

    # Add protected resource metadata endpoint if configured as RS
    if auth and auth.resource_server_url:  # pragma: no cover
        routes.extend(
            create_protected_resource_routes(
                resource_url=auth.resource_server_url,
                authorization_servers=[auth.issuer_url],
                scopes_supported=auth.required_scopes,
            )
        )

    if custom_starlette_routes:  # pragma: no cover
        routes.extend(custom_starlette_routes)

    return Starlette(
        debug=debug,
        routes=routes,
        middleware=middleware,
        lifespan=lambda app: session_manager.run(),
    )

MCPServer

Bases: Generic[LifespanResultT]

Source code in src/mcp/server/mcpserver/server.py

class MCPServer(Generic[LifespanResultT]):
    def __init__(
        self,
        name: str | None = None,
        title: str | None = None,
        description: str | None = None,
        instructions: str | None = None,
        website_url: str | None = None,
        icons: list[Icon] | None = None,
        version: str | None = None,
        auth_server_provider: OAuthAuthorizationServerProvider[Any, Any, Any] | None = None,
        token_verifier: TokenVerifier | None = None,
        *,
        tools: list[Tool] | None = None,
        resources: list[Resource] | None = None,
        debug: bool = False,
        log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO",
        warn_on_duplicate_resources: bool = True,
        warn_on_duplicate_tools: bool = True,
        warn_on_duplicate_prompts: bool = True,
        dependencies: list[str] | None = None,
        lifespan: Callable[[MCPServer[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]] | None = None,
        auth: AuthSettings | None = None,
    ):
        self.settings = Settings(
            debug=debug,
            log_level=log_level,
            warn_on_duplicate_resources=warn_on_duplicate_resources,
            warn_on_duplicate_tools=warn_on_duplicate_tools,
            warn_on_duplicate_prompts=warn_on_duplicate_prompts,
            dependencies=dependencies or [],
            lifespan=lifespan,
            auth=auth,
        )
        self.dependencies = self.settings.dependencies

        self._tool_manager = ToolManager(tools=tools, warn_on_duplicate_tools=self.settings.warn_on_duplicate_tools)
        self._resource_manager = ResourceManager(
            resources=resources, warn_on_duplicate_resources=self.settings.warn_on_duplicate_resources
        )
        self._prompt_manager = PromptManager(warn_on_duplicate_prompts=self.settings.warn_on_duplicate_prompts)
        self._lowlevel_server = Server(
            name=name or "mcp-server",
            title=title,
            description=description,
            instructions=instructions,
            website_url=website_url,
            icons=icons,
            version=version,
            on_list_tools=self._handle_list_tools,
            on_call_tool=self._handle_call_tool,
            on_list_resources=self._handle_list_resources,
            on_read_resource=self._handle_read_resource,
            on_list_resource_templates=self._handle_list_resource_templates,
            on_list_prompts=self._handle_list_prompts,
            on_get_prompt=self._handle_get_prompt,
            # TODO(Marcelo): It seems there's a type mismatch between the lifespan type from an MCPServer and Server.
            # We need to create a Lifespan type that is a generic on the server type, like Starlette does.
            lifespan=(lifespan_wrapper(self, self.settings.lifespan) if self.settings.lifespan else default_lifespan),  # type: ignore
        )
        # Validate auth configuration
        if self.settings.auth is not None:
            if auth_server_provider and token_verifier:  # pragma: no cover
                raise ValueError("Cannot specify both auth_server_provider and token_verifier")
            if not auth_server_provider and not token_verifier:  # pragma: no cover
                raise ValueError("Must specify either auth_server_provider or token_verifier when auth is enabled")
        elif auth_server_provider or token_verifier:  # pragma: no cover
            raise ValueError("Cannot specify auth_server_provider or token_verifier without auth settings")

        self._auth_server_provider = auth_server_provider
        self._token_verifier = token_verifier

        # Create token verifier from provider if needed (backwards compatibility)
        if auth_server_provider and not token_verifier:  # pragma: no cover
            self._token_verifier = ProviderTokenVerifier(auth_server_provider)
        self._custom_starlette_routes: list[Route] = []

        # Configure logging
        configure_logging(self.settings.log_level)

    @property
    def name(self) -> str:
        return self._lowlevel_server.name

    @property
    def title(self) -> str | None:
        return self._lowlevel_server.title

    @property
    def description(self) -> str | None:
        return self._lowlevel_server.description

    @property
    def instructions(self) -> str | None:
        return self._lowlevel_server.instructions

    @property
    def website_url(self) -> str | None:
        return self._lowlevel_server.website_url

    @property
    def icons(self) -> list[Icon] | None:
        return self._lowlevel_server.icons

    @property
    def version(self) -> str | None:
        return self._lowlevel_server.version

    @property
    def session_manager(self) -> StreamableHTTPSessionManager:
        """Get the StreamableHTTP session manager.

        This is exposed to enable advanced use cases like mounting multiple
        MCPServer instances in a single FastAPI application.

        Raises:
            RuntimeError: If called before streamable_http_app() has been called.
        """
        return self._lowlevel_server.session_manager  # pragma: no cover

    @overload
    def run(self, transport: Literal["stdio"] = ...) -> None: ...

    @overload
    def run(
        self,
        transport: Literal["sse"],
        *,
        host: str = ...,
        port: int = ...,
        sse_path: str = ...,
        message_path: str = ...,
        transport_security: TransportSecuritySettings | None = ...,
    ) -> None: ...

    @overload
    def run(
        self,
        transport: Literal["streamable-http"],
        *,
        host: str = ...,
        port: int = ...,
        streamable_http_path: str = ...,
        json_response: bool = ...,
        stateless_http: bool = ...,
        event_store: EventStore | None = ...,
        retry_interval: int | None = ...,
        transport_security: TransportSecuritySettings | None = ...,
    ) -> None: ...

    def run(
        self,
        transport: Literal["stdio", "sse", "streamable-http"] = "stdio",
        **kwargs: Any,
    ) -> None:
        """Run the MCP server. Note this is a synchronous function.

        Args:
            transport: Transport protocol to use ("stdio", "sse", or "streamable-http")
            **kwargs: Transport-specific options (see overloads for details)
        """
        TRANSPORTS = Literal["stdio", "sse", "streamable-http"]
        if transport not in TRANSPORTS.__args__:  # type: ignore  # pragma: no cover
            raise ValueError(f"Unknown transport: {transport}")

        match transport:
            case "stdio":
                anyio.run(self.run_stdio_async)
            case "sse":  # pragma: no cover
                anyio.run(lambda: self.run_sse_async(**kwargs))
            case "streamable-http":  # pragma: no cover
                anyio.run(lambda: self.run_streamable_http_async(**kwargs))

    async def _handle_list_tools(
        self, ctx: ServerRequestContext[LifespanResultT], params: PaginatedRequestParams | None
    ) -> ListToolsResult:
        return ListToolsResult(tools=await self.list_tools())

    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 {}, context)
        except MCPError:
            raise
        except Exception as e:
            return CallToolResult(content=[TextContent(type="text", text=str(e))], is_error=True)
        if isinstance(result, CallToolResult):
            return result
        if isinstance(result, tuple) and len(result) == 2:
            unstructured_content, structured_content = result
            return CallToolResult(
                content=list(unstructured_content),  # type: ignore[arg-type]
                structured_content=structured_content,  # type: ignore[arg-type]
            )
        if isinstance(result, dict):  # pragma: no cover
            # TODO: this code path is unreachable — convert_result never returns a raw dict.
            # The call_tool return type (Sequence[ContentBlock] | dict[str, Any]) is wrong
            # and needs to be cleaned up.
            return CallToolResult(
                content=[TextContent(type="text", text=json.dumps(result, indent=2))],
                structured_content=result,
            )
        return CallToolResult(content=list(result))

    async def _handle_list_resources(
        self, ctx: ServerRequestContext[LifespanResultT], params: PaginatedRequestParams | None
    ) -> ListResourcesResult:
        return ListResourcesResult(resources=await self.list_resources())

    async def _handle_read_resource(
        self, ctx: ServerRequestContext[LifespanResultT], params: ReadResourceRequestParams
    ) -> ReadResourceResult:
        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):
                contents.append(
                    BlobResourceContents(
                        uri=params.uri,
                        blob=base64.b64encode(item.content).decode(),
                        mime_type=item.mime_type or "application/octet-stream",
                        _meta=item.meta,
                    )
                )
            else:
                contents.append(
                    TextResourceContents(
                        uri=params.uri,
                        text=item.content,
                        mime_type=item.mime_type or "text/plain",
                        _meta=item.meta,
                    )
                )
        return ReadResourceResult(contents=contents)

    async def _handle_list_resource_templates(
        self, ctx: ServerRequestContext[LifespanResultT], params: PaginatedRequestParams | None
    ) -> ListResourceTemplatesResult:
        return ListResourceTemplatesResult(resource_templates=await self.list_resource_templates())

    async def _handle_list_prompts(
        self, ctx: ServerRequestContext[LifespanResultT], params: PaginatedRequestParams | None
    ) -> ListPromptsResult:
        return ListPromptsResult(prompts=await self.list_prompts())

    async def _handle_get_prompt(
        self, ctx: ServerRequestContext[LifespanResultT], params: GetPromptRequestParams
    ) -> GetPromptResult:
        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."""
        tools = self._tool_manager.list_tools()
        return [
            MCPTool(
                name=info.name,
                title=info.title,
                description=info.description,
                input_schema=info.parameters,
                output_schema=info.output_schema,
                annotations=info.annotations,
                icons=info.icons,
                _meta=info.meta,
            )
            for info in tools
        ]

    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."""
        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."""

        resources = self._resource_manager.list_resources()
        return [
            MCPResource(
                uri=resource.uri,
                name=resource.name or "",
                title=resource.title,
                description=resource.description,
                mime_type=resource.mime_type,
                icons=resource.icons,
                annotations=resource.annotations,
                _meta=resource.meta,
            )
            for resource in resources
        ]

    async def list_resource_templates(self) -> list[MCPResourceTemplate]:
        templates = self._resource_manager.list_templates()
        return [
            MCPResourceTemplate(
                uri_template=template.uri_template,
                name=template.name,
                title=template.title,
                description=template.description,
                mime_type=template.mime_type,
                icons=template.icons,
                annotations=template.annotations,
                _meta=template.meta,
            )
            for template in templates
        ]

    async def read_resource(
        self, uri: AnyUrl | str, context: Context[LifespanResultT, Any] | None = None
    ) -> Iterable[ReadResourceContents]:
        """Read a resource by URI."""
        if context is None:
            context = Context(mcp_server=self)
        try:
            resource = await self._resource_manager.get_resource(uri, context)
        except ValueError as exc:
            raise ResourceError(f"Unknown resource: {uri}") from exc

        try:
            content = await resource.read()
            return [ReadResourceContents(content=content, mime_type=resource.mime_type, meta=resource.meta)]
        except Exception as exc:
            logger.exception(f"Error getting resource {uri}")
            # If an exception happens when reading the resource, we should not leak the exception to the client.
            raise ResourceError(f"Error reading resource {uri}") from exc

    def add_tool(
        self,
        fn: Callable[..., Any],
        name: str | None = None,
        title: str | None = None,
        description: str | None = None,
        annotations: ToolAnnotations | None = None,
        icons: list[Icon] | None = None,
        meta: dict[str, Any] | None = None,
        structured_output: bool | None = None,
    ) -> None:
        """Add a tool to the server.

        The tool function can optionally request a Context object by adding a parameter
        with the Context type annotation. See the @tool decorator for examples.

        Args:
            fn: The function to register as a tool
            name: Optional name for the tool (defaults to function name)
            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)
                - If False, unconditionally creates an unstructured tool
        """
        self._tool_manager.add_tool(
            fn,
            name=name,
            title=title,
            description=description,
            annotations=annotations,
            icons=icons,
            meta=meta,
            structured_output=structured_output,
        )

    def remove_tool(self, name: str) -> None:
        """Remove a tool from the server by name.

        Args:
            name: The name of the tool to remove

        Raises:
            ToolError: If the tool does not exist
        """
        self._tool_manager.remove_tool(name)

    def tool(
        self,
        name: str | None = None,
        title: str | None = None,
        description: str | None = None,
        annotations: ToolAnnotations | None = None,
        icons: list[Icon] | None = None,
        meta: dict[str, Any] | None = None,
        structured_output: bool | None = None,
    ) -> Callable[[_CallableT], _CallableT]:
        """Decorator to register a tool.

        Tools can optionally request a Context object by adding a parameter with the
        Context type annotation. The context provides access to MCP capabilities like
        logging, progress reporting, and resource access.

        Args:
            name: Optional name for the tool (defaults to function name)
            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)
                - If False, unconditionally creates an unstructured tool

        Example:
            ```python
            @server.tool()
            def my_tool(x: int) -> str:
                return str(x)
            ```

            ```python
            @server.tool()
            async def tool_with_context(x: int, ctx: Context) -> str:
                await ctx.info(f"Processing {x}")
                return str(x)
            ```

            ```python
            @server.tool()
            async def async_tool(x: int, context: Context) -> str:
                await context.report_progress(50, 100)
                return str(x)
            ```
        """
        # Check if user passed function directly instead of calling decorator
        if callable(name):
            raise TypeError(
                "The @tool decorator was used incorrectly. Did you forget to call it? Use @tool() instead of @tool"
            )

        def decorator(fn: _CallableT) -> _CallableT:
            self.add_tool(
                fn,
                name=name,
                title=title,
                description=description,
                annotations=annotations,
                icons=icons,
                meta=meta,
                structured_output=structured_output,
            )
            return fn

        return decorator

    def completion(self):
        """Decorator to register a completion handler.

        The completion handler receives:
        - ref: PromptReference or ResourceTemplateReference
        - argument: CompletionArgument with name and partial value
        - context: Optional CompletionContext with previously resolved arguments

        Example:
            ```python
            @mcp.completion()
            async def handle_completion(ref, argument, context):
                if isinstance(ref, ResourceTemplateReference):
                    # Return completions based on ref, argument, and context
                    return Completion(values=["option1", "option2"])
                return None
            ```
        """

        def decorator(func: _CallableT) -> _CallableT:
            async def handler(
                ctx: ServerRequestContext[LifespanResultT], params: CompleteRequestParams
            ) -> CompleteResult:
                result = await func(params.ref, params.argument, params.context)
                return CompleteResult(
                    completion=result if result is not None else Completion(values=[], total=None, has_more=None),
                )

            # TODO(maxisbey): remove private access — completion needs post-construction
            #   handler registration, find a better pattern for this
            self._lowlevel_server._add_request_handler(  # pyright: ignore[reportPrivateUsage]
                "completion/complete", handler
            )
            return func

        return decorator

    def add_resource(self, resource: Resource) -> None:
        """Add a resource to the server.

        Args:
            resource: A Resource instance to add
        """
        self._resource_manager.add_resource(resource)

    def resource(
        self,
        uri: str,
        *,
        name: str | None = None,
        title: str | None = None,
        description: str | None = None,
        mime_type: str | None = None,
        icons: list[Icon] | None = None,
        annotations: Annotations | None = None,
        meta: dict[str, Any] | None = None,
    ) -> Callable[[_CallableT], _CallableT]:
        """Decorator to register a function as a resource.

        The function will be called when the resource is read to generate its content.
        The function can return:
        - str for text content
        - bytes for binary content
        - other types will be converted to JSON

        If the URI contains parameters (e.g. "resource://{param}") or the function
        has parameters, it will be registered as a template resource.

        Args:
            uri: URI for the resource (e.g. "resource://my-resource" or "resource://{param}")
            name: Optional name for the resource
            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:
            ```python
            @server.resource("resource://my-resource")
            def get_data() -> str:
                return "Hello, world!"

            @server.resource("resource://my-resource")
            async def get_data() -> str:
                data = await fetch_data()
                return f"Hello, world! {data}"

            @server.resource("resource://{city}/weather")
            def get_weather(city: str) -> str:
                return f"Weather for {city}"

            @server.resource("resource://{city}/weather")
            async def get_weather(city: str) -> str:
                data = await fetch_weather(city)
                return f"Weather for {city}: {data}"
            ```
        """
        # Check if user passed function directly instead of calling decorator
        if callable(uri):
            raise TypeError(
                "The @resource decorator was used incorrectly. "
                "Did you forget to call it? Use @resource('uri') instead of @resource"
            )

        def decorator(fn: _CallableT) -> _CallableT:
            # Check if this should be a template
            sig = inspect.signature(fn)
            has_uri_params = "{" in uri and "}" in uri
            has_func_params = bool(sig.parameters)

            if has_uri_params or has_func_params:
                # Check for Context parameter to exclude from validation
                context_param = find_context_parameter(fn)

                # Validate that URI params match function params (excluding context)
                uri_params = set(re.findall(r"{(\w+)}", uri))
                # We need to remove the context_param from the resource function if
                # there is any.
                func_params = {p for p in sig.parameters.keys() if p != context_param}

                if uri_params != func_params:
                    raise ValueError(
                        f"Mismatch between URI parameters {uri_params} and function parameters {func_params}"
                    )

                # Register as template
                self._resource_manager.add_template(
                    fn=fn,
                    uri_template=uri,
                    name=name,
                    title=title,
                    description=description,
                    mime_type=mime_type,
                    icons=icons,
                    annotations=annotations,
                    meta=meta,
                )
            else:
                # Register as regular resource
                resource = FunctionResource.from_function(
                    fn=fn,
                    uri=uri,
                    name=name,
                    title=title,
                    description=description,
                    mime_type=mime_type,
                    icons=icons,
                    annotations=annotations,
                    meta=meta,
                )
                self.add_resource(resource)
            return fn

        return decorator

    def add_prompt(self, prompt: Prompt) -> None:
        """Add a prompt to the server.

        Args:
            prompt: A Prompt instance to add
        """
        self._prompt_manager.add_prompt(prompt)

    def prompt(
        self,
        name: str | None = None,
        title: str | None = None,
        description: str | None = None,
        icons: list[Icon] | None = None,
    ) -> Callable[[_CallableT], _CallableT]:
        """Decorator to register a prompt.

        Args:
            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:
            ```python
            @server.prompt()
            def analyze_table(table_name: str) -> list[Message]:
                schema = read_table_schema(table_name)
                return [
                    {
                        "role": "user",
                        "content": f"Analyze this schema:\n{schema}"
                    }
                ]

            @server.prompt()
            async def analyze_file(path: str) -> list[Message]:
                content = await read_file(path)
                return [
                    {
                        "role": "user",
                        "content": {
                            "type": "resource",
                            "resource": {
                                "uri": f"file://{path}",
                                "text": content
                            }
                        }
                    }
                ]
            ```
        """
        # Check if user passed function directly instead of calling decorator
        if callable(name):
            raise TypeError(
                "The @prompt decorator was used incorrectly. "
                "Did you forget to call it? Use @prompt() instead of @prompt"
            )

        def decorator(func: _CallableT) -> _CallableT:
            prompt = Prompt.from_function(func, name=name, title=title, description=description, icons=icons)
            self.add_prompt(prompt)
            return func

        return decorator

    def custom_route(
        self,
        path: str,
        methods: list[str],
        name: str | None = None,
        include_in_schema: bool = True,
    ):
        """Decorator to register a custom HTTP route on the MCP server.

        Allows adding arbitrary HTTP endpoints outside the standard MCP protocol,
        which can be useful for OAuth callbacks, health checks, or admin APIs.
        The handler function must be an async function that accepts a Starlette
        Request and returns a Response.

        Routes using this decorator will not require authorization. It is intended
        for uses that are either a part of authorization flows or intended to be
        public such as health check endpoints.

        Args:
            path: URL path for the route (e.g., "/oauth/callback")
            methods: List of HTTP methods to support (e.g., ["GET", "POST"])
            name: Optional name for the route (to reference this route with
                  Starlette's reverse URL lookup feature)
            include_in_schema: Whether to include in OpenAPI schema, defaults to True

        Example:
            ```python
            @server.custom_route("/health", methods=["GET"])
            async def health_check(request: Request) -> Response:
                return JSONResponse({"status": "ok"})
            ```
        """

        def decorator(  # pragma: no cover
            func: Callable[[Request], Awaitable[Response]],
        ) -> Callable[[Request], Awaitable[Response]]:
            self._custom_starlette_routes.append(
                Route(path, endpoint=func, methods=methods, name=name, include_in_schema=include_in_schema)
            )
            return func

        return decorator  # pragma: no cover

    async def run_stdio_async(self) -> None:
        """Run the server using stdio transport."""
        async with stdio_server() as (read_stream, write_stream):
            await self._lowlevel_server.run(
                read_stream,
                write_stream,
                self._lowlevel_server.create_initialization_options(),
            )

    async def run_sse_async(  # pragma: no cover
        self,
        *,
        host: str = "127.0.0.1",
        port: int = 8000,
        sse_path: str = "/sse",
        message_path: str = "/messages/",
        transport_security: TransportSecuritySettings | None = None,
    ) -> None:
        """Run the server using SSE transport."""
        import uvicorn

        starlette_app = self.sse_app(
            sse_path=sse_path,
            message_path=message_path,
            transport_security=transport_security,
            host=host,
        )

        config = uvicorn.Config(
            starlette_app,
            host=host,
            port=port,
            log_level=self.settings.log_level.lower(),
        )
        server = uvicorn.Server(config)
        await server.serve()

    async def run_streamable_http_async(  # pragma: no cover
        self,
        *,
        host: str = "127.0.0.1",
        port: int = 8000,
        streamable_http_path: str = "/mcp",
        json_response: bool = False,
        stateless_http: bool = False,
        event_store: EventStore | None = None,
        retry_interval: int | None = None,
        transport_security: TransportSecuritySettings | None = None,
    ) -> None:
        """Run the server using StreamableHTTP transport."""
        import uvicorn

        starlette_app = self.streamable_http_app(
            streamable_http_path=streamable_http_path,
            json_response=json_response,
            stateless_http=stateless_http,
            event_store=event_store,
            retry_interval=retry_interval,
            transport_security=transport_security,
            host=host,
        )

        config = uvicorn.Config(
            starlette_app,
            host=host,
            port=port,
            log_level=self.settings.log_level.lower(),
        )
        server = uvicorn.Server(config)
        await server.serve()

    def sse_app(
        self,
        *,
        sse_path: str = "/sse",
        message_path: str = "/messages/",
        transport_security: TransportSecuritySettings | None = None,
        host: str = "127.0.0.1",
    ) -> Starlette:
        """Return an instance of the SSE server app."""
        # Auto-enable DNS rebinding protection for localhost (IPv4 and IPv6)
        if transport_security is None and host in ("127.0.0.1", "localhost", "::1"):
            transport_security = TransportSecuritySettings(
                enable_dns_rebinding_protection=True,
                allowed_hosts=["127.0.0.1:*", "localhost:*", "[::1]:*"],
                allowed_origins=["http://127.0.0.1:*", "http://localhost:*", "http://[::1]:*"],
            )

        sse = SseServerTransport(message_path, security_settings=transport_security)

        async def handle_sse(scope: Scope, receive: Receive, send: Send):  # pragma: no cover
            # Add client ID from auth context into request context if available

            async with sse.connect_sse(scope, receive, send) as streams:
                await self._lowlevel_server.run(
                    streams[0], streams[1], self._lowlevel_server.create_initialization_options()
                )
            return Response()

        # Create routes
        routes: list[Route | Mount] = []
        middleware: list[Middleware] = []
        required_scopes: list[str] = []

        # Set up auth if configured
        if self.settings.auth:  # pragma: no cover
            required_scopes = self.settings.auth.required_scopes or []

            # Add auth middleware if token verifier is available
            if self._token_verifier:
                middleware = [
                    # extract auth info from request (but do not require it)
                    Middleware(
                        AuthenticationMiddleware,
                        backend=BearerAuthBackend(self._token_verifier),
                    ),
                    # Add the auth context middleware to store
                    # authenticated user in a contextvar
                    Middleware(AuthContextMiddleware),
                ]

            # Add auth endpoints if auth server provider is configured
            if self._auth_server_provider:
                from mcp.server.auth.routes import create_auth_routes

                routes.extend(
                    create_auth_routes(
                        provider=self._auth_server_provider,
                        issuer_url=self.settings.auth.issuer_url,
                        service_documentation_url=self.settings.auth.service_documentation_url,
                        client_registration_options=self.settings.auth.client_registration_options,
                        revocation_options=self.settings.auth.revocation_options,
                    )
                )

        # When auth is configured, require authentication
        if self._token_verifier:  # pragma: no cover
            # Determine resource metadata URL
            resource_metadata_url = None
            if self.settings.auth and self.settings.auth.resource_server_url:
                from mcp.server.auth.routes import build_resource_metadata_url

                # Build compliant metadata URL for WWW-Authenticate header
                resource_metadata_url = build_resource_metadata_url(self.settings.auth.resource_server_url)

            # Auth is enabled, wrap the endpoints with RequireAuthMiddleware
            routes.append(
                Route(
                    sse_path,
                    endpoint=RequireAuthMiddleware(handle_sse, required_scopes, resource_metadata_url),
                    methods=["GET"],
                )
            )
            routes.append(
                Mount(
                    message_path,
                    app=RequireAuthMiddleware(sse.handle_post_message, required_scopes, resource_metadata_url),
                )
            )
        else:
            # Auth is disabled, no need for RequireAuthMiddleware
            # Since handle_sse is an ASGI app, we need to create a compatible endpoint
            async def sse_endpoint(request: Request) -> Response:  # pragma: no cover
                # Convert the Starlette request to ASGI parameters
                return await handle_sse(request.scope, request.receive, request._send)  # type: ignore[reportPrivateUsage]

            routes.append(
                Route(
                    sse_path,
                    endpoint=sse_endpoint,
                    methods=["GET"],
                )
            )
            routes.append(
                Mount(
                    message_path,
                    app=sse.handle_post_message,
                )
            )
        # Add protected resource metadata endpoint if configured as RS
        if self.settings.auth and self.settings.auth.resource_server_url:  # pragma: no cover
            from mcp.server.auth.routes import create_protected_resource_routes

            routes.extend(
                create_protected_resource_routes(
                    resource_url=self.settings.auth.resource_server_url,
                    authorization_servers=[self.settings.auth.issuer_url],
                    scopes_supported=self.settings.auth.required_scopes,
                )
            )

        # mount these routes last, so they have the lowest route matching precedence
        routes.extend(self._custom_starlette_routes)

        # Create Starlette app with routes and middleware
        return Starlette(debug=self.settings.debug, routes=routes, middleware=middleware)

    def streamable_http_app(
        self,
        *,
        streamable_http_path: str = "/mcp",
        json_response: bool = False,
        stateless_http: bool = False,
        event_store: EventStore | None = None,
        retry_interval: int | None = None,
        transport_security: TransportSecuritySettings | None = None,
        host: str = "127.0.0.1",
    ) -> Starlette:
        """Return an instance of the StreamableHTTP server app."""
        return self._lowlevel_server.streamable_http_app(
            streamable_http_path=streamable_http_path,
            json_response=json_response,
            stateless_http=stateless_http,
            event_store=event_store,
            retry_interval=retry_interval,
            transport_security=transport_security,
            host=host,
            auth=self.settings.auth,
            token_verifier=self._token_verifier,
            auth_server_provider=self._auth_server_provider,
            custom_starlette_routes=self._custom_starlette_routes,
            debug=self.settings.debug,
        )

    async def list_prompts(self) -> list[MCPPrompt]:
        """List all available prompts."""
        prompts = self._prompt_manager.list_prompts()
        return [
            MCPPrompt(
                name=prompt.name,
                title=prompt.title,
                description=prompt.description,
                arguments=[
                    MCPPromptArgument(
                        name=arg.name,
                        description=arg.description,
                        required=arg.required,
                    )
                    for arg in (prompt.arguments or [])
                ],
                icons=prompt.icons,
            )
            for prompt in prompts
        ]

    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)

            return GetPromptResult(
                description=prompt.description,
                messages=pydantic_core.to_jsonable_python(messages),
            )
        except Exception as e:
            logger.exception(f"Error getting prompt {name}")
            raise ValueError(str(e)) from e

session_manager property

session_manager: StreamableHTTPSessionManager

Get the StreamableHTTP session manager.

This is exposed to enable advanced use cases like mounting multiple MCPServer instances in a single FastAPI application.

Raises:

Type	Description
RuntimeError	If called before streamable_http_app() has been called.

run

run(transport: Literal['stdio'] = ...) -> None

run(
    transport: Literal["sse"],
    *,
    host: str = ...,
    port: int = ...,
    sse_path: str = ...,
    message_path: str = ...,
    transport_security: (
        TransportSecuritySettings | None
    ) = ...
) -> None

run(
    transport: Literal["streamable-http"],
    *,
    host: str = ...,
    port: int = ...,
    streamable_http_path: str = ...,
    json_response: bool = ...,
    stateless_http: bool = ...,
    event_store: EventStore | None = ...,
    retry_interval: int | None = ...,
    transport_security: (
        TransportSecuritySettings | None
    ) = ...
) -> None

run(
    transport: Literal[
        "stdio", "sse", "streamable-http"
    ] = "stdio",
    **kwargs: Any
) -> None

Run the MCP server. Note this is a synchronous function.

Parameters:

Name	Type	Description	Default
transport	Literal['stdio', 'sse', 'streamable-http']	Transport protocol to use ("stdio", "sse", or "streamable-http")	'stdio'
**kwargs	Any	Transport-specific options (see overloads for details)	{}

Source code in src/mcp/server/mcpserver/server.py

def run(
    self,
    transport: Literal["stdio", "sse", "streamable-http"] = "stdio",
    **kwargs: Any,
) -> None:
    """Run the MCP server. Note this is a synchronous function.

    Args:
        transport: Transport protocol to use ("stdio", "sse", or "streamable-http")
        **kwargs: Transport-specific options (see overloads for details)
    """
    TRANSPORTS = Literal["stdio", "sse", "streamable-http"]
    if transport not in TRANSPORTS.__args__:  # type: ignore  # pragma: no cover
        raise ValueError(f"Unknown transport: {transport}")

    match transport:
        case "stdio":
            anyio.run(self.run_stdio_async)
        case "sse":  # pragma: no cover
            anyio.run(lambda: self.run_sse_async(**kwargs))
        case "streamable-http":  # pragma: no cover
            anyio.run(lambda: self.run_streamable_http_async(**kwargs))

list_tools async

list_tools() -> list[Tool]

List all available tools.

Source code in src/mcp/server/mcpserver/server.py

async def list_tools(self) -> list[MCPTool]:
    """List all available tools."""
    tools = self._tool_manager.list_tools()
    return [
        MCPTool(
            name=info.name,
            title=info.title,
            description=info.description,
            input_schema=info.parameters,
            output_schema=info.output_schema,
            annotations=info.annotations,
            icons=info.icons,
            _meta=info.meta,
        )
        for info in tools
    ]

call_tool async

call_tool(
    name: str,
    arguments: dict[str, Any],
    context: Context[LifespanResultT, Any] | None = None,
) -> Sequence[ContentBlock] | dict[str, Any]

Call a tool by name with arguments.

Source code in src/mcp/server/mcpserver/server.py

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."""
    if context is None:
        context = Context(mcp_server=self)
    return await self._tool_manager.call_tool(name, arguments, context, convert_result=True)

list_resources async

list_resources() -> list[Resource]

List all available resources.

Source code in src/mcp/server/mcpserver/server.py

async def list_resources(self) -> list[MCPResource]:
    """List all available resources."""

    resources = self._resource_manager.list_resources()
    return [
        MCPResource(
            uri=resource.uri,
            name=resource.name or "",
            title=resource.title,
            description=resource.description,
            mime_type=resource.mime_type,
            icons=resource.icons,
            annotations=resource.annotations,
            _meta=resource.meta,
        )
        for resource in resources
    ]

read_resource async

read_resource(
    uri: AnyUrl | str,
    context: Context[LifespanResultT, Any] | None = None,
) -> Iterable[ReadResourceContents]

Read a resource by URI.

Source code in src/mcp/server/mcpserver/server.py

async def read_resource(
    self, uri: AnyUrl | str, context: Context[LifespanResultT, Any] | None = None
) -> Iterable[ReadResourceContents]:
    """Read a resource by URI."""
    if context is None:
        context = Context(mcp_server=self)
    try:
        resource = await self._resource_manager.get_resource(uri, context)
    except ValueError as exc:
        raise ResourceError(f"Unknown resource: {uri}") from exc

    try:
        content = await resource.read()
        return [ReadResourceContents(content=content, mime_type=resource.mime_type, meta=resource.meta)]
    except Exception as exc:
        logger.exception(f"Error getting resource {uri}")
        # If an exception happens when reading the resource, we should not leak the exception to the client.
        raise ResourceError(f"Error reading resource {uri}") from exc

add_tool

add_tool(
    fn: Callable[..., Any],
    name: str | None = None,
    title: str | None = None,
    description: str | None = None,
    annotations: ToolAnnotations | None = None,
    icons: list[Icon] | None = None,
    meta: dict[str, Any] | None = None,
    structured_output: bool | None = None,
) -> None

Add a tool to the server.

The tool function can optionally request a Context object by adding a parameter with the Context type annotation. See the @tool decorator for examples.

Parameters:

Name	Type	Description	Default
fn	Callable[..., Any]	The function to register as a tool	required
name	str | None	Optional name for the tool (defaults to function name)	None
title	str | None	Optional human-readable title for the tool	None
description	str | None	Optional description of what the tool does	None
annotations	ToolAnnotations | None	Optional ToolAnnotations providing additional tool information	None
icons	list[Icon] | None	Optional list of icons for the tool	None
meta	dict[str, Any] | None	Optional metadata dictionary for the tool	None
structured_output	bool | None	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) - If False, unconditionally creates an unstructured tool	None

Source code in src/mcp/server/mcpserver/server.py

def add_tool(
    self,
    fn: Callable[..., Any],
    name: str | None = None,
    title: str | None = None,
    description: str | None = None,
    annotations: ToolAnnotations | None = None,
    icons: list[Icon] | None = None,
    meta: dict[str, Any] | None = None,
    structured_output: bool | None = None,
) -> None:
    """Add a tool to the server.

    The tool function can optionally request a Context object by adding a parameter
    with the Context type annotation. See the @tool decorator for examples.

    Args:
        fn: The function to register as a tool
        name: Optional name for the tool (defaults to function name)
        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)
            - If False, unconditionally creates an unstructured tool
    """
    self._tool_manager.add_tool(
        fn,
        name=name,
        title=title,
        description=description,
        annotations=annotations,
        icons=icons,
        meta=meta,
        structured_output=structured_output,
    )

remove_tool

remove_tool(name: str) -> None

Remove a tool from the server by name.

Parameters:

Name	Type	Description	Default
name	str	The name of the tool to remove	required

Raises:

Type	Description
ToolError	If the tool does not exist

Source code in src/mcp/server/mcpserver/server.py

def remove_tool(self, name: str) -> None:
    """Remove a tool from the server by name.

    Args:
        name: The name of the tool to remove

    Raises:
        ToolError: If the tool does not exist
    """
    self._tool_manager.remove_tool(name)

tool

tool(
    name: str | None = None,
    title: str | None = None,
    description: str | None = None,
    annotations: ToolAnnotations | None = None,
    icons: list[Icon] | None = None,
    meta: dict[str, Any] | None = None,
    structured_output: bool | None = None,
) -> Callable[[_CallableT], _CallableT]

Decorator to register a tool.

Tools can optionally request a Context object by adding a parameter with the Context type annotation. The context provides access to MCP capabilities like logging, progress reporting, and resource access.

Parameters:

Name	Type	Description	Default
name	str | None	Optional name for the tool (defaults to function name)	None
title	str | None	Optional human-readable title for the tool	None
description	str | None	Optional description of what the tool does	None
annotations	ToolAnnotations | None	Optional ToolAnnotations providing additional tool information	None
icons	list[Icon] | None	Optional list of icons for the tool	None
meta	dict[str, Any] | None	Optional metadata dictionary for the tool	None
structured_output	bool | None	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) - If False, unconditionally creates an unstructured tool	None

Example

@server.tool()
def my_tool(x: int) -> str:
    return str(x)

@server.tool()
async def tool_with_context(x: int, ctx: Context) -> str:
    await ctx.info(f"Processing {x}")
    return str(x)

@server.tool()
async def async_tool(x: int, context: Context) -> str:
    await context.report_progress(50, 100)
    return str(x)

Source code in src/mcp/server/mcpserver/server.py

def tool(
    self,
    name: str | None = None,
    title: str | None = None,
    description: str | None = None,
    annotations: ToolAnnotations | None = None,
    icons: list[Icon] | None = None,
    meta: dict[str, Any] | None = None,
    structured_output: bool | None = None,
) -> Callable[[_CallableT], _CallableT]:
    """Decorator to register a tool.

    Tools can optionally request a Context object by adding a parameter with the
    Context type annotation. The context provides access to MCP capabilities like
    logging, progress reporting, and resource access.

    Args:
        name: Optional name for the tool (defaults to function name)
        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)
            - If False, unconditionally creates an unstructured tool

    Example:
        ```python
        @server.tool()
        def my_tool(x: int) -> str:
            return str(x)
        ```

        ```python
        @server.tool()
        async def tool_with_context(x: int, ctx: Context) -> str:
            await ctx.info(f"Processing {x}")
            return str(x)
        ```

        ```python
        @server.tool()
        async def async_tool(x: int, context: Context) -> str:
            await context.report_progress(50, 100)
            return str(x)
        ```
    """
    # Check if user passed function directly instead of calling decorator
    if callable(name):
        raise TypeError(
            "The @tool decorator was used incorrectly. Did you forget to call it? Use @tool() instead of @tool"
        )

    def decorator(fn: _CallableT) -> _CallableT:
        self.add_tool(
            fn,
            name=name,
            title=title,
            description=description,
            annotations=annotations,
            icons=icons,
            meta=meta,
            structured_output=structured_output,
        )
        return fn

    return decorator

completion

completion()

Decorator to register a completion handler.

The completion handler receives: - ref: PromptReference or ResourceTemplateReference - argument: CompletionArgument with name and partial value - context: Optional CompletionContext with previously resolved arguments

Example

@mcp.completion()
async def handle_completion(ref, argument, context):
    if isinstance(ref, ResourceTemplateReference):
        # Return completions based on ref, argument, and context
        return Completion(values=["option1", "option2"])
    return None

Source code in src/mcp/server/mcpserver/server.py

def completion(self):
    """Decorator to register a completion handler.

    The completion handler receives:
    - ref: PromptReference or ResourceTemplateReference
    - argument: CompletionArgument with name and partial value
    - context: Optional CompletionContext with previously resolved arguments

    Example:
        ```python
        @mcp.completion()
        async def handle_completion(ref, argument, context):
            if isinstance(ref, ResourceTemplateReference):
                # Return completions based on ref, argument, and context
                return Completion(values=["option1", "option2"])
            return None
        ```
    """

    def decorator(func: _CallableT) -> _CallableT:
        async def handler(
            ctx: ServerRequestContext[LifespanResultT], params: CompleteRequestParams
        ) -> CompleteResult:
            result = await func(params.ref, params.argument, params.context)
            return CompleteResult(
                completion=result if result is not None else Completion(values=[], total=None, has_more=None),
            )

        # TODO(maxisbey): remove private access — completion needs post-construction
        #   handler registration, find a better pattern for this
        self._lowlevel_server._add_request_handler(  # pyright: ignore[reportPrivateUsage]
            "completion/complete", handler
        )
        return func

    return decorator

add_resource

add_resource(resource: Resource) -> None

Add a resource to the server.

Parameters:

Name	Type	Description	Default
resource	Resource	A Resource instance to add	required

Source code in src/mcp/server/mcpserver/server.py

def add_resource(self, resource: Resource) -> None:
    """Add a resource to the server.

    Args:
        resource: A Resource instance to add
    """
    self._resource_manager.add_resource(resource)

resource

resource(
    uri: str,
    *,
    name: str | None = None,
    title: str | None = None,
    description: str | None = None,
    mime_type: str | None = None,
    icons: list[Icon] | None = None,
    annotations: Annotations | None = None,
    meta: dict[str, Any] | None = None
) -> Callable[[_CallableT], _CallableT]

Decorator to register a function as a resource.

The function will be called when the resource is read to generate its content. The function can return: - str for text content - bytes for binary content - other types will be converted to JSON

If the URI contains parameters (e.g. "resource://{param}") or the function has parameters, it will be registered as a template resource.

Parameters:

Name	Type	Description	Default
uri	str	URI for the resource (e.g. "resource://my-resource" or "resource://{param}")	required
name	str | None	Optional name for the resource	None
title	str | None	Optional human-readable title for the resource	None
description	str | None	Optional description of the resource	None
mime_type	str | None	Optional MIME type for the resource	None
icons	list[Icon] | None	Optional list of icons for the resource	None
annotations	Annotations | None	Optional annotations for the resource	None
meta	dict[str, Any] | None	Optional metadata dictionary for the resource	None

Example

@server.resource("resource://my-resource")
def get_data() -> str:
    return "Hello, world!"

@server.resource("resource://my-resource")
async def get_data() -> str:
    data = await fetch_data()
    return f"Hello, world! {data}"

@server.resource("resource://{city}/weather")
def get_weather(city: str) -> str:
    return f"Weather for {city}"

@server.resource("resource://{city}/weather")
async def get_weather(city: str) -> str:
    data = await fetch_weather(city)
    return f"Weather for {city}: {data}"

Source code in src/mcp/server/mcpserver/server.py

def resource(
    self,
    uri: str,
    *,
    name: str | None = None,
    title: str | None = None,
    description: str | None = None,
    mime_type: str | None = None,
    icons: list[Icon] | None = None,
    annotations: Annotations | None = None,
    meta: dict[str, Any] | None = None,
) -> Callable[[_CallableT], _CallableT]:
    """Decorator to register a function as a resource.

    The function will be called when the resource is read to generate its content.
    The function can return:
    - str for text content
    - bytes for binary content
    - other types will be converted to JSON

    If the URI contains parameters (e.g. "resource://{param}") or the function
    has parameters, it will be registered as a template resource.

    Args:
        uri: URI for the resource (e.g. "resource://my-resource" or "resource://{param}")
        name: Optional name for the resource
        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:
        ```python
        @server.resource("resource://my-resource")
        def get_data() -> str:
            return "Hello, world!"

        @server.resource("resource://my-resource")
        async def get_data() -> str:
            data = await fetch_data()
            return f"Hello, world! {data}"

        @server.resource("resource://{city}/weather")
        def get_weather(city: str) -> str:
            return f"Weather for {city}"

        @server.resource("resource://{city}/weather")
        async def get_weather(city: str) -> str:
            data = await fetch_weather(city)
            return f"Weather for {city}: {data}"
        ```
    """
    # Check if user passed function directly instead of calling decorator
    if callable(uri):
        raise TypeError(
            "The @resource decorator was used incorrectly. "
            "Did you forget to call it? Use @resource('uri') instead of @resource"
        )

    def decorator(fn: _CallableT) -> _CallableT:
        # Check if this should be a template
        sig = inspect.signature(fn)
        has_uri_params = "{" in uri and "}" in uri
        has_func_params = bool(sig.parameters)

        if has_uri_params or has_func_params:
            # Check for Context parameter to exclude from validation
            context_param = find_context_parameter(fn)

            # Validate that URI params match function params (excluding context)
            uri_params = set(re.findall(r"{(\w+)}", uri))
            # We need to remove the context_param from the resource function if
            # there is any.
            func_params = {p for p in sig.parameters.keys() if p != context_param}

            if uri_params != func_params:
                raise ValueError(
                    f"Mismatch between URI parameters {uri_params} and function parameters {func_params}"
                )

            # Register as template
            self._resource_manager.add_template(
                fn=fn,
                uri_template=uri,
                name=name,
                title=title,
                description=description,
                mime_type=mime_type,
                icons=icons,
                annotations=annotations,
                meta=meta,
            )
        else:
            # Register as regular resource
            resource = FunctionResource.from_function(
                fn=fn,
                uri=uri,
                name=name,
                title=title,
                description=description,
                mime_type=mime_type,
                icons=icons,
                annotations=annotations,
                meta=meta,
            )
            self.add_resource(resource)
        return fn

    return decorator

add_prompt

add_prompt(prompt: Prompt) -> None

Add a prompt to the server.

Parameters:

Name	Type	Description	Default
prompt	Prompt	A Prompt instance to add	required

Source code in src/mcp/server/mcpserver/server.py

def add_prompt(self, prompt: Prompt) -> None:
    """Add a prompt to the server.

    Args:
        prompt: A Prompt instance to add
    """
    self._prompt_manager.add_prompt(prompt)

prompt

prompt(
    name: str | None = None,
    title: str | None = None,
    description: str | None = None,
    icons: list[Icon] | None = None,
) -> Callable[[_CallableT], _CallableT]

Decorator to register a prompt.

    Args:
        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:
        ```python
        @server.prompt()
        def analyze_table(table_name: str) -> list[Message]:
            schema = read_table_schema(table_name)
            return [
                {
                    "role": "user",
                    "content": f"Analyze this schema:

{schema}" } ]

        @server.prompt()
        async def analyze_file(path: str) -> list[Message]:
            content = await read_file(path)
            return [
                {
                    "role": "user",
                    "content": {
                        "type": "resource",
                        "resource": {
                            "uri": f"file://{path}",
                            "text": content
                        }
                    }
                }
            ]
        ```

Source code in src/mcp/server/mcpserver/server.py

def prompt(
    self,
    name: str | None = None,
    title: str | None = None,
    description: str | None = None,
    icons: list[Icon] | None = None,
) -> Callable[[_CallableT], _CallableT]:
    """Decorator to register a prompt.

    Args:
        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:
        ```python
        @server.prompt()
        def analyze_table(table_name: str) -> list[Message]:
            schema = read_table_schema(table_name)
            return [
                {
                    "role": "user",
                    "content": f"Analyze this schema:\n{schema}"
                }
            ]

        @server.prompt()
        async def analyze_file(path: str) -> list[Message]:
            content = await read_file(path)
            return [
                {
                    "role": "user",
                    "content": {
                        "type": "resource",
                        "resource": {
                            "uri": f"file://{path}",
                            "text": content
                        }
                    }
                }
            ]
        ```
    """
    # Check if user passed function directly instead of calling decorator
    if callable(name):
        raise TypeError(
            "The @prompt decorator was used incorrectly. "
            "Did you forget to call it? Use @prompt() instead of @prompt"
        )

    def decorator(func: _CallableT) -> _CallableT:
        prompt = Prompt.from_function(func, name=name, title=title, description=description, icons=icons)
        self.add_prompt(prompt)
        return func

    return decorator

custom_route

custom_route(
    path: str,
    methods: list[str],
    name: str | None = None,
    include_in_schema: bool = True,
)

Decorator to register a custom HTTP route on the MCP server.

Allows adding arbitrary HTTP endpoints outside the standard MCP protocol, which can be useful for OAuth callbacks, health checks, or admin APIs. The handler function must be an async function that accepts a Starlette Request and returns a Response.

Routes using this decorator will not require authorization. It is intended for uses that are either a part of authorization flows or intended to be public such as health check endpoints.

Parameters:

Name	Type	Description	Default
path	str	URL path for the route (e.g., "/oauth/callback")	required
methods	list[str]	List of HTTP methods to support (e.g., ["GET", "POST"])	required
name	str | None	Optional name for the route (to reference this route with Starlette's reverse URL lookup feature)	None
include_in_schema	bool	Whether to include in OpenAPI schema, defaults to True	True

Example

@server.custom_route("/health", methods=["GET"])
async def health_check(request: Request) -> Response:
    return JSONResponse({"status": "ok"})

Source code in src/mcp/server/mcpserver/server.py

def custom_route(
    self,
    path: str,
    methods: list[str],
    name: str | None = None,
    include_in_schema: bool = True,
):
    """Decorator to register a custom HTTP route on the MCP server.

    Allows adding arbitrary HTTP endpoints outside the standard MCP protocol,
    which can be useful for OAuth callbacks, health checks, or admin APIs.
    The handler function must be an async function that accepts a Starlette
    Request and returns a Response.

    Routes using this decorator will not require authorization. It is intended
    for uses that are either a part of authorization flows or intended to be
    public such as health check endpoints.

    Args:
        path: URL path for the route (e.g., "/oauth/callback")
        methods: List of HTTP methods to support (e.g., ["GET", "POST"])
        name: Optional name for the route (to reference this route with
              Starlette's reverse URL lookup feature)
        include_in_schema: Whether to include in OpenAPI schema, defaults to True

    Example:
        ```python
        @server.custom_route("/health", methods=["GET"])
        async def health_check(request: Request) -> Response:
            return JSONResponse({"status": "ok"})
        ```
    """

    def decorator(  # pragma: no cover
        func: Callable[[Request], Awaitable[Response]],
    ) -> Callable[[Request], Awaitable[Response]]:
        self._custom_starlette_routes.append(
            Route(path, endpoint=func, methods=methods, name=name, include_in_schema=include_in_schema)
        )
        return func

    return decorator  # pragma: no cover

run_stdio_async async

run_stdio_async() -> None

Run the server using stdio transport.

Source code in src/mcp/server/mcpserver/server.py

async def run_stdio_async(self) -> None:
    """Run the server using stdio transport."""
    async with stdio_server() as (read_stream, write_stream):
        await self._lowlevel_server.run(
            read_stream,
            write_stream,
            self._lowlevel_server.create_initialization_options(),
        )

run_sse_async async

run_sse_async(
    *,
    host: str = "127.0.0.1",
    port: int = 8000,
    sse_path: str = "/sse",
    message_path: str = "/messages/",
    transport_security: (
        TransportSecuritySettings | None
    ) = None
) -> None

Run the server using SSE transport.

Source code in src/mcp/server/mcpserver/server.py

async def run_sse_async(  # pragma: no cover
    self,
    *,
    host: str = "127.0.0.1",
    port: int = 8000,
    sse_path: str = "/sse",
    message_path: str = "/messages/",
    transport_security: TransportSecuritySettings | None = None,
) -> None:
    """Run the server using SSE transport."""
    import uvicorn

    starlette_app = self.sse_app(
        sse_path=sse_path,
        message_path=message_path,
        transport_security=transport_security,
        host=host,
    )

    config = uvicorn.Config(
        starlette_app,
        host=host,
        port=port,
        log_level=self.settings.log_level.lower(),
    )
    server = uvicorn.Server(config)
    await server.serve()

run_streamable_http_async async

run_streamable_http_async(
    *,
    host: str = "127.0.0.1",
    port: int = 8000,
    streamable_http_path: str = "/mcp",
    json_response: bool = False,
    stateless_http: bool = False,
    event_store: EventStore | None = None,
    retry_interval: int | None = None,
    transport_security: (
        TransportSecuritySettings | None
    ) = None
) -> None

Run the server using StreamableHTTP transport.

Source code in src/mcp/server/mcpserver/server.py

async def run_streamable_http_async(  # pragma: no cover
    self,
    *,
    host: str = "127.0.0.1",
    port: int = 8000,
    streamable_http_path: str = "/mcp",
    json_response: bool = False,
    stateless_http: bool = False,
    event_store: EventStore | None = None,
    retry_interval: int | None = None,
    transport_security: TransportSecuritySettings | None = None,
) -> None:
    """Run the server using StreamableHTTP transport."""
    import uvicorn

    starlette_app = self.streamable_http_app(
        streamable_http_path=streamable_http_path,
        json_response=json_response,
        stateless_http=stateless_http,
        event_store=event_store,
        retry_interval=retry_interval,
        transport_security=transport_security,
        host=host,
    )

    config = uvicorn.Config(
        starlette_app,
        host=host,
        port=port,
        log_level=self.settings.log_level.lower(),
    )
    server = uvicorn.Server(config)
    await server.serve()

sse_app

sse_app(
    *,
    sse_path: str = "/sse",
    message_path: str = "/messages/",
    transport_security: (
        TransportSecuritySettings | None
    ) = None,
    host: str = "127.0.0.1"
) -> Starlette

Return an instance of the SSE server app.

Source code in src/mcp/server/mcpserver/server.py

def sse_app(
    self,
    *,
    sse_path: str = "/sse",
    message_path: str = "/messages/",
    transport_security: TransportSecuritySettings | None = None,
    host: str = "127.0.0.1",
) -> Starlette:
    """Return an instance of the SSE server app."""
    # Auto-enable DNS rebinding protection for localhost (IPv4 and IPv6)
    if transport_security is None and host in ("127.0.0.1", "localhost", "::1"):
        transport_security = TransportSecuritySettings(
            enable_dns_rebinding_protection=True,
            allowed_hosts=["127.0.0.1:*", "localhost:*", "[::1]:*"],
            allowed_origins=["http://127.0.0.1:*", "http://localhost:*", "http://[::1]:*"],
        )

    sse = SseServerTransport(message_path, security_settings=transport_security)

    async def handle_sse(scope: Scope, receive: Receive, send: Send):  # pragma: no cover
        # Add client ID from auth context into request context if available

        async with sse.connect_sse(scope, receive, send) as streams:
            await self._lowlevel_server.run(
                streams[0], streams[1], self._lowlevel_server.create_initialization_options()
            )
        return Response()

    # Create routes
    routes: list[Route | Mount] = []
    middleware: list[Middleware] = []
    required_scopes: list[str] = []

    # Set up auth if configured
    if self.settings.auth:  # pragma: no cover
        required_scopes = self.settings.auth.required_scopes or []

        # Add auth middleware if token verifier is available
        if self._token_verifier:
            middleware = [
                # extract auth info from request (but do not require it)
                Middleware(
                    AuthenticationMiddleware,
                    backend=BearerAuthBackend(self._token_verifier),
                ),
                # Add the auth context middleware to store
                # authenticated user in a contextvar
                Middleware(AuthContextMiddleware),
            ]

        # Add auth endpoints if auth server provider is configured
        if self._auth_server_provider:
            from mcp.server.auth.routes import create_auth_routes

            routes.extend(
                create_auth_routes(
                    provider=self._auth_server_provider,
                    issuer_url=self.settings.auth.issuer_url,
                    service_documentation_url=self.settings.auth.service_documentation_url,
                    client_registration_options=self.settings.auth.client_registration_options,
                    revocation_options=self.settings.auth.revocation_options,
                )
            )

    # When auth is configured, require authentication
    if self._token_verifier:  # pragma: no cover
        # Determine resource metadata URL
        resource_metadata_url = None
        if self.settings.auth and self.settings.auth.resource_server_url:
            from mcp.server.auth.routes import build_resource_metadata_url

            # Build compliant metadata URL for WWW-Authenticate header
            resource_metadata_url = build_resource_metadata_url(self.settings.auth.resource_server_url)

        # Auth is enabled, wrap the endpoints with RequireAuthMiddleware
        routes.append(
            Route(
                sse_path,
                endpoint=RequireAuthMiddleware(handle_sse, required_scopes, resource_metadata_url),
                methods=["GET"],
            )
        )
        routes.append(
            Mount(
                message_path,
                app=RequireAuthMiddleware(sse.handle_post_message, required_scopes, resource_metadata_url),
            )
        )
    else:
        # Auth is disabled, no need for RequireAuthMiddleware
        # Since handle_sse is an ASGI app, we need to create a compatible endpoint
        async def sse_endpoint(request: Request) -> Response:  # pragma: no cover
            # Convert the Starlette request to ASGI parameters
            return await handle_sse(request.scope, request.receive, request._send)  # type: ignore[reportPrivateUsage]

        routes.append(
            Route(
                sse_path,
                endpoint=sse_endpoint,
                methods=["GET"],
            )
        )
        routes.append(
            Mount(
                message_path,
                app=sse.handle_post_message,
            )
        )
    # Add protected resource metadata endpoint if configured as RS
    if self.settings.auth and self.settings.auth.resource_server_url:  # pragma: no cover
        from mcp.server.auth.routes import create_protected_resource_routes

        routes.extend(
            create_protected_resource_routes(
                resource_url=self.settings.auth.resource_server_url,
                authorization_servers=[self.settings.auth.issuer_url],
                scopes_supported=self.settings.auth.required_scopes,
            )
        )

    # mount these routes last, so they have the lowest route matching precedence
    routes.extend(self._custom_starlette_routes)

    # Create Starlette app with routes and middleware
    return Starlette(debug=self.settings.debug, routes=routes, middleware=middleware)

streamable_http_app

streamable_http_app(
    *,
    streamable_http_path: str = "/mcp",
    json_response: bool = False,
    stateless_http: bool = False,
    event_store: EventStore | None = None,
    retry_interval: int | None = None,
    transport_security: (
        TransportSecuritySettings | None
    ) = None,
    host: str = "127.0.0.1"
) -> Starlette

Return an instance of the StreamableHTTP server app.

Source code in src/mcp/server/mcpserver/server.py

def streamable_http_app(
    self,
    *,
    streamable_http_path: str = "/mcp",
    json_response: bool = False,
    stateless_http: bool = False,
    event_store: EventStore | None = None,
    retry_interval: int | None = None,
    transport_security: TransportSecuritySettings | None = None,
    host: str = "127.0.0.1",
) -> Starlette:
    """Return an instance of the StreamableHTTP server app."""
    return self._lowlevel_server.streamable_http_app(
        streamable_http_path=streamable_http_path,
        json_response=json_response,
        stateless_http=stateless_http,
        event_store=event_store,
        retry_interval=retry_interval,
        transport_security=transport_security,
        host=host,
        auth=self.settings.auth,
        token_verifier=self._token_verifier,
        auth_server_provider=self._auth_server_provider,
        custom_starlette_routes=self._custom_starlette_routes,
        debug=self.settings.debug,
    )

list_prompts async

list_prompts() -> list[Prompt]

List all available prompts.

Source code in src/mcp/server/mcpserver/server.py

async def list_prompts(self) -> list[MCPPrompt]:
    """List all available prompts."""
    prompts = self._prompt_manager.list_prompts()
    return [
        MCPPrompt(
            name=prompt.name,
            title=prompt.title,
            description=prompt.description,
            arguments=[
                MCPPromptArgument(
                    name=arg.name,
                    description=arg.description,
                    required=arg.required,
                )
                for arg in (prompt.arguments or [])
            ],
            icons=prompt.icons,
        )
        for prompt in prompts
    ]

get_prompt async

get_prompt(
    name: str,
    arguments: dict[str, Any] | None = None,
    context: Context[LifespanResultT, Any] | None = None,
) -> GetPromptResult

Get a prompt by name with arguments.

Source code in src/mcp/server/mcpserver/server.py

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)

        return GetPromptResult(
            description=prompt.description,
            messages=pydantic_core.to_jsonable_python(messages),
        )
    except Exception as e:
        logger.exception(f"Error getting prompt {name}")
        raise ValueError(str(e)) from e