server

MCP Server Module

This module provides a framework for creating an MCP (Model Context Protocol) server. It allows you to easily define and handle various types of requests and notifications using constructor-based handler registration.

Usage: 1. Define handler functions: async def my_list_tools(ctx, params): return types.ListToolsResult(tools=[...])

async def my_call_tool(ctx, params): return types.CallToolResult(content=[...])

2. Create a Server instance with on_* handlers: server = Server( "your_server_name", on_list_tools=my_list_tools, on_call_tool=my_call_tool, )

3. Run the server: async def main(): async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, server.create_initialization_options(), )

asyncio.run(main())

The Server class dispatches incoming requests and notifications to registered handler callables by method string.

RequestHandler module-attribute

RequestHandler = Callable[
    [ServerRequestContext[LifespanResultT], _ParamsT],
    Awaitable[HandlerResult],
]

A registered request handler: (ctx, params) -> result.

NotificationHandler module-attribute

NotificationHandler = Callable[
    [ServerRequestContext[LifespanResultT], _ParamsT],
    Awaitable[None],
]

A registered notification handler: (ctx, params) -> None.

HandlerEntry dataclass

Bases: Generic[LifespanResultT]

A registered handler and the params model to validate incoming params against.

Stored in Server._request_handlers / _notification_handlers and consumed by ServerRunner to validate, build Context, and invoke. The handler's second-argument type is erased to Any in storage (each entry has a different concrete params type and Callable parameters are contravariant); the precise type is recoverable via params_type. The correlation is enforced at registration time by Server.add_request_handler.

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

@dataclass(frozen=True, slots=True)
class HandlerEntry(Generic[LifespanResultT]):
    """A registered handler and the params model to validate incoming params against.

    Stored in `Server._request_handlers` / `_notification_handlers` and consumed
    by `ServerRunner` to validate, build `Context`, and invoke. The handler's
    second-argument type is erased to `Any` in storage (each entry has a
    different concrete params type and `Callable` parameters are contravariant);
    the precise type is recoverable via `params_type`. The correlation is
    enforced at registration time by `Server.add_request_handler`.
    """

    params_type: type[BaseModel]
    handler: RequestHandler[LifespanResultT, Any]

lifespan async

lifespan(_: Server[Any]) -> AsyncIterator[dict[str, Any]]

Default lifespan context manager that does nothing.

Returns:

Type	Description
AsyncIterator[dict[str, Any]]	An empty context object

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

@asynccontextmanager
async def lifespan(_: Server[Any]) -> AsyncIterator[dict[str, Any]]:
    """Default lifespan context manager that does nothing.

    Returns:
        An empty context object
    """
    yield {}

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],
        ]
        | 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, HandlerEntry[LifespanResultT]] = {}
        self._notification_handlers: dict[str, HandlerEntry[LifespanResultT]] = {}
        self._session_manager: StreamableHTTPSessionManager | None = None
        # Context-tier middleware: wraps every inbound request (including
        # `initialize`, lookup, validation, handler) with
        # `(ctx, method, params, call_next)`. Applied in `ServerRunner._on_request`.
        # TODO(maxisbey): provisional - signature and semantics change with the
        # Context/middleware rework (covariant `Context[L]`, outbound seam) before
        # v2 final.
        self.middleware: list[ServerMiddleware[LifespanResultT]] = []
        logger.debug("Initializing server %r", name)

        _spec_requests: list[tuple[str, type[BaseModel], RequestHandler[LifespanResultT, Any] | None]] = [
            ("ping", types.RequestParams, on_ping),
            ("prompts/list", types.PaginatedRequestParams, on_list_prompts),
            ("prompts/get", types.GetPromptRequestParams, on_get_prompt),
            ("resources/list", types.PaginatedRequestParams, on_list_resources),
            ("resources/templates/list", types.PaginatedRequestParams, on_list_resource_templates),
            ("resources/read", types.ReadResourceRequestParams, on_read_resource),
            ("resources/subscribe", types.SubscribeRequestParams, on_subscribe_resource),
            ("resources/unsubscribe", types.UnsubscribeRequestParams, on_unsubscribe_resource),
            ("tools/list", types.PaginatedRequestParams, on_list_tools),
            ("tools/call", types.CallToolRequestParams, on_call_tool),
            ("logging/setLevel", types.SetLevelRequestParams, on_set_logging_level),
            ("completion/complete", types.CompleteRequestParams, on_completion),
        ]
        self._request_handlers.update({m: HandlerEntry(pt, h) for m, pt, h in _spec_requests if h is not None})

        _spec_notifications: list[tuple[str, type[BaseModel], NotificationHandler[LifespanResultT, Any] | None]] = [
            ("notifications/roots/list_changed", types.NotificationParams, on_roots_list_changed),
            ("notifications/progress", types.ProgressNotificationParams, on_progress),
        ]
        self._notification_handlers.update(
            {m: HandlerEntry(pt, h) for m, pt, h in _spec_notifications if h is not None}
        )

    def add_request_handler(
        self,
        method: str,
        params_type: type[_ParamsT],
        handler: RequestHandler[LifespanResultT, _ParamsT],
    ) -> None:
        """Register a request handler for `method`.

        `params_type` is the model incoming params are validated against
        before the handler is invoked. It should subclass `RequestParams` so
        `_meta` parses uniformly. A message with no `params` member validates
        `{}` against `params_type`: models with required fields reject it as
        INVALID_PARAMS, all-optional models reach the handler with their
        defaults - the handler never receives `None`. Replaces any existing
        handler for the same method, except `initialize`, which is reserved:
        the runner owns the handshake, so registering it raises `ValueError`.
        Use `Server.middleware` to observe or wrap initialization.
        """
        if method == "initialize":
            raise ValueError(
                "'initialize' is handled by the server runner and cannot be overridden; "
                "use Server.middleware to observe or wrap initialization"
            )
        self._request_handlers[method] = HandlerEntry(params_type, handler)

    def add_notification_handler(
        self,
        method: str,
        params_type: type[_ParamsT],
        handler: NotificationHandler[LifespanResultT, _ParamsT],
    ) -> None:
        """Register a notification handler for `method`.

        `params_type` should subclass `NotificationParams` so `_meta`
        parses uniformly. Absent params follow the same contract as requests:
        `{}` is validated, so the handler receives the model with its defaults,
        never `None`. Replaces any existing handler. A handler for
        `notifications/initialized` runs after the runner has marked the
        connection initialized.
        """
        self._notification_handlers[method] = HandlerEntry(params_type, handler)

    def get_request_handler(self, method: str) -> HandlerEntry[LifespanResultT] | None:
        """Return the registered entry for a request method, or `None`."""
        return self._request_handlers.get(method)

    def get_notification_handler(self, method: str) -> HandlerEntry[LifespanResultT] | None:
        """Return the registered entry for a notification method, or `None`."""
        return self._notification_handlers.get(method)

    # 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,
        )
        return capabilities

    @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:
            raise RuntimeError(  # pragma: no cover
                "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

    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,
    ) -> None:
        async with self.lifespan(self) as lifespan_context:
            dispatcher: JSONRPCDispatcher[TransportContext] = JSONRPCDispatcher(
                read_stream,
                write_stream,
                raise_handler_exceptions=raise_exceptions,
                # Handle `initialize` inline so a client that pipelines it with
                # the next request (spec says SHOULD NOT, not MUST NOT) sees
                # the initialized state instead of failing the init-gate.
                inline_methods=frozenset({"initialize"}),
            )
            runner = ServerRunner(
                server=self,
                dispatcher=dispatcher,
                lifespan_state=lifespan_context,
                init_options=initialization_options,
                # Stateless HTTP has no standalone GET stream, so server-initiated
                # requests on `runner.connection` must fail fast with
                # `NoBackChannelError` rather than write to a channel that will
                # never deliver a response.
                has_standalone_channel=not stateless,
                stateless=stateless,
                dispatch_middleware=[otel_middleware],
            )
            await runner.run()

    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:
            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:
            # Determine resource metadata URL
            resource_metadata_url = None
            if auth and auth.resource_server_url:  # pragma: no branch
                # 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:
            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(),
        )

add_request_handler

add_request_handler(
    method: str,
    params_type: type[_ParamsT],
    handler: RequestHandler[LifespanResultT, _ParamsT],
) -> None

Register a request handler for method.

params_type is the model incoming params are validated against before the handler is invoked. It should subclass RequestParams so _meta parses uniformly. A message with no params member validates {} against params_type: models with required fields reject it as INVALID_PARAMS, all-optional models reach the handler with their defaults - the handler never receives None. Replaces any existing handler for the same method, except initialize, which is reserved: the runner owns the handshake, so registering it raises ValueError. Use Server.middleware to observe or wrap initialization.

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

def add_request_handler(
    self,
    method: str,
    params_type: type[_ParamsT],
    handler: RequestHandler[LifespanResultT, _ParamsT],
) -> None:
    """Register a request handler for `method`.

    `params_type` is the model incoming params are validated against
    before the handler is invoked. It should subclass `RequestParams` so
    `_meta` parses uniformly. A message with no `params` member validates
    `{}` against `params_type`: models with required fields reject it as
    INVALID_PARAMS, all-optional models reach the handler with their
    defaults - the handler never receives `None`. Replaces any existing
    handler for the same method, except `initialize`, which is reserved:
    the runner owns the handshake, so registering it raises `ValueError`.
    Use `Server.middleware` to observe or wrap initialization.
    """
    if method == "initialize":
        raise ValueError(
            "'initialize' is handled by the server runner and cannot be overridden; "
            "use Server.middleware to observe or wrap initialization"
        )
    self._request_handlers[method] = HandlerEntry(params_type, handler)

add_notification_handler

add_notification_handler(
    method: str,
    params_type: type[_ParamsT],
    handler: NotificationHandler[LifespanResultT, _ParamsT],
) -> None

Register a notification handler for method.

params_type should subclass NotificationParams so _meta parses uniformly. Absent params follow the same contract as requests: {} is validated, so the handler receives the model with its defaults, never None. Replaces any existing handler. A handler for notifications/initialized runs after the runner has marked the connection initialized.

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

def add_notification_handler(
    self,
    method: str,
    params_type: type[_ParamsT],
    handler: NotificationHandler[LifespanResultT, _ParamsT],
) -> None:
    """Register a notification handler for `method`.

    `params_type` should subclass `NotificationParams` so `_meta`
    parses uniformly. Absent params follow the same contract as requests:
    `{}` is validated, so the handler receives the model with its defaults,
    never `None`. Replaces any existing handler. A handler for
    `notifications/initialized` runs after the runner has marked the
    connection initialized.
    """
    self._notification_handlers[method] = HandlerEntry(params_type, handler)

get_request_handler

get_request_handler(
    method: str,
) -> HandlerEntry[LifespanResultT] | None

Return the registered entry for a request method, or None.

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

def get_request_handler(self, method: str) -> HandlerEntry[LifespanResultT] | None:
    """Return the registered entry for a request method, or `None`."""
    return self._request_handlers.get(method)

get_notification_handler

get_notification_handler(
    method: str,
) -> HandlerEntry[LifespanResultT] | None

Return the registered entry for a notification method, or None.

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

def get_notification_handler(self, method: str) -> HandlerEntry[LifespanResultT] | None:
    """Return the registered entry for a notification method, or `None`."""
    return self._notification_handlers.get(method)

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,
    )
    return capabilities

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:
        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:
        # Determine resource metadata URL
        resource_metadata_url = None
        if auth and auth.resource_server_url:  # pragma: no branch
            # 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:
        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(),
    )