INNER CODE UNIT · Python

_run_streamable_http_async

RekklesNA/ProxmoxMCP-Plus · src/proxmox_mcp/server.py:341

    async def _run_streamable_http_async(self, *, sse: bool = False) -> None:
        """Run Streamable HTTP with SDK OAuth or legacy Bearer authentication."""
        import uvicorn

        if self.oauth_provider is not None and sse:
            raise ValueError("MCP OAuth mode supports STREAMABLE HTTP only")

        app: Any = self.mcp.sse_app() if sse else self.mcp.streamable_http_app()
        if self.oauth_provider is not None:
            app = self.oauth_provider.bind_http_lifespan(app)
        api_key = os.getenv("MCP_API_KEY")
        if self.oauth_provider is not None:
            self.logger.info(
                "MCP HTTP OAuth is enabled for issuer %s",
                self.oauth_provider.issuer_url,
            )
        elif api_key:
            app = MCPBearerAuthMiddleware(app, api_key=api_key)

View source record →