content: """FastMCP - A more ergonomic interface for MCP servers."""
  
  from __future__ import annotations as _annotations
  
  import inspect
  import re
  from collections.abc import (
      AsyncIterator,
      Awaitable,
      Callable,
      Collection,
      Iterable,
      Sequence,
  )
  from contextlib import AbstractAsyncContextManager, asynccontextmanager
  from typing import Any, Generic, Literal
  
  import anyio
  import pydantic_core
  from pydantic import BaseModel
  from pydantic.networks import AnyUrl
  from pydantic_settings import BaseSettings, SettingsConfigDict
  from starlette.applications import Starlette
  from starlette.middleware import Middleware
  from starlette.middleware.authentication import AuthenticationMiddleware
  from starlette.requests import Request
  from starlette.responses import Response
  from starlette.routing import Mount, Route
  from starlette.types import Receive, Scope, Send
  
  from mcp.server.auth.middleware.auth_context import AuthContextMiddleware
  from mcp.server.auth.middleware.bearer_auth import (
      BearerAuthBackend,
      RequireAuthMiddleware,
  )
  from mcp.server.auth.provider import (
      OAuthAuthorizationServerProvider,
      ProviderTokenVerifier,
      TokenVerifier,
  )
  from mcp.server.auth.settings import AuthSettings
  from mcp.server.elicitation import (
      ElicitationResult,
      ElicitSchemaModelT,
      UrlElicitationResult,
      elicit_with_validation,
  )
  from mcp.server.elicitation import (
      elicit_url as _elicit_url,
  )
  from mcp.server.fastmcp.exceptions import ResourceError
  from mcp.server.fastmcp.prompts import Prompt, PromptManager
  from mcp.server.fastmcp.resources import FunctionResource, Resource, ResourceManager
  from mcp.server.fastmcp.tools import Tool, ToolManager
  from mcp.server.fastmcp.utilities.context_injection import find_context_parameter
  from mcp.server.fastmcp.utilities.logging import configure_logging, get_logger
  from mcp.server.lowlevel.helper_types import ReadResourceContents
  from mcp.server.lowlevel.server import LifespanResultT
  from mcp.server.lowlevel.server import Server as MCPServer
  from mcp.server.lowlevel.server import lifespan as default_lifespan
  from mcp.server.session import ServerSession, ServerSessionT
  from mcp.server.sse import SseServerTransport
  from mcp.server.stdio import stdio_server
  from mcp.server.streamable_http import EventStore
  from mcp.server.streamable_http_manager import StreamableHTTPSessionManager
  from mcp.server.transport_security import TransportSecuritySettings
  from mcp.shared.context import LifespanContextT, RequestContext, RequestT
  from mcp.types import Annotations, AnyFunction, ContentBlock, GetPromptResult, Icon, ToolAnnotations
  from mcp.types import Prompt as MCPPrompt
  from mcp.types import PromptArgument as MCPPromptArgument
  from mcp.types import Resource as MCPResource
  from mcp.types import ResourceTemplate as MCPResourceTemplate
  from mcp.types import Tool as MCPTool
  
  logger = get_logger(__name__)
  
  
  class Settings(BaseSettings, Generic[LifespanResultT]):
      """FastMCP server settings.
  
      All settings can be configured via environment variables with the prefix FASTMCP_.
      For example, FASTMCP_DEBUG=true will set debug=True.
      """
  
      model_config = SettingsConfigDict(
          env_prefix="FASTMCP_",
          env_file=".env",
          env_nested_delimiter="__",
          nested_model_default_partial_update=True,
          extra="ignore",
      )
  
      # Server settings
      debug: bool
      log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
  
      # HTTP settings
      host: str
      port: int
      mount_path: str
      sse_path: str
      message_path: str
      streamable_http_path: str
  
      # StreamableHTTP settings
      json_response: bool
      stateless_http: bool
      """Define if the server should create a new transport per request."""
  
      # resource settings
      warn_on_duplicate_resources: bool
  
      # tool settings
      warn_on_duplicate_tools: bool
  
      # prompt settings
      warn_on_duplicate_prompts: bool
  
      # TODO(Marcelo): Investigate if this is used. If it is, it's probably a good idea to remove it.
      dependencies: list[str]
      """A list of dependencies to install in the server environment."""
  
      lifespan: Callable[[FastMCP[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]] | None
      """A async context manager that will be called when the server is started."""
  
      auth: AuthSettings | None
  
      # Transport security settings (DNS rebinding protection)
      transport_security: TransportSecuritySettings | None
  
  
  def lifespan_wrapper(
      app: FastMCP[LifespanResultT],
      lifespan: Callable[[FastMCP[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]],
  ) -> Callable[[MCPServer[LifespanResultT, Request]], AbstractAsyncContextManager[LifespanResultT]]:
      @asynccontextmanager
      async def wrap(
          _: MCPServer[LifespanResultT, Request],
      ) -> AsyncIterator[LifespanResultT]:
          async with lifespan(app) as context:
              yield context
  
      return wrap
  
  
  class FastMCP(Generic[LifespanResultT]):
      def __init__(  # noqa: PLR0913
          self,
          name: str | None = None,
          instructions: str | None = None,
          website_url: str | None = None,
          icons: list[Icon] | None = None,
          auth_server_provider: (OAuthAuthorizationServerProvider[Any, Any, Any] | None) = None,
          token_verifier: TokenVerifier | None = None,
          event_store: EventStore | None = None,
          retry_interval: int | None = None,
          *,
          tools: list[Tool] | None = None,
          debug: bool = False,
          log_level: Literal["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"] = "INFO",
          host: str = "127.0.0.1",
          port: int = 8000,
          mount_path: str = "/",
          sse_path: str = "/sse",
          message_path: str = "/messages/",
          streamable_http_path: str = "/mcp",
          json_response: bool = False,
          stateless_http: bool = False,
          warn_on_duplicate_resources: bool = True,
          warn_on_duplicate_tools: bool = True,
          warn_on_duplicate_prompts: bool = True,
          dependencies: Collection[str] = (),
          lifespan: (Callable[[FastMCP[LifespanResultT]], AbstractAsyncContextManager[LifespanResultT]] | None) = None,
          auth: AuthSettings | None = None,
          transport_security: TransportSecuritySettings | None = None,
      ):
          # 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]:*"],
              )
  
          self.settings = Settings(
              debug=debug,
              log_level=log_level,
              host=host,
              port=port,
              mount_path=mount_path,
              sse_path=sse_path,
              message_path=message_path,
              streamable_http_path=streamable_http_path,
              json_response=json_response,
              stateless_http=stateless_http,
              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=list(dependencies),
              lifespan=lifespan,
              auth=auth,
              transport_security=transport_security,
          )
  
          self._mcp_server = MCPServer(
              name=name or "FastMCP",
              instructions=instructions,
              website_url=website_url,
              icons=icons,
              # TODO(Marcelo): It seems there's a type mismatch between the lifespan type from an FastMCP 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
          )
          self._tool_manager = ToolManager(tools=tools, warn_on_duplicate_tools=self.settings.warn_on_duplicate_tools)
          self._resource_manager = ResourceManager(warn_on_duplicate_resources=self.settings.warn_on_duplicate_resources)
          self._prompt_manager = PromptManager(warn_on_duplicate_prompts=self.settings.warn_on_duplicate_prompts)
          # 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._event_store = event_store
          self._retry_interval = retry_interval
          self._custom_starlette_routes: list[Route] = []
          self.dependencies = self.settings.dependencies
          self._session_manager: StreamableHTTPSessionManager | None = None
  
          # Set up MCP protocol handlers
          self._setup_handlers()
  
          # Configure logging
          configure_logging(self.settings.log_level)
  
      @property
      def name(self) -> str:
          return self._mcp_server.name
  
      @property
      def instructions(self) -> str | None:
          return self._mcp_server.instructions
  
      @property
      def website_url(self) -> str | None:
          return self._mcp_server.website_url
  
      @property
      def icons(self) -> list[Icon] | None:
          return self._mcp_server.icons
  
      @property
      def session_manager(self) -> StreamableHTTPSessionManager:
          """Get the StreamableHTTP session manager.
  
          This is exposed to enable advanced use cases like mounting multiple
          FastMCP servers in a single FastAPI application.
  
          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
  
      def run(
          self,
          transport: Literal["stdio", "sse", "streamable-http"] = "stdio",
          mount_path: str | None = None,
      ) -> None:
          """Run the FastMCP server. Note this is a synchronous function.
  
          Args:
              transport: Transport protocol to use ("stdio", "sse", or "streamable-http")
              mount_path: Optional mount path for SSE transport
          """
          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(mount_path))
              case "streamable-http":  # pragma: no cover
                  anyio.run(self.run_streamable_http_async)
  
      def _setup_handlers(self) -> None:
          """Set up core MCP protocol handlers."""
          self._mcp_server.list_tools()(self.list_tools)
          # Note: we disable the lowlevel server's input validation.
          # FastMCP does ad hoc conversion of incoming data before validating -
          # for now we preserve this for backwards compatibility.
          self._mcp_server.call_tool(validate_input=False)(self.call_tool)
          self._mcp_server.list_resources()(self.list_resources)
          self._mcp_server.read_resource()(self.read_resource)
          self._mcp_server.list_prompts()(self.list_prompts)
          self._mcp_server.get_prompt()(self.get_prompt)
          self._mcp_server.list_resource_templates()(self.list_resource_templates)
  
      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,
                  inputSchema=info.parameters,
                  outputSchema=info.output_schema,
                  annotations=info.annotations,
                  icons=info.icons,
                  _meta=info.meta,
              )
              for info in tools
          ]
  
      def get_context(self) -> Context[ServerSession, LifespanResultT, Request]:
          """
          Returns a Context object. Note that the context will only be valid
          during a request; outside a request, most methods will error.
          """
          try:
              request_context = self._mcp_server.request_context
          except LookupError:
              request_context = None
          return Context(request_context=request_context, fastmcp=self)
  
      async def call_tool(self, name: str, arguments: dict[str, Any]) -> Sequence[ContentBlock] | dict[str, Any]:
          """Call a tool by name with arguments."""
          context = self.get_context()
          return await self._tool_manager.call_tool(name, arguments, context=context, convert_result=True)
  
      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,
                  mimeType=resource.mime_type,
                  icons=resource.icons,
                  annotations=resource.annotations,
              )
              for resource in resources
          ]
  
      async def list_resource_templates(self) -> list[MCPResourceTemplate]:
          templates = self._resource_manager.list_templates()
          return [
              MCPResourceTemplate(
                  uriTemplate=template.uri_template,
                  name=template.name,
                  title=template.title,
                  description=template.description,
                  mimeType=template.mime_type,
                  icons=template.icons,
                  annotations=template.annotations,
              )
              for template in templates
          ]
  
      async def read_resource(self, uri: AnyUrl | str) -> Iterable[ReadResourceContents]:
          """Read a resource by URI."""
  
          context = self.get_context()
          resource = await self._resource_manager.get_resource(uri, context=context)
          if not resource:  # pragma: no cover
              raise ResourceError(f"Unknown resource: {uri}")
  
          try:
              content = await resource.read()
              return [ReadResourceContents(content=content, mime_type=resource.mime_type)]
          except Exception as e:  # pragma: no cover
              logger.exception(f"Error reading resource {uri}")
              raise ResourceError(str(e))
  
      def add_tool(
          self,
          fn: AnyFunction,
          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
              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[[AnyFunction], AnyFunction]:
          """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
              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:
              @server.tool()
              def my_tool(x: int) -> str:
                  return str(x)
  
              @server.tool()
              def tool_with_context(x: int, ctx: Context) -> str:
                  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)
          """
          # 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: AnyFunction) -> AnyFunction:
              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:
              @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
          """
          return self._mcp_server.completion()
  
      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,
      ) -> Callable[[AnyFunction], AnyFunction]:
          """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
  
          Example:
              @server.resource("resource://my-resource")
              def get_data() -> str:
                  return "Hello, world!"
  
              @server.resource("resource://my-resource")
              async 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: AnyFunction) -> AnyFunction:
              # 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,
                  )
              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,
                  )
                  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[[AnyFunction], AnyFunction]:
          """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
  
          Example:
              @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: AnyFunction) -> AnyFunction:
              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 FastMCP 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:
              @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._mcp_server.run(
                  read_stream,
                  write_stream,
                  self._mcp_server.create_initialization_options(),
              )
  
      async def run_sse_async(self, mount_path: str | None = None) -> None:  # pragma: no cover
          """Run the server using SSE transport."""
          import uvicorn
  
          starlette_app = self.sse_app(mount_path)
  
          config = uvicorn.Config(
              starlette_app,
              host=self.settings.host,
              port=self.settings.port,
              log_level=self.settings.log_level.lower(),
          )
          server = uvicorn.Server(config)
          await server.serve()
  
      async def run_streamable_http_async(self) -> None:  # pragma: no cover
          """Run the server using StreamableHTTP transport."""
          import uvicorn
  
          starlette_app = self.streamable_http_app()
  
          config = uvicorn.Config(
              starlette_app,
              host=self.settings.host,
              port=self.settings.port,
              log_level=self.settings.log_level.lower(),
          )
          server = uvicorn.Server(config)
          await server.serve()
  
      def _normalize_path(self, mount_path: str, endpoint: str) -> str:
          """
          Combine mount path and endpoint to return a normalized path.
  
          Args:
              mount_path: The mount path (e.g. "/github" or "/")
              endpoint: The endpoint path (e.g. "/messages/")
  
          Returns:
              Normalized path (e.g. "/github/messages/")
          """
          # Special case: root path
          if mount_path == "/":
              return endpoint
  
          # Remove trailing slash from mount path
          if mount_path.endswith("/"):
              mount_path = mount_path[:-1]
  
          # Ensure endpoint starts with slash
          if not endpoint.startswith("/"):
              endpoint = "/" + endpoint
  
          # Combine paths
          return mount_path + endpoint
  
      def sse_app(self, mount_path: str | None = None) -> Starlette:
          """Return an instance of the SSE server app."""
          from starlette.middleware import Middleware
          from starlette.routing import Mount, Route
  
          # Update mount_path in settings if provided
          if mount_path is not None:
              self.settings.mount_path = mount_path
  
          # Create normalized endpoint considering the mount path
          normalized_message_endpoint = self._normalize_path(self.settings.mount_path, self.settings.message_path)
  
          # Set up auth context and dependencies
  
          sse = SseServerTransport(
              normalized_message_endpoint,
              security_settings=self.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._mcp_server.run(
                      streams[0],
                      streams[1],
                      self._mcp_server.create_initialization_options(),
                  )
              return Response()
  
          # Create routes
          routes: list[Route | Mount] = []
          middleware: list[Middleware] = []
          required_scopes = []
  
          # Set up auth if configured
          if self.settings.auth:  # pragma: no cover
              required_scopes = self.settings.auth.required_scopes or []
  
  